# -*- coding: utf-8 -*-
"""tests.regression.test_real_merge_hooks_v2_step0_flow — task-2639.

Step 0 흐름 (input → auth match → snapshot crossref → sanctioned split →
existing gates) 흐름 + schema v2 (`allow_reason` + `snapshot_crossref`) 단언.

Spec: memory/specs/system_real_merge_hooks_snapshot_crossref_spec_260523.md §3 / §5
sha256: 12b8af006913833596562c55ab9a0acca935830be90c5f17f2af4b7e1e632621

ANCHOR-3 (task md): "schema v1→v2 bump · allow_reason + snapshot_crossref 필드 추가"
ANCHOR-6 (task md): "real_merge_hooks/real_merge_artifact_schema 만 정정 · 나머지
wiring stack 무수정"
"""
from __future__ import annotations

import json
from datetime import datetime
from pathlib import Path
from typing import Any

import pytest

from utils.real_merge_artifact_schema import MERGE_DECISION_SCHEMA
from utils.real_merge_hooks import (
    CHAIR_REQUIRED_ADMIN_OVERRIDE_REQUIRED,
    CHAIR_REQUIRED_BLOCKING_SECRET_IN_SNAPSHOT,
    CHAIR_REQUIRED_PRODUCTION_IN_SNAPSHOT,
    NO_OP_AUTH_MISMATCH,
    NO_OP_FORBIDDEN_PATH,
    NO_OP_RESULTS,
    REAL_MERGE_DONE,
    real_merge_execute,
)
from utils.snapshot_crossref_validator import ALLOW_REASON_SNAPSHOT_CROSSREF

FIXTURE_ROOT = (
    Path(__file__).resolve().parent.parent / "fixtures" / "snapshot_crossref"
)


def _kst_clock(text: str):
    dt = datetime.strptime(text, "%Y-%m-%dT%H:%M:%S%z")
    return lambda: dt


def _runner_must_never_run(*_args: Any, **_kwargs: Any) -> Any:
    raise AssertionError(
        "subprocess_runner was invoked during a snapshot crossref fixture — "
        "task-2639 safety invariant violated (live gh pr merge 호출 0)"
    )


def _load(scenario: str):
    base = FIXTURE_ROOT / scenario
    evidence = json.loads((base / "evidence.json").read_text(encoding="utf-8"))
    expected = json.loads((base / "expected.json").read_text(encoding="utf-8"))
    return evidence, expected


SCENARIOS = [
    ("fixture_in_snapshot_pass_candidate", REAL_MERGE_DONE),
    ("fixture_in_snapshot_mismatch_no_op", NO_OP_FORBIDDEN_PATH),
    ("fixture_wrong_head_sha", NO_OP_AUTH_MISMATCH),
    ("sanctioned_lock_separated", REAL_MERGE_DONE),
    ("production_in_snapshot_chair_required", CHAIR_REQUIRED_PRODUCTION_IN_SNAPSHOT),
    (
        "blocking_secret_in_snapshot_chair_required",
        CHAIR_REQUIRED_BLOCKING_SECRET_IN_SNAPSHOT,
    ),
    (
        "admin_override_required_chair_required",
        CHAIR_REQUIRED_ADMIN_OVERRIDE_REQUIRED,
    ),
]


@pytest.mark.parametrize("scenario,expected_enum", SCENARIOS)
def test_step0_flow_result_enum_and_schema_v2(scenario, expected_enum, tmp_path):
    """7 fixture 각각의 result_enum + schema v2 + snapshot_crossref 필드 단언."""
    evidence, expected = _load(scenario)
    clock = _kst_clock(evidence["frozen_now_kst"])
    canonical_root = str(tmp_path)
    # REAL_MERGE_DONE 경로는 inert (subprocess_runner=None) 로 통과하고, NO_OP/
    # CHAIR_REQUIRED 경로는 runner 가 절대 호출되지 않음을 sentinel 로 단언.
    runner = None if expected_enum == REAL_MERGE_DONE else _runner_must_never_run

    out = real_merge_execute(
        merge_ready_result=evidence.get("merge_ready_result"),
        pr_identity=evidence["pr_identity"],
        gate_snapshot=evidence.get("gate_snapshot"),
        dryrun_artifact=evidence.get("dryrun_artifact"),
        callback_envelope=evidence.get("callback_envelope"),
        chair_authorization=evidence.get("chair_authorization"),
        activation_flag=bool(evidence["activation_flag"]),
        subprocess_runner=runner,
        canonical_root=canonical_root,
        changed_files=evidence.get("changed_files"),
        clock=clock,
    )

    # result_enum 정확 일치 (expected.json 과 fixture 표 둘 다 일치).
    assert out["result_enum"] == expected_enum == expected["real_merge_result_enum"], scenario

    # subprocess 호출 0 — task-2639 safety invariant.
    assert out["subprocess_invocations"] == [], scenario

    decision = out["decision"]
    # schema v2 bump 단언.
    assert decision["schema"] == MERGE_DECISION_SCHEMA == "real_merge.decision.v2", scenario
    # v2 신규 필드 2종 존재 단언.
    assert "allow_reason" in decision, scenario
    assert "snapshot_crossref" in decision, scenario

    # allow_reason — fixture expected 와 일치.
    assert decision["allow_reason"] == expected["decision_allow_reason"], scenario

    # snapshot_crossref 필드 자체가 항상 dict 로 직렬화 되어 있어야 함.
    sc = decision["snapshot_crossref"]
    assert isinstance(sc, dict), scenario
    # 기대 crossref 의 핵심 필드 일치 (보수적 — 일부 필드는 substring 비교).
    exp_sc = expected["crossref"]
    assert sc["pr_match"] is exp_sc["pr_match"], scenario
    assert sc["sha_match"] is exp_sc["sha_match"], scenario
    assert sc["snapshot_present"] is exp_sc["snapshot_present"], scenario

    # chair_report_required boolean 일치.
    assert (
        bool(out.get("chair_report_required", False))
        == bool(expected["real_merge_chair_report_required"])
    ), scenario

    # NO_OP_RESULTS 멤버는 actually_executed=False.
    if out["result_enum"] in NO_OP_RESULTS:
        assert decision["actually_executed"] is False, scenario


def test_pass_candidate_records_allow_reason_token(tmp_path):
    """fixture_in_snapshot_pass_candidate — forbidden file in snapshot →
    decision.allow_reason 가 verbatim token 으로 기록되어야 함."""
    evidence, _ = _load("fixture_in_snapshot_pass_candidate")
    clock = _kst_clock(evidence["frozen_now_kst"])
    out = real_merge_execute(
        merge_ready_result=evidence["merge_ready_result"],
        pr_identity=evidence["pr_identity"],
        gate_snapshot=evidence["gate_snapshot"],
        dryrun_artifact=evidence["dryrun_artifact"],
        callback_envelope=evidence["callback_envelope"],
        chair_authorization=evidence["chair_authorization"],
        activation_flag=True,
        subprocess_runner=None,  # inert path
        canonical_root=str(tmp_path),
        changed_files=evidence["changed_files"],
        clock=clock,
    )
    assert out["result_enum"] == REAL_MERGE_DONE
    assert out["decision"]["allow_reason"] == ALLOW_REASON_SNAPSHOT_CROSSREF
    sc = out["decision"]["snapshot_crossref"]
    assert "tests/fixtures/snapshot_crossref/INDEX.md" in sc[
        "classification"
    ]["authorized_forbidden_hits"]


def test_mismatch_records_unauthorized_forbidden_hits(tmp_path):
    """fixture_in_snapshot_mismatch_no_op — snapshot 외 forbidden 파일은
    unauthorized_forbidden_hits 에 누적되고 result_enum 이 NO_OP_FORBIDDEN_PATH."""
    evidence, _ = _load("fixture_in_snapshot_mismatch_no_op")
    clock = _kst_clock(evidence["frozen_now_kst"])
    out = real_merge_execute(
        merge_ready_result=evidence["merge_ready_result"],
        pr_identity=evidence["pr_identity"],
        gate_snapshot=evidence["gate_snapshot"],
        dryrun_artifact=evidence["dryrun_artifact"],
        callback_envelope=evidence["callback_envelope"],
        chair_authorization=evidence["chair_authorization"],
        activation_flag=True,
        subprocess_runner=_runner_must_never_run,
        canonical_root=str(tmp_path),
        changed_files=evidence["changed_files"],
        clock=clock,
    )
    assert out["result_enum"] == NO_OP_FORBIDDEN_PATH
    sc = out["decision"]["snapshot_crossref"]
    assert (
        "tests/fixtures/snapshot_crossref/UNAUTHORIZED_FILE.md"
        in sc["classification"]["unauthorized_forbidden_hits"]
    )
    assert out["chair_report_required"] is True


def test_sanctioned_lock_separated_keeps_task_outputs(tmp_path):
    """sanctioned_lock_separated — `.tasks/locks/*` 가 task_outputs 와 별도
    sanctioned_artifacts 로 분리 기록되어야 함."""
    evidence, _ = _load("sanctioned_lock_separated")
    clock = _kst_clock(evidence["frozen_now_kst"])
    out = real_merge_execute(
        merge_ready_result=evidence["merge_ready_result"],
        pr_identity=evidence["pr_identity"],
        gate_snapshot=evidence["gate_snapshot"],
        dryrun_artifact=evidence["dryrun_artifact"],
        callback_envelope=evidence["callback_envelope"],
        chair_authorization=evidence["chair_authorization"],
        activation_flag=True,
        subprocess_runner=None,
        canonical_root=str(tmp_path),
        changed_files=evidence["changed_files"],
        clock=clock,
    )
    assert out["result_enum"] == REAL_MERGE_DONE
    clf = out["decision"]["snapshot_crossref"]["classification"]
    assert clf["sanctioned_artifacts"] == [".tasks/locks/task-2639-fx-004.lock"]
    assert ".tasks/locks/task-2639-fx-004.lock" not in clf["task_outputs"]


def test_v2_schema_decision_artifact_persists_snapshot_crossref(tmp_path):
    """REAL_MERGE_DONE 경로의 merge_decision.json 가 디스크 atomic write 후에도
    schema=v2 + allow_reason + snapshot_crossref 필드를 보존해야 함."""
    evidence, _ = _load("fixture_in_snapshot_pass_candidate")
    clock = _kst_clock(evidence["frozen_now_kst"])
    out = real_merge_execute(
        merge_ready_result=evidence["merge_ready_result"],
        pr_identity=evidence["pr_identity"],
        gate_snapshot=evidence["gate_snapshot"],
        dryrun_artifact=evidence["dryrun_artifact"],
        callback_envelope=evidence["callback_envelope"],
        chair_authorization=evidence["chair_authorization"],
        activation_flag=True,
        subprocess_runner=None,
        canonical_root=str(tmp_path),
        changed_files=evidence["changed_files"],
        clock=clock,
    )
    decision_path = Path(out["merge_decision_path"])
    assert decision_path.is_file()
    persisted = json.loads(decision_path.read_text(encoding="utf-8"))
    assert persisted["schema"] == "real_merge.decision.v2"
    assert persisted["allow_reason"] == ALLOW_REASON_SNAPSHOT_CROSSREF
    assert persisted["snapshot_crossref"]["pr_match"] is True
    assert persisted["snapshot_crossref"]["sha_match"] is True
    assert persisted["snapshot_crossref"]["snapshot_present"] is True


def test_no_chair_authorization_existing_flow_preserved(tmp_path):
    """ANCHOR-6: chair_authorization=None 일 때 기존 NO_OP_NO_AUTHORIZATION
    흐름이 보존되고 snapshot_crossref 는 dict 이지만 snapshot_present=False."""
    out = real_merge_execute(
        merge_ready_result={"verdict": "PASS"},
        pr_identity={"pr": 999, "head_sha": "deadbeef"},
        gate_snapshot=None,
        dryrun_artifact={
            "schema": "merge_ready.auto_merge_candidate.v1",
            "executor_action": "WOULD_MERGE",
        },
        callback_envelope=None,
        chair_authorization=None,
        activation_flag=True,
        subprocess_runner=_runner_must_never_run,
        canonical_root=str(tmp_path),
        changed_files=["memory/reports/x.md"],
    )
    assert out["result_enum"] == "NO_OP_NO_AUTHORIZATION"
    sc = out["decision"]["snapshot_crossref"]
    assert sc["snapshot_present"] is False
    assert sc["pr_match"] is False
    assert sc["sha_match"] is False