"""Watcher ownership and completion/read races need GC and busy-host control that daemon E2E cannot reliably provide."""
from pathlib import Path
import sys
import textwrap
import unittest
sys.path.insert(0, str(Path(__file__).resolve().parent))
from job_wake_test import Owner # noqa: E402
class JobWatchersTest(unittest.TestCase):
def setUp(self):
self.owner = Owner(["run", "files"])
self.addCleanup(self.owner.close)
self.owner.wait_for(lambda frame: frame.get("type") == "ready")
self.cells = 0
self.cell(
"import __main__ as kernel, asyncio, gc, weakref\n"
"from albedo_plugins import run as runner, files as file_plugin"
)
def cell(self, code):
self.cells += 1
cell_id = f"c{self.cells}"
self.owner.send(
{"type": "execute", "id": cell_id, "code": textwrap.dedent(code)}
)
done = self.owner.wait_for(
lambda frame: frame.get("type") == "done" and frame.get("id") == cell_id
)
self.assertEqual(done["status"], "ok", done)
return done
def test_an_early_read_does_not_consume_the_completion_watcher(self):
self.cell("""
job = run('echo', 'result')
output.read(job.id)
await asyncio.shield(job.task)
assert not job._read and not job._awaited
assert output.read(job.id).strip() == 'result'
assert job._read
""")
def test_forgotten_jobs_and_internal_file_commands_release_watchers(self):
self.cell("""
job = run('true')
await job
reference = weakref.ref(job)
runner.forget(job)
del job
await file_plugin._run('printf internal')
await asyncio.sleep(0)
gc.collect()
assert reference() is None
assert not jobs and not kernel.READ_WATCHERS
assert not [c for c in kernel.ARCHIVES.values() if c.kind == 'job']
""")
def test_eviction_releases_jobs_but_preserves_held_handles(self):
self.cell("""
held = run('echo', 'held output')
await held
disposable = run('true')
await disposable
reference = weakref.ref(disposable)
capture = disposable.capture
del disposable
for index in range(runner.RETAINED_LIMIT):
# Keep this output past job eviction to catch strong watcher ownership.
kernel.remember(capture)
await run('true')
await asyncio.sleep(0)
gc.collect()
assert reference() is None and capture.id in kernel.ARCHIVES
assert capture.id not in kernel.READ_WATCHERS
assert held.id not in jobs and held.id not in kernel.ARCHIVES
assert held.id not in kernel.READ_WATCHERS
assert held.tail().strip() == 'held output' and await held is held
""")
def test_other_callables_and_replacement_watchers_remain_usable(self):
self.cell("""
capture = kernel.background_capture('callables')
reads = []
kernel.watch_output(capture.id, lambda: reads.append('function'))
gc.collect()
output.read(capture.id)
class Callback:
def __call__(self):
reads.append('callable')
callback = Callback()
kernel.watch_output(capture.id, callback)
output.read(capture.id)
kernel.watch_output(capture.id, callback.__call__)
kernel.watch_output(capture.id, lambda: reads.append('replacement'))
del callback
gc.collect()
output.read(capture.id)
assert reads == ['function', 'callable', 'replacement']
kernel.forget_output(capture.id)
assert capture.id not in kernel.READ_WATCHERS
""")
if __name__ == "__main__":
unittest.main()