"""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()