from __future__ import annotations
import hashlib
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
import pytest

HERE = Path(__file__).parent
CONTROLLER = HERE / "cleanup_controller.py"
CONTRACT = HERE / "schema-contract-v2.json"
FIXTURE = HERE / "fixtures/success"
CANDIDATE = "2e0894eac92bc396cc4723bf1f18ebc653b95018dd41574df435941c235da925"
WHEEL = "af4a9d0a1ffffb6eb7551c1d6dc2b32853ca6d024332a4f8f5702bbf992f141b"
PLAN = "7ace03c6dad33d2fc3ef223621cbca68a150fde8429932138e252fb8498ac582"
APPROVAL = "TASK26_POST_LIFECYCLE_ARCHIVE_CLEANUP_APPROVED"
ACTOR = "8527916639"


def sha(p: Path) -> str:
    return hashlib.sha256(p.read_bytes()).hexdigest()


def setup(tmp: Path, active=False):
    p = tmp / "profile"
    o = tmp / "other"
    shutil.copytree(FIXTURE, p)
    (p / "fixture-ids.json").unlink()
    o.mkdir(mode=0o700)
    (o / "marker").write_bytes(b"other")
    (o / "marker").chmod(0o600)
    system = tmp / "systemctl"
    system.write_text('#!/bin/sh\nprintf "inactive\\ndead\\n0\\n"\n')
    system.chmod(0o700)
    if active:
        system.write_text('#!/bin/sh\nprintf "active\\nrunning\\n123\\n"\n')
        system.chmod(0o700)
    perm = tmp / "permission.json"
    v = {
        "schema": "task26-post-lifecycle-cleanup-permission-v2",
        "approval": APPROVAL,
        "candidate_digest": CANDIDATE,
        "wheel_sha256": WHEEL,
        "plan_sha256": PLAN,
        "actor_id": ACTOR,
        "controller_sha256": sha(CONTROLLER),
        "contract_sha256": sha(CONTRACT),
        "execute_allowed": True,
        "lifecycle_binding": "derive_unique_active_and_seal_in_archive",
    }
    perm.write_text(json.dumps(v, sort_keys=True, separators=(",", ":")) + "\n")
    perm.chmod(0o600)
    return p, o, system, perm


def run(p, o, s, perm, mode, *extra):
    return subprocess.run(
        [
            sys.executable,
            str(CONTROLLER),
            mode,
            "--profile",
            str(p),
            "--other-profile",
            str(o),
            "--archive-root",
            str(p / "data/post-lifecycle-cleanup-archives"),
            "--contract",
            str(CONTRACT),
            "--permission",
            str(perm),
            "--candidate",
            CANDIDATE,
            "--wheel-sha256",
            WHEEL,
            "--plan-sha256",
            PLAN,
            "--approval",
            APPROVAL,
            "--actor-id",
            ACTOR,
            "--run-id",
            "fixture-cleanup",
            "--systemctl",
            str(s),
            *extra,
        ],
        text=True,
        capture_output=True,
    )


def edit_json(p: Path, rel: str, fn):
    x = p / rel
    v = json.loads(x.read_text())
    fn(v)
    x.write_text(json.dumps(v, sort_keys=True, separators=(",", ":")) + "\n")
    x.chmod(0o600)


def test_success_archives_active_ledger_before_clear_and_verifies(tmp_path):
    p, o, s, perm = setup(tmp_path)
    raw = (
        p / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    ).read_bytes()
    r = run(p, o, s, perm, "execute")
    assert r.returncode == 0, r.stderr
    a = p / "data/post-lifecycle-cleanup-archives/fixture-cleanup"
    m = json.loads((a / "manifest.json").read_text())
    assert m["active_lifecycle"]["session_id"] == "session-new-001"
    assert m["archive_committed_before_clear"] is True
    assert (
        a / "payload/data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    ).read_bytes() == raw
    assert not (p / "customers").exists()
    v = run(p, o, s, perm, "verify", "--archive", str(a))
    assert v.returncode == 0, v.stderr


@pytest.mark.parametrize(
    "case",
    [
        "enabled",
        "pending_delivery",
        "pending_provider",
        "unknown",
        "missing_delivery",
        "duplicate_delivery",
        "duplicate_session",
        "active_service",
        "id_mismatch",
        "unknown_schema",
        "symlink",
        "hardlink",
        "mode",
    ],
)
def test_strict_negatives_are_nonmutating(tmp_path, case):
    p, o, s, perm = setup(tmp_path, case == "active_service")
    ledger = p / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    before = ledger.read_bytes()
    if case == "enabled":
        edit_json(
            p,
            "customers/registry.json",
            lambda v: v["customers"][0].update(enabled=True),
        )
    elif case == "pending_delivery":
        edit_json(
            p,
            "data/owner-actions/draft-deliveries.json",
            lambda v: next(iter(v.values())).update(status="pending"),
        )
    elif case == "pending_provider":
        edit_json(
            p,
            "data/owner-actions/draft-generations.json",
            lambda v: next(iter(v.values()))[-1].update(
                state="claimed", claim_id="claim"
            ),
        )
    elif case == "unknown":
        edit_json(
            p,
            "data/owner-actions/draft-deliveries.json",
            lambda v: next(iter(v.values())).update(status="unknown_provider_outcome"),
        )
    elif case == "missing_delivery":
        (p / "data/owner-actions/draft-deliveries.json").write_text("{}\n")
    elif case == "duplicate_delivery":
        edit_json(
            p,
            "data/owner-actions/draft-deliveries.json",
            lambda v: v.update({"second": dict(next(iter(v.values())))}),
        )
    elif case == "duplicate_session":
        edit_json(
            p,
            "data/onboarding/telegram-customer-bootstrap-v1/ledger.json",
            lambda v: v["sessions"].append(
                dict(v["sessions"][-1], session_id="second-active")
            ),
        )
    elif case == "id_mismatch":
        edit_json(
            p,
            "data/onboarding/telegram-customer-bootstrap-v1/ledger.json",
            lambda v: v["sessions"][-1]["role_claims"][0].update(user_id="999"),
        )
    elif case == "unknown_schema":
        edit_json(
            p,
            "data/onboarding/telegram-customer-bootstrap-v1/ledger.json",
            lambda v: v.update(schema="future-v9"),
        )
    elif case == "symlink":
        ledger.unlink()
        ledger.symlink_to(p / "customers/registry.json")
    elif case == "hardlink":
        os.link(ledger, ledger.with_name("alias"))
    elif case == "mode":
        ledger.chmod(0o644)
    r = run(p, o, s, perm, "execute")
    assert r.returncode != 0
    assert not (p / "data/post-lifecycle-cleanup-archives/fixture-cleanup").exists()
    if case not in {"symlink", "unknown_schema", "duplicate_session", "id_mismatch"}:
        assert ledger.read_bytes() == before


@pytest.mark.parametrize(
    "case",
    [
        "future_job",
        "claim",
        "attempt_lock",
        "pending_outbox",
        "owner_nonterminal",
        "duplicate_customer",
    ],
)
def test_no_future_or_duplicate_authority(tmp_path, case):
    p, o, s, perm = setup(tmp_path)
    if case == "future_job":
        edit_json(
            p,
            "cron/jobs.json",
            lambda v: v["jobs"].append({"id": "future", "enabled": True}),
        )
    elif case == "claim":
        x = p / "data/customer-schedule-claims/customer-new-001"
        x.mkdir(parents=True)
        x.chmod(0o700)
    elif case == "attempt_lock":
        x = p / "data/.scheduled-delivery-attempt-x.lock"
        x.touch()
        x.chmod(0o600)
    elif case == "pending_outbox":
        edit_json(
            p,
            "data/onboarding/telegram-publication-outbox-v1/ledger.json",
            lambda v: v["records"][0].update(state="pending"),
        )
    elif case == "owner_nonterminal":
        edit_json(
            p,
            "data/owner-actions/customer-service-state.json",
            lambda v: next(iter(v["states"].values())).update(
                owner_action_state="pending"
            ),
        )
    elif case == "duplicate_customer":
        edit_json(
            p,
            "customers/registry.json",
            lambda v: v["customers"].append(
                dict(v["customers"][0], customer_key="second")
            ),
        )
    assert run(p, o, s, perm, "execute").returncode != 0


def test_injected_archive_commit_failure_never_clears(tmp_path):
    p, o, s, perm = setup(tmp_path)
    before = (p / "customers/registry.json").read_bytes()
    r = run(p, o, s, perm, "execute", "--test-fail-before-archive-commit")
    assert r.returncode != 0
    assert (p / "customers/registry.json").read_bytes() == before
    assert not (p / "data/post-lifecycle-cleanup-archives/fixture-cleanup").exists()