from __future__ import annotations

import hashlib
import importlib.util
import json
import os
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock

import pytest

HERE = Path(__file__).parent
V3 = HERE.parent / "task26-post-lifecycle-cleanup-v3-4a6c7ee5-st_01a00e50"


def load(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


MODULE = load("cleanup_v5", HERE / "cleanup_controller.py")
V3_MODULE = load("cleanup_v3_regression", V3 / "cleanup_controller.py")


def temp_entries(path: Path) -> list[Path]:
    return list(path.glob(".task26-cleanup-*"))


def test_atomic_repeated_partial_writes_publish_exact_bytes(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    real_write = os.write
    calls: list[int] = []

    def partial(fd: int, raw: bytes | memoryview) -> int:
        size = min(2, len(raw))
        calls.append(size)
        return real_write(fd, raw[:size])

    monkeypatch.setattr(MODULE.os, "write", partial)
    target = tmp_path / "payload"
    expected = b"0123456789abcdef"
    MODULE.atomic(target, expected)
    assert target.read_bytes() == expected
    assert len(calls) == 8
    assert temp_entries(tmp_path) == []


def test_atomic_retries_interrupted_write_then_succeeds(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    real_write = os.write
    attempts = 0

    def interrupted_once(fd: int, raw: bytes | memoryview) -> int:
        nonlocal attempts
        attempts += 1
        if attempts == 1:
            raise InterruptedError
        return real_write(fd, raw)

    monkeypatch.setattr(MODULE.os, "write", interrupted_once)
    target = tmp_path / "receipt.json"
    MODULE.atomic(target, b'{"status":"PASS"}\n')
    assert target.read_bytes() == b'{"status":"PASS"}\n'
    assert attempts == 2
    assert temp_entries(tmp_path) == []


@pytest.mark.parametrize("bad_count", [0, -1, 99, True])
def test_atomic_rejects_invalid_write_counts_without_rename(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, bad_count: int
) -> None:
    target = tmp_path / "archive"
    target.write_bytes(b"sealed-old")
    monkeypatch.setattr(MODULE.os, "write", lambda _fd, _raw: bad_count)
    with pytest.raises(MODULE.CleanupError, match="invalid write count"):
        MODULE.atomic(target, b"new-archive")
    assert target.read_bytes() == b"sealed-old"
    assert temp_entries(tmp_path) == []


def test_atomic_oserror_never_renames_or_leaves_temp(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    target = tmp_path / "receipt.json"

    def fail(_fd: int, _raw: bytes | memoryview) -> int:
        raise OSError("simulated media failure")

    monkeypatch.setattr(MODULE.os, "write", fail)
    with pytest.raises(OSError, match="media failure"):
        MODULE.atomic(target, b"must-not-publish")
    assert not target.exists()
    assert temp_entries(tmp_path) == []


def test_archive_write_failure_removes_pending_and_runs_no_mutation_command(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    profile = tmp_path / "profile"
    profile.mkdir()
    (profile / "authority.json").write_bytes(b"authority")
    args = SimpleNamespace(profile=profile, archive_root=tmp_path / "archives", run_id="run")
    mutation_runner = Mock(side_effect=AssertionError("mutation command must not run"))
    monkeypatch.setattr(MODULE.subprocess, "run", mutation_runner)
    monkeypatch.setattr(MODULE.os, "write", Mock(side_effect=OSError("archive write failed")))
    rows = [{"path": "authority.json", "size": 9, "sha256": MODULE.digest(b"authority"), "mode": "0600"}]
    with pytest.raises(OSError, match="archive write failed"):
        MODULE.archive_prestate(args, {"bindings": {}, "service": {"name": "unused"}}, ["authority.json"], rows, {}, {})
    assert not (args.archive_root / "run").exists()
    assert not (args.archive_root / ".pending-run").exists()
    assert mutation_runner.call_count == 0


def test_regression_v3_truncates_partial_write_while_v4_writes_all(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    real_write = os.write

    def partial(fd: int, raw: bytes | memoryview) -> int:
        return real_write(fd, raw[:3])

    old_target = tmp_path / "v3"
    monkeypatch.setattr(V3_MODULE.os, "write", partial)
    V3_MODULE.atomic(old_target, b"abcdefgh")
    assert old_target.read_bytes() == b"abc"
    new_target = tmp_path / "v4"
    MODULE.atomic(new_target, b"abcdefgh")
    assert new_target.read_bytes() == b"abcdefgh"


def test_completed_cleanup_blocks_execute_before_any_mutation(monkeypatch: pytest.MonkeyPatch) -> None:
    mutation_runner = Mock(side_effect=AssertionError("mutation command must not run"))
    monkeypatch.setattr(MODULE.subprocess, "run", mutation_runner)
    with pytest.raises(MODULE.CleanupError, match="forbids mutation replay"):
        MODULE.execute(SimpleNamespace(), {"completed_cleanup": {"status": "PASS_VERIFIED_DISABLED_ARCHIVED_CLEAN"}})
    assert mutation_runner.call_count == 0


def test_successor_bindings_are_exact_and_predecessors_are_immutable() -> None:
    contract = json.loads((HERE / "schema-contract-v5.json").read_text())
    completed = contract["completed_cleanup"]
    assert contract["bindings"]["candidate_digest"] == "4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb"
    assert contract["bindings"]["lifecycle_seal_sha256"] == "50e7556e58a876d8136fbfb2023ede8bcb46ac816d97d4b85e1d0559f463d008"
    assert completed["archive_root_sha256"] == "9f5d72840b4cc1fcb57720a61ebc53529fa00202d9201615e4ebdbb3e047a635"
    assert hashlib.sha256((V3 / "SEAL.json").read_bytes()).hexdigest() == completed["v3_seal_sha256"]
    assert hashlib.sha256((V3 / "inventory.json").read_bytes()).hexdigest() == completed["v3_inventory_sha256"]
    assert hashlib.sha256((V3 / "cleanup_controller.py").read_bytes()).hexdigest() == completed["v3_controller_sha256"]


def test_every_atomic_payload_and_receipt_write_uses_write_all() -> None:
    source = (HERE / "cleanup_controller.py").read_text()
    outside_helper = source[source.index("def atomic(") :]
    assert "os.write(" not in outside_helper
    assert "write_all(fd, raw)" in outside_helper
    assert source.count("def atomic(") == 1


def proc_stat(pid: int, *, ppid: int = 1, session: int = 10, start: int = 100) -> bytes:
    # stat fields 3..22: state, ppid, pgrp, session, then fillers through starttime.
    fields = ["S", str(ppid), "10", str(session)] + ["0"] * 15 + [str(start)]
    return f"{pid} (test process) {' '.join(fields)}\n".encode()


def add_process(root: Path, pid: int, argv: list[str], *, start: int = 100, ppid: int = 1) -> None:
    process = root / str(pid)
    process.mkdir()
    (process / "stat").write_bytes(proc_stat(pid, start=start, ppid=ppid))
    (process / "cmdline").write_bytes(b"\0".join(os.fsencode(arg) for arg in argv) + b"\0")


def test_journalctl_fu_follower_is_rejected_with_identity_evidence(tmp_path: Path) -> None:
    add_process(tmp_path, 101, ["journalctl", "-fu", MODULE.SERVICE_NAME], start=777)
    result = MODULE.enumerate_processes(Path("/target/profile"), MODULE.SERVICE_NAME, proc_root=tmp_path, self_pid=999)
    violation = result["violations"][0]
    assert violation["pid"] == 101
    assert violation["start_time_ticks"] == 777
    assert violation["cmdline"] == ["journalctl", "-fu", MODULE.SERVICE_NAME]
    assert "target_journal_follower" in violation["reasons"]


def test_target_inotify_observer_is_rejected(tmp_path: Path) -> None:
    profile = Path("/target/profile")
    add_process(tmp_path, 102, ["inotifywait", "-m", str(profile / "live-v64-events.jsonl")])
    result = MODULE.enumerate_processes(profile, MODULE.SERVICE_NAME, proc_root=tmp_path, self_pid=999)
    assert result["violations"][0]["reasons"] == [
        "exact_profile_path_reference",
        "observer_script_or_event",
        "target_inotify_watcher",
    ]


def test_profile_process_without_hermes_home_is_rejected(tmp_path: Path) -> None:
    profile = Path("/target/profile")
    add_process(tmp_path, 103, ["python", "gateway.py", "--state", str(profile / "state.db")])
    result = MODULE.enumerate_processes(profile, MODULE.SERVICE_NAME, proc_root=tmp_path, self_pid=999)
    assert result["violations"][0]["reasons"] == ["exact_profile_path_reference"]


def test_unrelated_service_follower_is_allowed(tmp_path: Path) -> None:
    add_process(tmp_path, 104, ["journalctl", "--user", "-fu", "other.service"])
    result = MODULE.enumerate_processes(Path("/target/profile"), MODULE.SERVICE_NAME, proc_root=tmp_path, self_pid=999)
    assert result["violations"] == []
    assert result["same_uid_processes"][0]["classification"] == "unrelated"


def test_self_is_excluded_even_when_command_references_every_target(tmp_path: Path) -> None:
    add_process(tmp_path, 105, ["journalctl", "-fu", MODULE.SERVICE_NAME, "/target/profile"])
    result = MODULE.enumerate_processes(Path("/target/profile"), MODULE.SERVICE_NAME, proc_root=tmp_path, self_pid=105)
    assert result["violations"] == []
    assert result["same_uid_processes"][0]["classification"] == "self_excluded"


def test_parent_shell_running_verifier_command_is_excluded(tmp_path: Path) -> None:
    profile = Path("/target/profile")
    add_process(tmp_path, 109, ["bash", "-c", f"python cleanup_controller.py verify --profile {profile}"])
    add_process(tmp_path, 110, ["python", "cleanup_controller.py", "verify", "--profile", str(profile)], ppid=109)
    result = MODULE.enumerate_processes(profile, MODULE.SERVICE_NAME, proc_root=tmp_path, self_pid=110)
    assert result["violations"] == []
    assert [row["classification"] for row in result["same_uid_processes"]] == [
        "verifier_command_excluded",
        "self_excluded",
    ]


def test_disappeared_pid_is_captured_as_race(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    add_process(tmp_path, 106, ["sleep", "1"])
    real_read = MODULE._read_proc_file
    calls = 0

    def disappear(proc_fd: int, name: str) -> bytes:
        nonlocal calls
        calls += 1
        if calls == 2:
            raise FileNotFoundError
        return real_read(proc_fd, name)

    monkeypatch.setattr(MODULE, "_read_proc_file", disappear)
    result = MODULE.enumerate_processes(Path("/target/profile"), MODULE.SERVICE_NAME, proc_root=tmp_path, self_pid=999)
    assert result["same_uid_processes"] == []
    assert result["process_races"] == [{"pid": 106, "race": "disappeared_during_read"}]


@pytest.mark.parametrize(
    ("filename", "raw", "message"),
    [
        ("stat", b"not a proc stat\n", "malformed /proc/107/stat"),
        ("cmdline", b"unterminated", "malformed /proc/107/cmdline"),
    ],
)
def test_malformed_same_uid_proc_entries_fail_closed(
    tmp_path: Path, filename: str, raw: bytes, message: str
) -> None:
    add_process(tmp_path, 107, ["sleep", "1"])
    (tmp_path / "107" / filename).write_bytes(raw)
    with pytest.raises(MODULE.CleanupError, match=message):
        MODULE.enumerate_processes(Path("/target/profile"), MODULE.SERVICE_NAME, proc_root=tmp_path, self_pid=999)


def test_two_pass_poststate_reports_no_background_sessions(tmp_path: Path) -> None:
    add_process(tmp_path, 108, ["journalctl", "-fu", "unrelated.service"])
    proof = MODULE.process_poststate(Path("/target/profile"), MODULE.SERVICE_NAME, proc_root=tmp_path)
    assert proof["status"] == "PASS"
    assert proof["passes"] == 2
    assert proof["target_processes"] == []