"""Binary pipe boundaries and early reader closure are race-prone kernel behavior the daemon E2E suite cannot drive reliably.""" from pathlib import Path import sys import unittest sys.path.insert(0, str(Path(__file__).resolve().parent)) sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "priv" / "python")) from job_wake_test import Owner # noqa: E402 class RunTest(unittest.TestCase): def setUp(self): self.owner = Owner(["run"]) self.addCleanup(self.owner.close) self.owner.wait_for(lambda f: f.get("type") == "ready") self.cells = 0 def cell(self, code): self.cells += 1 id = f"c{self.cells}" self.owner.send({"type": "execute", "id": id, "code": code}) done = self.owner.wait_for( lambda f: f.get("type") == "done" and f.get("id") == id ) return done def test_a_pipe_carries_bytes_exactly_and_past_the_retention_cap(self): done = self.cell( "import hashlib, sys\n" "blob = bytes(range(256)) * 20000\n" "writer = run(sys.executable, '-c', 'import sys; sys.stdout.buffer.write(bytes(range(256)) * 20000)')\n" "reader = await writer.pipe(sys.executable, '-c', 'import hashlib, sys; " "print(hashlib.sha256(sys.stdin.buffer.read()).hexdigest())')\n" "reader.tail().strip() == hashlib.sha256(blob).hexdigest()" ) self.assertEqual(done["value"], "True") def test_late_pipe_preserves_non_utf8_bytes(self): done = self.cell( "import sys\n" "writer = run(sys.executable, '-c', 'import sys; sys.stdout.buffer.write(bytes([0, 255, 254, 10]))')\n" "await writer\n" "reader = await run(sys.executable, '-c', 'import sys; print(list(sys.stdin.buffer.read()))', stdin=writer)\n" "reader.tail().strip()" ) self.assertEqual(done["value"], "'[0, 255, 254, 10]'") def test_a_reader_that_stops_ends_the_writer(self): done = self.cell( "import asyncio\nwriter = run('yes')\nreader = await writer.pipe('head', '-2')\n" "await asyncio.wait_for(asyncio.shield(writer.task), 5)\n" "(reader.tail(), writer.exit_code != 0)" ) self.assertEqual(done["value"], "('y\\ny\\n', True)") def test_a_finished_job_feeds_what_it_retained(self): done = self.cell( "out = await run('printf', 'x\\ny\\n')\n(await run('wc', '-l', stdin=out)).tail().strip()" ) self.assertEqual(done["value"], "'2'") done = self.cell( "import sys\nbig = await run(sys.executable, '-c', 'print(\"x\" * 2000000)')\n" "run('wc', '-c', stdin=big)" ) self.assertIn("pipe from it before it runs", done["output"]) if __name__ == "__main__": unittest.main()