"""plate_select — 같은 location 복수 플레이트 VLM 선택 결정론 테스트.

E2E6 피드백 ⑤ (S18sh1 안방↔거실 오용). VLM 완성도는 육안/E2E — 여기는
후보 구성·판정 배선·fail-open 계약만 잠근다.
"""
from pathlib import Path
from typing import Any, Dict

import pytest

from app.modules.pipeline.plate_select import (
    location_of_bg_id,
    plate_candidates_by_location,
    resolve_prompt_version,
    select_plate_for_shot,
)


def test_location_of_bg_id_contract_format_only():
    assert location_of_bg_id("L04B02") == "L04"
    assert location_of_bg_id("L4B2") == "L4"
    assert location_of_bg_id("bg_urban_alley") is None
    assert location_of_bg_id("") is None


def test_candidates_by_location_ok_and_existing_only(tmp_path):
    p1 = tmp_path / "a.png"
    p1.write_bytes(b"x")
    p2 = tmp_path / "b.png"
    p2.write_bytes(b"x")
    groups = {
        "L04B01": {"status": "ok", "png_path": str(p1)},
        "L04B02": {"status": "ok", "png_path": str(p2)},
        "L04B03": {"status": "failed", "png_path": str(p2)},
        "L04B04": {"status": "ok", "png_path": str(tmp_path / "no.png")},
        "L05B01": {"status": "ok", "png_path": str(p1)},
        "bg_x": {"status": "ok", "png_path": str(p1)},  # 계약 형식 밖
    }
    out = plate_candidates_by_location(groups)
    assert [bid for bid, _ in out["L04"]] == ["L04B01", "L04B02"]
    assert [bid for bid, _ in out["L05"]] == ["L05B01"]
    assert "bg_x" not in str(out)


def _mk_cands(tmp_path, n=2, loc="L04"):
    out = []
    for i in range(1, n + 1):
        p = tmp_path / f"{loc}B0{i}.png"
        # png_part 가 실제 파일을 읽으므로 최소 PNG 헤더
        p.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 16)
        out.append((f"{loc}B0{i}", p))
    return out


def test_select_single_candidate_skips_judgment(tmp_path):
    cands = _mk_cands(tmp_path, 1)

    def boom(*a, **k):
        raise AssertionError("판정 호출되면 안 됨")

    bid, path, rec = select_plate_for_shot(
        shot_desc="SAMPLE", place_text="SAMPLE",
        assigned_bg_id="L04B01", candidates=cands,
        call_structured_fn=boom,
    )
    assert bid == "L04B01" and path == cands[0][1]
    assert "skipped" in rec


def test_select_switches_on_clear_choice(tmp_path):
    cands = _mk_cands(tmp_path, 2)
    seen: Dict[str, Any] = {}

    def fake(step_tag, system, parts, schema, **kw):
        seen["parts"] = parts
        return {"choice": "B", "confident": True, "reason_ko": "침대 근거"}

    bid, path, rec = select_plate_for_shot(
        shot_desc="침대에 눕는다", place_text="SAMPLE 방",
        assigned_bg_id="L04B01", candidates=cands,
        call_structured_fn=fake,
    )
    assert bid == "L04B02" and path == cands[1][1]
    assert rec["chosen"] == "L04B02" and rec["assigned"] == "L04B01"
    # CURRENTLY ASSIGNED 마킹이 배정 후보에 붙는다
    texts = [p["text"] for p in seen["parts"] if p.get("type") == "text"]
    assert any("Candidate A — CURRENTLY ASSIGNED" in t for t in texts)
    assert any(t == "Candidate B:" for t in texts)


def test_select_low_confidence_switch_blocked(tmp_path):
    """Codex BLOCKING-1: confident=false 전환은 배정 유지 fail-open."""
    cands = _mk_cands(tmp_path, 2)

    def fake(*a, **k):
        return {"choice": "B", "confident": False, "reason_ko": "애매"}

    bid, path, rec = select_plate_for_shot(
        shot_desc="SAMPLE", place_text="SAMPLE",
        assigned_bg_id="L04B01", candidates=cands,
        call_structured_fn=fake,
    )
    assert bid == "L04B01" and path == cands[0][1]
    assert rec["kept"] == "L04B01"
    assert rec["blocked_low_confidence"] == "L04B02"
    # 배정 자체를 low-confidence 로 유지하는 건 전환이 아님 — 통과
    def fake_same(*a, **k):
        return {"choice": "A", "confident": False, "reason_ko": ""}

    bid2, _p, rec2 = select_plate_for_shot(
        shot_desc="SAMPLE", place_text="SAMPLE",
        assigned_bg_id="L04B01", candidates=cands,
        call_structured_fn=fake_same,
    )
    assert bid2 == "L04B01" and rec2.get("chosen") == "L04B01"


def test_select_out_of_contract_choice_keeps_assigned(tmp_path):
    cands = _mk_cands(tmp_path, 2)

    def fake(*a, **k):
        return {"choice": "Z", "confident": True, "reason_ko": ""}

    bid, _path, rec = select_plate_for_shot(
        shot_desc="SAMPLE", place_text="SAMPLE",
        assigned_bg_id="L04B02", candidates=cands,
        call_structured_fn=fake,
    )
    assert bid == "L04B02"
    assert rec["kept"] == "L04B02"


def test_select_judge_error_keeps_assigned(tmp_path):
    cands = _mk_cands(tmp_path, 2)

    def fake(*a, **k):
        raise RuntimeError("SAMPLE transient")

    bid, _path, rec = select_plate_for_shot(
        shot_desc="SAMPLE", place_text="SAMPLE",
        assigned_bg_id="L04B01", candidates=cands,
        call_structured_fn=fake,
    )
    assert bid == "L04B01"
    assert "error" in rec


def test_select_assigned_outside_candidates_noop(tmp_path):
    cands = _mk_cands(tmp_path, 2)
    bid, _path, rec = select_plate_for_shot(
        shot_desc="SAMPLE", place_text="SAMPLE",
        assigned_bg_id="L99B01", candidates=cands,
        call_structured_fn=lambda *a, **k: {},
    )
    assert bid == "L99B01"
    assert "skipped" in rec


def test_pack_is_scenario_neutral():
    repo = Path(__file__).resolve().parents[3]
    ver = resolve_prompt_version("1")
    text = (repo / "prompts" / "_base" / "plate_select" / ver
            / "judge_system.md").read_text(encoding="utf-8").lower()
    for word in ("안방", "거실", "bedroom", "living room", "rooftop"):
        assert word not in text
    n = " ".join(text.split())
    assert "currently assigned" in n
    assert "only switch when the evidence is clear" in n


def test_unknown_version_raises():
    with pytest.raises(ValueError):
        resolve_prompt_version("99")


# ── R1 (2026-07-16): 플레이트 권위 선행 고정 — run_plate_authority ──────


def _authority_env(tmp_path, n=2):
    from app.modules.pipeline.plate_select import (
        plate_candidates_by_location,
    )

    groups = {}
    for i in range(1, n + 1):
        p = tmp_path / f"L04B0{i}.png"
        p.write_bytes(f"SAMPLE{i}".encode())
        groups[f"L04B0{i}"] = {"status": "ok", "png_path": str(p)}
    cands_by_loc = plate_candidates_by_location(groups)
    plate_map = {"4_1": Path(groups["L04B01"]["png_path"])}
    assign = {"4_1": "L04B01"}
    return plate_map, assign, cands_by_loc


def _judge_choice_b(calls):
    def fake(_mod, _sys, _parts, _schema, **_k):
        calls.append(1)
        return {"choice": "B", "confident": True, "reason_ko": "SAMPLE"}
    return fake


def test_authority_judges_and_persists_sidecar(tmp_path):
    from app.modules.pipeline.plate_select import run_plate_authority

    plate_map, assign, cands = _authority_env(tmp_path)
    calls: list = []
    sidecar: Dict[str, Any] = {}
    out = run_plate_authority(
        tags=["S4sh1"], plate_map=plate_map, assign_by_key=assign,
        cands_by_loc=cands, shot_desc_by_tag={"S4sh1": "SAMPLE desc"},
        place_text_by_scene={4: "SAMPLE place"}, sidecar=sidecar,
        call_structured_fn=_judge_choice_b(calls),
        judge_model="SAMPLE-judge",
    )
    assert calls == [1]
    entry = out["S4sh1"]
    assert entry["assigned"] == "L04B01"
    assert entry["chosen"] == "L04B02"
    assert entry["plate_path"].endswith("L04B02.png")
    assert entry["record"]["chosen"] == "L04B02"
    # 사이드카 영속 — 지문 포함
    assert sidecar["S4sh1"]["fingerprint"] == entry["fingerprint"]


def test_authority_fingerprint_reuse_zero_calls(tmp_path):
    from app.modules.pipeline.plate_select import run_plate_authority

    plate_map, assign, cands = _authority_env(tmp_path)
    calls: list = []
    sidecar: Dict[str, Any] = {}
    kwargs = dict(
        tags=["S4sh1"], plate_map=plate_map, assign_by_key=assign,
        cands_by_loc=cands, shot_desc_by_tag={"S4sh1": "SAMPLE desc"},
        place_text_by_scene={4: "SAMPLE place"}, sidecar=sidecar,
        call_structured_fn=_judge_choice_b(calls),
        judge_model="SAMPLE-judge",
    )
    run_plate_authority(**kwargs)
    out2 = run_plate_authority(**kwargs)
    assert calls == [1]  # 2회차 판정 0콜
    assert out2["S4sh1"]["reused"] is True
    assert out2["S4sh1"]["chosen"] == "L04B02"
    # force = 재판정
    run_plate_authority(**{**kwargs, "force": True})
    assert calls == [1, 1]


def test_authority_refingerprints_on_plate_bytes_change(tmp_path):
    from app.modules.pipeline.plate_select import run_plate_authority

    plate_map, assign, cands = _authority_env(tmp_path)
    calls: list = []
    sidecar: Dict[str, Any] = {}
    kwargs = dict(
        tags=["S4sh1"], plate_map=plate_map, assign_by_key=assign,
        cands_by_loc=cands, shot_desc_by_tag={"S4sh1": "SAMPLE desc"},
        place_text_by_scene={4: "SAMPLE place"}, sidecar=sidecar,
        call_structured_fn=_judge_choice_b(calls),
        judge_model="SAMPLE-judge",
    )
    run_plate_authority(**kwargs)
    # 후보 플레이트 재생성(bytes 변경) → 지문 불일치 → 재판정
    for bid, p in cands["L04"]:
        if bid == "L04B02":
            p.write_bytes(b"SAMPLE-regen")
    run_plate_authority(**kwargs)
    assert calls == [1, 1]


def test_authority_skips_ineligible_shots(tmp_path):
    from app.modules.pipeline.plate_select import run_plate_authority

    plate_map, assign, cands = _authority_env(tmp_path)
    calls: list = []
    out = run_plate_authority(
        tags=["S4sh1", "S5sh1", "S6sh1", "S7sh1"],
        # S5sh1=플레이트 없음 / S6sh1=배정 없음 / S7sh1=후보 1장 location
        plate_map={**plate_map, "6_1": plate_map["4_1"],
                   "7_1": plate_map["4_1"]},
        assign_by_key={**assign, "7_1": "L05B01"},
        cands_by_loc={**cands, "L05": cands["L04"][:1]},
        shot_desc_by_tag={t: "SAMPLE" for t in
                          ("S4sh1", "S5sh1", "S6sh1", "S7sh1")},
        place_text_by_scene={4: "SAMPLE"}, sidecar={},
        call_structured_fn=_judge_choice_b(calls),
        judge_model="SAMPLE-judge",
    )
    assert set(out) == {"S4sh1"}
    assert calls == [1]


def test_authority_skips_assigned_outside_candidates(tmp_path):
    """배치 리뷰 HIGH-4: 배정이 후보 밖이면 entry 미생성(배정 유지) —
    select 의 Path('') 반환이 '.' 로 영속되던 결함 차단."""
    from app.modules.pipeline.plate_select import run_plate_authority

    plate_map, assign, cands = _authority_env(tmp_path)
    calls: list = []
    out = run_plate_authority(
        tags=["S4sh1"], plate_map=plate_map,
        assign_by_key={"4_1": "L04B99"},  # 후보 밖 배정
        cands_by_loc=cands,
        shot_desc_by_tag={"S4sh1": "SAMPLE"},
        place_text_by_scene={4: "SAMPLE"}, sidecar={},
        call_structured_fn=_judge_choice_b(calls),
        judge_model="SAMPLE-judge",
    )
    assert out == {} and calls == []


def test_authority_rejects_non_file_choice_and_stale_sidecar(tmp_path):
    """HIGH-4: 선택 결과·사이드카 plate_path 는 is_file 만 신뢰 —
    디렉토리('.')·소실 파일은 영속/재사용 금지."""
    from app.modules.pipeline.plate_select import run_plate_authority

    plate_map, assign, cands = _authority_env(tmp_path)
    calls: list = []
    sidecar: Dict[str, Any] = {}
    kwargs = dict(
        tags=["S4sh1"], plate_map=plate_map, assign_by_key=assign,
        cands_by_loc=cands, shot_desc_by_tag={"S4sh1": "SAMPLE"},
        place_text_by_scene={4: "SAMPLE"}, sidecar=sidecar,
        call_structured_fn=_judge_choice_b(calls),
        judge_model="SAMPLE-judge",
    )
    out = run_plate_authority(**kwargs)
    assert out["S4sh1"]["plate_path"].endswith("L04B02.png")
    # 선택된 플레이트 파일 소실 → 사이드카 재사용 금지(재판정)
    for bid, pth in cands["L04"]:
        if bid == "L04B02":
            pth.unlink()
            pth.write_bytes(b"SAMPLE-new")  # 재생성(bytes 상이=재판정)
    run_plate_authority(**kwargs)
    assert calls == [1, 1]


def test_candidates_reject_directories(tmp_path):
    """재리뷰 NARROW-3: 디렉토리는 후보 제외 (is_file) — fingerprint
    read 예외·stale dir 통과 차단."""
    from app.modules.pipeline.plate_select import (
        plate_candidates_by_location,
    )

    d = tmp_path / "L04B01.png"
    d.mkdir()  # 디렉토리로 위장
    f = tmp_path / "L04B02.png"
    f.write_bytes(b"SAMPLE")
    groups = {
        "L04B01": {"status": "ok", "png_path": str(d)},
        "L04B02": {"status": "ok", "png_path": str(f)},
    }
    out = plate_candidates_by_location(groups)
    assert [bid for bid, _ in out["L04"]] == ["L04B02"]