okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.petFormat Python sources and remove root HTML diagram59d7370

main
branchesmainall refs →
2.8 KiB70 linesraw
1"""Binary pipe boundaries and early reader closure are race-prone kernel behavior the daemon E2E suite cannot drive reliably."""
2
3from pathlib import Path
4import sys
5import unittest
6
7sys.path.insert(0, str(Path(__file__).resolve().parent))
8sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "priv" / "python"))
9from job_wake_test import Owner  # noqa: E402
10
11
12class RunTest(unittest.TestCase):
13    def setUp(self):
14        self.owner = Owner(["run"])
15        self.addCleanup(self.owner.close)
16        self.owner.wait_for(lambda f: f.get("type") == "ready")
17        self.cells = 0
18
19    def cell(self, code):
20        self.cells += 1
21        id = f"c{self.cells}"
22        self.owner.send({"type": "execute", "id": id, "code": code})
23        done = self.owner.wait_for(
24            lambda f: f.get("type") == "done" and f.get("id") == id
25        )
26        return done
27
28    def test_a_pipe_carries_bytes_exactly_and_past_the_retention_cap(self):
29        done = self.cell(
30            "import hashlib, sys\n"
31            "blob = bytes(range(256)) * 20000\n"
32            "writer = run(sys.executable, '-c', 'import sys; sys.stdout.buffer.write(bytes(range(256)) * 20000)')\n"
33            "reader = await writer.pipe(sys.executable, '-c', 'import hashlib, sys; "
34            "print(hashlib.sha256(sys.stdin.buffer.read()).hexdigest())')\n"
35            "reader.tail().strip() == hashlib.sha256(blob).hexdigest()"
36        )
37        self.assertEqual(done["value"], "True")
38
39    def test_late_pipe_preserves_non_utf8_bytes(self):
40        done = self.cell(
41            "import sys\n"
42            "writer = run(sys.executable, '-c', 'import sys; sys.stdout.buffer.write(bytes([0, 255, 254, 10]))')\n"
43            "await writer\n"
44            "reader = await run(sys.executable, '-c', 'import sys; print(list(sys.stdin.buffer.read()))', stdin=writer)\n"
45            "reader.tail().strip()"
46        )
47        self.assertEqual(done["value"], "'[0, 255, 254, 10]'")
48
49    def test_a_reader_that_stops_ends_the_writer(self):
50        done = self.cell(
51            "import asyncio\nwriter = run('yes')\nreader = await writer.pipe('head', '-2')\n"
52            "await asyncio.wait_for(asyncio.shield(writer.task), 5)\n"
53            "(reader.tail(), writer.exit_code != 0)"
54        )
55        self.assertEqual(done["value"], "('y\\ny\\n', True)")
56
57    def test_a_finished_job_feeds_what_it_retained(self):
58        done = self.cell(
59            "out = await run('printf', 'x\\ny\\n')\n(await run('wc', '-l', stdin=out)).tail().strip()"
60        )
61        self.assertEqual(done["value"], "'2'")
62        done = self.cell(
63            "import sys\nbig = await run(sys.executable, '-c', 'print(\"x\" * 2000000)')\n"
64            "run('wc', '-c', stdin=big)"
65        )
66        self.assertIn("pipe from it before it runs", done["output"])
67
68
69if __name__ == "__main__":
70    unittest.main()