from __future__ import annotations

import importlib.util
import json
import os
import subprocess
import sys
from pathlib import Path

import pytest

HERE = Path(__file__).parent
HARNESS = HERE / "consent_observer.py"
SPEC = importlib.util.spec_from_file_location("consent_observer", HARNESS)
assert SPEC and SPEC.loader
observer = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = observer
SPEC.loader.exec_module(observer)


def atomic(path: Path, value: object) -> None:
    tmp = path.with_name(f".{path.name}.next")
    tmp.write_text(json.dumps(value, sort_keys=True, separators=(",", ":")))
    tmp.chmod(0o600)
    os.replace(tmp, path)


def ledger_session() -> dict[str, object]:
    return {
        "session_id": observer.SESSION_ID, "sid_hash": observer.SID_HASH,
        "customer_draft": {"customer_key": observer.CUSTOMER_KEY, "customer_user_id": observer.ACTOR_ID},
        "customer_draft_digest": "d" * 64, "bot_username": "dual_coach_pilot_test_bot",
        "state": "AWAITING_CONSENT", "generation": 4, "created_at": "x", "updated_at": "x",
        "expires_at": "x", "owner_id": observer.OWNER_ID,
        "role_claims": [{"role": "customer", "user_id": observer.ACTOR_ID,
                         "chat_id": observer.CHAT_ID, "topic_id": observer.TOPIC_ID,
                         "message_id": observer.CLAIM_MESSAGE_ID}],
        "consent_publication_attempt": 1, "consent_card_message_id": observer.CARD_MESSAGE_ID,
        "recovery_attempt_generation": 1, "recovery_attempts": [], "failure_code": None,
    }


def ledger(rows: list[dict[str, object]]) -> dict[str, object]:
    payload = {"schema": "telegram-customer-bootstrap-v1", "sessions": rows}
    return {**payload, "digest": observer.digest_json(payload)}


def registry(granted: bool = False) -> dict[str, object]:
    return {"version": 1, "registry_mode": "ordinary_v1", "diagnostic_session_digest": None,
            "owner": {"user_id": observer.OWNER_ID, "chat_id": observer.OWNER_ID, "topic_id": "0"},
            "customers": [{"customer_key": observer.CUSTOMER_KEY, "enabled": False,
                "telegram": {"user_id": observer.ACTOR_ID, "chat_id": observer.CHAT_ID,
                             "topic_id": observer.TOPIC_ID},
                "ai_processing_consent": {"granted": granted,
                    "recorded_on": "2026-08-16" if granted else None,
                    "notice_version": "privacy-v1" if granted else None}}]}


def profile(tmp_path: Path) -> Path:
    root = tmp_path / "profile"
    (root / "customers").mkdir(parents=True)
    (root / "data/onboarding/telegram-customer-bootstrap-v1").mkdir(parents=True)
    for path in (root, root / "customers", root / "data", root / "data/onboarding",
                 root / "data/onboarding/telegram-customer-bootstrap-v1"):
        path.chmod(0o700)
    atomic(root / "customers/registry.json", registry())
    atomic(root / observer.LEDGER_RELATIVE, ledger([ledger_session()]))
    return root


def test_baseline_seals_exact_current_callback_authority(tmp_path: Path) -> None:
    root = profile(tmp_path)
    baseline = observer.arm_baseline(root, enforce_runtime=False)
    assert baseline.session["generation"] == 4
    assert baseline.session["consent_card_message_id"] == "159"
    assert baseline.expected_consent_receipt_sha256 == observer.CONSENT_RECEIPT_SHA256


@pytest.mark.parametrize("field,value", [("state", "ACTIVE"), ("generation", 5),
                                           ("consent_card_message_id", "160")])
def test_rejects_stale_consent_baseline(tmp_path: Path, field: str, value: object) -> None:
    root = profile(tmp_path)
    row = ledger_session()
    row[field] = value
    atomic(root / observer.LEDGER_RELATIVE, ledger([row]))
    with pytest.raises(observer.ObserverError):
        observer.arm_baseline(root, enforce_runtime=False)


def test_event_first_observes_registry_then_exact_bootstrap_commit(tmp_path: Path) -> None:
    root = profile(tmp_path)
    ready = tmp_path / "ready.json"
    receipt = tmp_path / "receipt.json"
    read_fd, write_fd = os.pipe()
    cmd = [sys.executable, "-B", str(HARNESS), "observe", "--profile", str(root),
           "--ready", str(ready), "--receipt", str(receipt), "--timeout", "3",
           "--skip-runtime-hashes", "--ready-fd", str(write_fd)]
    proc = subprocess.Popen(cmd, pass_fds=(write_fd,), stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE, text=True)
    os.close(write_fd)
    ready_wire = json.loads(os.read(read_fd, 4096))
    os.close(read_fd)
    assert ready_wire["status"] == "READY_BEFORE_CONSENT"
    atomic(root / "customers/registry.json", registry(True))
    row = ledger_session()
    row.update({"state": "AWAITING_ACTIVATION", "generation": 5,
                                        "consent_card_message_id": None, "updated_at": "y"})
    atomic(root / observer.LEDGER_RELATIVE, ledger([row]))
    stdout, stderr = proc.communicate(timeout=4)
    assert proc.returncode == 0, stderr
    result = json.loads(receipt.read_text())
    assert result["status"] == "PASS_CONSENT_COMMITTED"
    assert result["callback_authority"]["message_id"] == "159"
    assert json.loads(stdout.splitlines()[-1])["status"] == "PASS_CONSENT_COMMITTED"


def test_wrong_registry_or_bootstrap_transition_fails_closed(tmp_path: Path) -> None:
    root = profile(tmp_path)
    base = observer.arm_baseline(root, enforce_runtime=False)
    wrong = registry(True)
    wrong["customers"][0]["telegram"]["chat_id"] = "9"
    with pytest.raises(observer.ObserverError):
        observer.validate_registry_transition(base.registry, wrong)
    row = ledger_session()
    row.update({"state": "AWAITING_ACTIVATION", "generation": 6,
                                        "consent_card_message_id": None})
    with pytest.raises(observer.ObserverError):
        observer.validate_ledger_transition(base.ledger, ledger([row]))


def test_arm_only_emits_ready_receipt_without_profile_mutation(tmp_path: Path) -> None:
    root = profile(tmp_path)
    before = observer.tree_digest(root)
    receipt = tmp_path / "arm.json"
    result = subprocess.run([sys.executable, "-B", str(HARNESS), "arm-only",
        "--profile", str(root), "--receipt", str(receipt), "--skip-runtime-hashes"],
        text=True, capture_output=True, check=False)
    assert result.returncode == 0, result.stderr
    assert json.loads(receipt.read_text())["status"] == "READY_BEFORE_CONSENT"
    assert observer.tree_digest(root) == before


def test_timeout_is_bounded_without_polling(tmp_path: Path) -> None:
    root = profile(tmp_path)
    result = subprocess.run([sys.executable, "-B", str(HARNESS), "observe",
        "--profile", str(root), "--ready", str(tmp_path / "ready.json"),
        "--receipt", str(tmp_path / "receipt.json"), "--timeout", "0.03",
        "--skip-runtime-hashes"], text=True, capture_output=True, check=False)
    assert result.returncode == 3
    assert "monotonic deadline expired" in result.stderr