from __future__ import annotations

import hashlib
import importlib.util
import json
import os

import pytest
from pathlib import Path
from types import ModuleType
from collections.abc import Callable
from typing import Protocol, cast


class ResidueRuntime(Protocol):
    def open_exact_lock_residue(self, path: Path, expected: dict[str, object]) -> int: ...


class Launcher(Protocol):
    def sealed_memfd(self, name: str, content: bytes) -> int: ...
    def source_snapshots(self) -> object: ...
    def run_snapshot_child(
        self,
        entries: dict[str, list[object]],
        descriptor: int,
        digest: str,
        after_fork: object | None = None,
        lock_handoff: tuple[Path, Path, int, dict[str, object]] | None = None,
    ) -> int: ...


ROOT = Path(__file__).resolve().parent


def _json_object(path: Path) -> dict[str, object]:
    import json

    value = cast(object, json.loads(path.read_text(encoding="utf-8")))
    if not isinstance(value, dict):
        raise AssertionError("proof is not a JSON object")
    mapping = cast(dict[object, object], value)
    if not all(isinstance(key, str) for key in mapping):
        raise AssertionError("proof has a non-string key")
    return cast(dict[str, object], mapping)


def _launcher() -> Launcher:
    path = ROOT / "task22_launcher_test_support.py"
    spec = importlib.util.spec_from_file_location("identity_support", path)
    if spec is None or spec.loader is None:
        raise AssertionError("test support has no loader")
    module: ModuleType = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    load = cast(Callable[[Path], object], getattr(module, "load_launcher", None))
    if not callable(load):
        raise AssertionError("test support has no launcher adapter")
    return cast(Launcher, load(ROOT / "task22-trainer-authority-removal-canonical.py"))


def test_sealed_module_file_identity_reads_pinned_bytes_after_path_replacement(
    tmp_path: Path,
) -> None:
    launcher = _launcher()
    proof = tmp_path / "proof.json"
    canonical = tmp_path / "fixture.py"
    replacement_marker = tmp_path / "replacement-ran"
    module = (
        "import hashlib, json, sys\n"
        "from pathlib import Path\n"
        "def _source_bindings():\n"
        " source = Path(__file__).read_bytes(); spec = __spec__\n"
        " packages={name:{'file':m.__file__,'origin':m.__spec__.origin,'loader_path':m.__loader__.path,'package':m.__package__,'path':m.__path__} for name,m in ((n,sys.modules[n]) for n in ('gateway','gateway.platforms'))}\n"
        " return {'file':__file__,'digest':hashlib.sha256(source).hexdigest(),"
        "'origin':spec.origin,'loader_path':__loader__.path,"
        "'package':__package__,'has_location':spec.has_location,'packages':packages}\n"
    ).encode()
    replacement = (
        f"from pathlib import Path\nPath({str(replacement_marker)!r}).write_text('bad')\n"
    ).encode()
    cli = (
        "import json\nfrom pathlib import Path\n"
        "import gateway.platforms.fixture as fixture\n"
        f"Path({str(proof)!r}).write_text(json.dumps(fixture._source_bindings(),sort_keys=True))\n"
    ).encode()
    cli_fd = launcher.sealed_memfd("file-identity-cli", cli)
    module_fd = launcher.sealed_memfd("file-identity-module", module)
    gateway_fd = launcher.sealed_memfd("file-identity-gateway", b"")
    platforms_fd = launcher.sealed_memfd("file-identity-platforms", b"")
    entries: dict[str, list[object]] = {
        "gateway": [f"/proc/self/fd/{gateway_fd}", hashlib.sha256(b"").hexdigest(), True],
        "gateway.platforms": [f"/proc/self/fd/{platforms_fd}", hashlib.sha256(b"").hexdigest(), True],
        "gateway.platforms.fixture": [
            f"/proc/self/fd/{module_fd}", hashlib.sha256(module).hexdigest(), False,
        ],
    }
    descriptors = (cli_fd, module_fd, gateway_fd, platforms_fd)
    try:
        status = launcher.run_snapshot_child(
            entries,
            cli_fd,
            hashlib.sha256(cli).hexdigest(),
            lambda: canonical.write_bytes(replacement),
        )
    finally:
        for descriptor in descriptors:
            os.close(descriptor)

    assert status == 0
    identity = _json_object(proof)
    sealed_path = f"/proc/self/fd/{module_fd}"
    packages_value = identity.pop("packages")
    if not isinstance(packages_value, dict):
        raise AssertionError("package identities are not an object")
    packages = cast(dict[str, object], packages_value)
    assert identity == {
        "digest": hashlib.sha256(module).hexdigest(),
        "file": sealed_path,
        "has_location": True,
        "loader_path": sealed_path,
        "origin": sealed_path,
        "package": "gateway.platforms",
    }
    assert set(packages) == {"gateway", "gateway.platforms"}
    for name, package_value in packages.items():
        if not isinstance(package_value, dict):
            raise AssertionError("package identity is not an object")
        package = cast(dict[str, object], package_value)
        package_file = package["file"]
        assert package_file == package["origin"] == package["loader_path"]
        assert isinstance(package_file, str) and package_file.startswith("/proc/self/fd/")
        assert package["package"] == name
        assert package["path"] == []
    assert not replacement_marker.exists()


def test_exact_empty_live_residue_model_is_fail_closed(tmp_path: Path) -> None:
    path = ROOT / "task22-trainer-authority-removal-canonical.py"
    spec = importlib.util.spec_from_file_location("residue_launcher", path)
    if spec is None or spec.loader is None:
        raise AssertionError("launcher has no loader")
    module: ModuleType = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    expected = cast(dict[str, object], getattr(module, "EXPECTED_LOCK_RESIDUE"))
    assert expected == {
        "inode": 33166558,
        "mode": 0o600,
        "uid": 1000,
        "gid": 1000,
        "size": 0,
        "sha256": hashlib.sha256(b"").hexdigest(),
    }
    residue = tmp_path / ".trainer-authority-removal-v1.lock"
    residue.touch(mode=0o600)
    residue.chmod(0o600)
    modeled = {**expected, "inode": residue.stat().st_ino}
    runtime_factory = cast(Callable[[], ResidueRuntime], getattr(module, "_runtime"))
    assert callable(runtime_factory)
    runtime = runtime_factory()
    descriptor = runtime.open_exact_lock_residue(residue, modeled)
    os.close(descriptor)
    with open(residue, "wb") as stream:
        _ = stream.write(b"x")
    try:
        _ = runtime.open_exact_lock_residue(residue, modeled)
    except RuntimeError as exc:
        assert "identity changed" in str(exc)
    else:
        raise AssertionError("changed lock residue was accepted")


def test_recovery_closure_loads_only_the_sealed_reachable_modules(
    capfd: pytest.CaptureFixture[str],
) -> None:
    launcher = _launcher()
    raw = launcher.source_snapshots()
    if not isinstance(raw, tuple):
        raise AssertionError("launcher recovery snapshots are invalid")
    pair = cast(tuple[object, ...], raw)
    if len(pair) != 2:
        raise AssertionError("launcher recovery snapshots are invalid")
    raw_entries, raw_descriptors = pair
    if not isinstance(raw_entries, dict) or not isinstance(raw_descriptors, list):
        raise AssertionError("launcher recovery snapshots are invalid")
    entry_values = cast(dict[object, object], raw_entries)
    descriptor_values = cast(list[object], raw_descriptors)
    entries: dict[str, list[object]] = {}
    for name, entry in entry_values.items():
        if not isinstance(name, str) or not isinstance(entry, list):
            raise AssertionError("launcher recovery snapshot entry is invalid")
        entries[name] = cast(list[object], entry)
    descriptors: list[int] = []
    for descriptor_value in descriptor_values:
        if not isinstance(descriptor_value, int) or isinstance(descriptor_value, bool):
            raise AssertionError("launcher recovery descriptor is invalid")
        descriptors.append(descriptor_value)
    expected = {
        "gateway",
        "gateway.platforms",
        "gateway.platforms.task22_callback_storage_adoption",
        "gateway.platforms.task22_protected_session_inventory",
        "gateway.platforms.task22_trainer_authority_recovery",
        "gateway.platforms.task22_trainer_authority_removal",
        "gateway.platforms.telegram_nutrition_onboarding_publication_outbox",
    }
    assert set(entries) == expected
    cli = (
        "import json,sys\n"
        "import gateway.platforms.task22_trainer_authority_recovery as recovery\n"
        "import gateway.platforms.task22_trainer_authority_removal as removal\n"
        "loaded=sorted(name for name in sys.modules if name.startswith('gateway.platforms.task22_') or name.endswith('publication_outbox'))\n"
        "print(json.dumps({'loaded':loaded,'recovery':recovery.__spec__.origin,'predecessor':removal.__spec__.origin},sort_keys=True))\n"
    ).encode()
    descriptor = launcher.sealed_memfd("recovery-closure-cli", cli)
    try:
        status = launcher.run_snapshot_child(
            entries, descriptor, hashlib.sha256(cli).hexdigest(), None
        )
    finally:
        os.close(descriptor)
        for item in descriptors:
            os.close(item)
    assert status == 0
    value = cast(object, json.loads(capfd.readouterr().out))
    if not isinstance(value, dict):
        raise AssertionError("recovery closure result is invalid")
    result_values = cast(dict[object, object], value)
    if not all(isinstance(key, str) for key in result_values):
        raise AssertionError("recovery closure result is invalid")
    result = cast(dict[str, object], result_values)
    assert result["loaded"] == [
        "gateway.platforms.task22_callback_storage_adoption",
        "gateway.platforms.task22_protected_session_inventory",
        "gateway.platforms.task22_trainer_authority_recovery",
        "gateway.platforms.task22_trainer_authority_removal",
        "gateway.platforms.telegram_nutrition_onboarding_publication_outbox",
    ]
    recovery_origin, predecessor_origin = result["recovery"], result["predecessor"]
    assert isinstance(recovery_origin, str) and recovery_origin.startswith("/proc/self/fd/")
    assert isinstance(predecessor_origin, str) and predecessor_origin.startswith("/proc/self/fd/")