"""P8 I-1 — registered_pose_guide_service 단위 (gate / cache / hash / underlay).

생성(gpt-image-2 edit)은 generate_floor_plan_image 를 monkeypatch 로 대체해 실제
API 0. 최종 시각 품질은 라이브 canary 갤러리(육안)로 판단 — 여기선 결정론적 계약
(gate/캐시 무결성/no-white-bg degrade/프롬프트 토큰)만 잠근다.
"""
from __future__ import annotations

from io import BytesIO
from pathlib import Path

import pytest

from app.services import registered_pose_guide_service as rpg


def _png_bytes(color=(120, 120, 120), size=(64, 48)) -> bytes:
    from PIL import Image
    buf = BytesIO()
    Image.new("RGB", size, color).save(buf, format="PNG")
    return buf.getvalue()


_ANCHOR = {
    "character_short_id": "C04",
    "subject_state": "dead",
    "shared_state_contract": "the immobilized body sits slumped against the bed edge, "
                             "head fallen forward, one hand resting on the mattress.",
    "locked_elements": [{"description": "the right hand rests on the bed surface"}],
    "per_shot_visible_focus": {"13": "tight insert on the resting hand"},
}


def test_make_underlay_variants_produce_valid_png():
    bg = _png_bytes()
    for variant in ("original", "blur_gray", "edge"):
        out = rpg.make_registration_underlay(bg, variant)
        assert out[:8] == b"\x89PNG\r\n\x1a\n", variant
        assert len(out) >= 1024 or variant != "original"  # original 은 full
    with pytest.raises(ValueError):
        rpg.make_registration_underlay(bg, "nope")


def test_prompt_carries_contract_no_scenario_tokens():
    p = rpg.build_registered_pose_guide_prompt(_ANCHOR)
    assert "slumped against the bed edge" in p          # 계약이 SOT
    assert "right hand rests on the bed surface" in p   # locked desc
    assert "MANNEQUIN" in p and "NO blood" in p          # gore-free 등록
    assert "C04" not in p                                # entity id 누출 0
    # Fix2 (Codex BLOCKING2): guide 는 상태중립 — death 의미("lifeless")를 generic
    # immobilized 경로에 하드코딩하지 않는다 (unconscious/severely_injured 도 같은 가이드).
    assert "lifeless" not in p.lower()
    assert "limp and fully passive" in p


def test_gate_no_contract_returns_none(tmp_path):
    png, diag = rpg.build_registered_pose_guide(
        group_id="g1", subject_anchor={"shared_state_contract": ""},
        env_bg_bytes=_png_bytes(), bg_key="L04B05",
        cache_dir=tmp_path, openai_client=object(),
    )
    assert png is None and diag["status"] == "skipped"
    assert diag["reason"] == "empty_shared_state_contract"


def test_gate_no_bg_returns_none_no_white_fallback(tmp_path):
    """★ bg plate 부재 → None (white-bg fallback 없음 = no visual guide degrade)."""
    png, diag = rpg.build_registered_pose_guide(
        group_id="g1", subject_anchor=_ANCHOR,
        env_bg_bytes=None, bg_key="",
        cache_dir=tmp_path, openai_client=object(),
    )
    assert png is None and diag["status"] == "skipped"
    assert diag["reason"] == "no_environment_bg_plate"
    # 디스크에 어떤 가이드 파일도 만들지 않는다.
    assert not list(tmp_path.glob("*.png"))


def test_generation_writes_cache_and_status(tmp_path, monkeypatch):
    fake_png = _png_bytes((10, 20, 30), (1024, 1024))

    def _fake_gen(*, prompt, openai_client, ref_paths, model, size, **kw):
        # underlay temp 가 실제 파일로 존재해야 함(서비스가 씀)
        assert ref_paths and Path(ref_paths[0]).exists()
        # Phase C: build_registered_pose_guide 가 capture_role/capture_extra_metadata 를
        # 추가 전달 — fake 는 흡수(**kw). 실 capture 경로는 test_phase_c_c2 가 검증.
        return fake_png
    monkeypatch.setattr(
        "app.modules.pipeline.location_floor_plan.generate_floor_plan_image", _fake_gen)

    png, diag = rpg.build_registered_pose_guide(
        group_id="vca-s12-c04-immobilized", subject_anchor=_ANCHOR,
        env_bg_bytes=_png_bytes(), bg_key="L04B05",
        cache_dir=tmp_path, openai_client=object(),
    )
    assert png == fake_png and diag["status"] == "generated"
    cached = list(tmp_path.glob("vca-s12-c04-immobilized_*.png"))
    assert len(cached) == 1
    # underlay temp 는 정리됐다
    assert not list(tmp_path.glob(".*underlay*"))


def test_cache_hit_skips_regeneration(tmp_path, monkeypatch):
    calls = {"n": 0}

    def _fake_gen(**kw):
        calls["n"] += 1
        return _png_bytes((1, 2, 3), (1024, 1024))
    monkeypatch.setattr(
        "app.modules.pipeline.location_floor_plan.generate_floor_plan_image", _fake_gen)
    kw = dict(group_id="g", subject_anchor=_ANCHOR, env_bg_bytes=_png_bytes(),
              bg_key="L04B05", cache_dir=tmp_path, openai_client=object())
    p1, d1 = rpg.build_registered_pose_guide(**kw)
    p2, d2 = rpg.build_registered_pose_guide(**kw)
    assert d1["status"] == "generated" and d2["status"] == "cache_hit"
    assert calls["n"] == 1 and p1 == p2


def test_concurrent_same_group_generates_once(tmp_path, monkeypatch):
    """★ Fix1 (Codex BLOCKING1): 같은 (group, hash) 를 동시에 cache-miss 로 본 두
    worker 가 각자 stochastic guide 를 만들지 않는다 — per-path lock 으로 생성 1회 +
    나머지는 cache 재사용, 두 caller 가 동일 bytes 를 받는다."""
    import threading
    import time

    calls = {"n": 0}
    cguard = threading.Lock()

    def _fake_gen(**kw):
        with cguard:
            calls["n"] += 1
            n = calls["n"]
        time.sleep(0.05)  # lock 보유 중 두 번째 thread 가 도착하도록 overlap 유도
        return _png_bytes((n, n, n), (1024, 1024))  # 호출마다 다른 bytes (분기 감지용)
    monkeypatch.setattr(
        "app.modules.pipeline.location_floor_plan.generate_floor_plan_image", _fake_gen)

    kw = dict(group_id="g", subject_anchor=_ANCHOR, env_bg_bytes=_png_bytes(),
              bg_key="L04B05", cache_dir=tmp_path, openai_client=object())
    results: dict = {}
    start = threading.Event()

    def _worker(idx):
        start.wait()
        png, diag = rpg.build_registered_pose_guide(**kw)
        results[idx] = (png, diag["status"])

    threads = [threading.Thread(target=_worker, args=(i,)) for i in range(2)]
    for t in threads:
        t.start()
    start.set()  # 두 thread 거의 동시 진입
    for t in threads:
        t.join()

    assert calls["n"] == 1                              # 생성 단 1회
    assert results[0][0] == results[1][0]               # 두 caller 동일 bytes
    assert len(list(tmp_path.glob("g_*.png"))) == 1     # 캐시 파일 1개
    statuses = {results[0][1], results[1][1]}
    assert "generated" in statuses                      # 하나는 생성
    assert statuses & {"cache_hit", "cache_hit_after_lock"}  # 다른 하나는 재사용
    # temp 파일 잔존 없음 (pid+uuid 격리 후 정리)
    assert not list(tmp_path.glob(".*underlay*"))
    assert not list(tmp_path.glob(".*tmp*"))


def test_hash_sensitive_to_anchor_change(tmp_path):
    h1 = rpg._guide_hash(group_id="g", bg_key="L04B05", subject_anchor=_ANCHOR,
                         model="gpt-image-2", variant="original")
    changed = dict(_ANCHOR, shared_state_contract="lying flat on the floor")
    h2 = rpg._guide_hash(group_id="g", bg_key="L04B05", subject_anchor=changed,
                         model="gpt-image-2", variant="original")
    h3 = rpg._guide_hash(group_id="g", bg_key="OTHER", subject_anchor=_ANCHOR,
                         model="gpt-image-2", variant="original")
    assert h1 != h2 and h1 != h3   # 계약/배경 바뀌면 mismatch → 재생성