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_v4", 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-v4.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