# SPDX-License-Identifier: CC0-1.0
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch

HERE = Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location("lifetime", HERE / "zero-copy-lifetime.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)


class LifetimeTests(unittest.TestCase):
    def test_buffered_control_is_a_copy(self):
        row = module.probe("buffered-overwrite")
        self.assertEqual(row["received"]["uniform_byte"], "A")
        self.assertEqual(row["path_after"]["uniform_byte"], "B")

    def test_consumed_control_retains_received_bytes(self):
        row = module.probe("sendfile-consume-then-overwrite")
        self.assertEqual(row["received"]["uniform_byte"], "A")
        self.assertLess(row["events"].index("consume"), row["events"].index("overwrite-source"))

    def test_replace_changes_identity_not_old_file(self):
        row = module.probe("sendfile-replace-path")
        self.assertFalse(row["path_same_inode"])
        self.assertEqual(row["received"]["uniform_byte"], "A")
        self.assertEqual(row["old_inode_after"]["uniform_byte"], "A")
        self.assertEqual(row["path_after"]["uniform_byte"], "B")

    def test_replace_does_not_revoke_old_writer(self):
        row = module.probe("sendfile-replace-then-write-old")
        self.assertFalse(row["path_same_inode"])
        self.assertEqual(row["old_inode_after"]["uniform_byte"], "C")
        self.assertEqual(row["path_after"]["uniform_byte"], "B")

    def test_explicitly_checks_short_transfers(self):
        with patch.object(module.os, "sendfile", return_value=0):
            with self.assertRaisesRegex(RuntimeError, "short transfer"):
                module.probe("sendfile-overwrite")

    def test_cleanup_and_case_coverage(self):
        with tempfile.TemporaryDirectory() as parent:
            report = module.run(2, parent)
            self.assertEqual(list(Path(parent).iterdir()), [])
        self.assertEqual(len(report["observations"]), 16)
        for row in report["observations"]:
            self.assertEqual(row["received"]["bytes"], row["page_size_bytes"])
            self.assertEqual(row["enqueue_return_bytes"], row["page_size_bytes"])
            self.assertEqual(row["received_matches_pretransfer_hash"],
                             row["received"]["sha256"] == row["source_before"]["sha256"])

    def test_invalid_inputs(self):
        for value in (0, -1, 101, 1.5, True):
            with self.assertRaises(ValueError):
                module.run(value)
        with self.assertRaises(ValueError):
            module.probe("unknown")

    def test_published_receipt(self):
        report = json.loads((HERE / "zero-copy-lifetime.json").read_text())
        self.assertEqual(report["sourceSha256"], module.digest((HERE / "zero-copy-lifetime.py").read_bytes()))
        observed = {case: set() for case in module.CASES}
        for row in report["observations"]:
            observed[row["case"]].add(row["received"]["uniform_byte"])
        expected = dict(zip(module.CASES, ["A", "B", "B", "B", "A", "C", "B", "A"]))
        self.assertEqual(observed, {case: {value} for case, value in expected.items()})


if __name__ == "__main__":
    unittest.main()