"""Group 1 #3 (visual_pipeline_contracts_plan) — consumer step verify_completion 회귀 가드.

Producer step (RefImageGenStep / CompositeImageGenStep / chain_bg / floor_plan) 들은
이미 verify_completion 갖춤. 본 PR 가 consumer 측 빈자리 채움:
- SceneImagePipelineStep
- SceneConsistencyStep
- SceneDetailStep

각 step 의 verify_completion 이 산출물 누락 시 partial/missing 으로 정확히 분류해야
silent fallback (status=completed 인데 산출물 0/품질 X) 회귀 차단.

시나리오 의존성 0 — placeholder ID 만 사용.
"""
from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest


# ─────────────────────────────────────────────
# SceneImagePipelineStep.verify_completion
# ─────────────────────────────────────────────

def _make_scene_image_step(monkeypatch, expected_stills=2, found_with_disk=2, tmp_path=None):
    from app.core.steps.image_steps import SceneImagePipelineStep

    step = SceneImagePipelineStep.__new__(SceneImagePipelineStep)
    step.project_id = "P1"
    step.episode_id = "E1"

    db = MagicMock()
    step.db = db

    # SceneStill query — selected ids (still_id list).
    selected_ids = [f"still_{i}" for i in range(expected_stills)]
    selected_q = MagicMock()
    selected_q.filter.return_value = selected_q
    selected_q.all.return_value = [(sid,) for sid in selected_ids]
    # ImageAsset query — primary scene assets.
    asset_q = MagicMock()
    asset_q.filter.return_value = asset_q
    fake_assets = []
    for i, sid in enumerate(selected_ids[:found_with_disk]):
        fake_a = MagicMock()
        fake_a.still_id = sid
        if tmp_path is not None:
            p = tmp_path / f"img_{i}.png"
            p.write_bytes(b"PNG")
            fake_a.file_path = str(p)
        else:
            fake_a.file_path = f"missing/{i}.png"
        fake_assets.append(fake_a)
    asset_q.all.return_value = fake_assets

    # db.query routing — by model.
    from app.models.project import ImageAsset, SceneStill
    def _query(model):
        if model is ImageAsset:
            return asset_q
        if model is SceneStill or model == SceneStill.id:
            return selected_q
        # SceneStill.id query (col):
        return selected_q
    db.query.side_effect = _query
    return step


def test_scene_image_pipeline_clean_when_all_disk_found(tmp_path, monkeypatch):
    step = _make_scene_image_step(monkeypatch, expected_stills=3, found_with_disk=3, tmp_path=tmp_path)
    report = step.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"
    assert report.metadata["expected"] == 3
    assert report.metadata["found"] == 3


def test_scene_image_pipeline_rows_는_행_개수여야_한다(tmp_path, monkeypatch):
    """★`rows` 는 **정수 행 수**다 — ORM 목록이 아니다 (2026-08-29 Codex BLOCK).

    ref 쪽 verify 를 정본 helper 로 옮기면서 `rows=len(rows)` 를 `rows=rows` 로
    바꾼 치환이 **이 자리까지** 번졌다. 여기 `rows` 는 `ImageAsset` ORM 객체
    목록이라, 그대로 두면 count 진단 계약이 깨지고 JSON 기록에 ORM 목록이
    흘러간다. 「행은 있는데 파일이 없다」를 가르는 값이므로 수도 맞아야 한다.
    """
    # 행 3개 중 파일이 있는 것은 1개 — rows 3 · found 1 이어야 한다.
    step = _make_scene_image_step(monkeypatch, expected_stills=3,
                                  found_with_disk=3, tmp_path=None)
    report = step.verify_completion()
    rows = report.metadata["rows"]
    assert isinstance(rows, int), f"rows 가 정수가 아니다: {type(rows).__name__}"
    assert rows == 3, report.metadata
    assert report.metadata["found"] == 0, report.metadata

    step2 = _make_scene_image_step(monkeypatch, expected_stills=3,
                                   found_with_disk=3, tmp_path=tmp_path)
    r2 = step2.verify_completion()
    assert isinstance(r2.metadata["rows"], int)
    assert r2.metadata["rows"] == 3 and r2.metadata["found"] == 3, r2.metadata


def test_scene_image_pipeline_partial_when_some_missing(tmp_path, monkeypatch):
    step = _make_scene_image_step(monkeypatch, expected_stills=3, found_with_disk=1, tmp_path=tmp_path)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert report.metadata["expected"] == 3
    assert report.metadata["found"] == 1


def test_scene_image_pipeline_missing_when_none_found(tmp_path, monkeypatch):
    step = _make_scene_image_step(monkeypatch, expected_stills=2, found_with_disk=0, tmp_path=tmp_path)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "missing"
    assert report.metadata["found"] == 0


def test_scene_image_pipeline_clean_when_no_expected(tmp_path, monkeypatch):
    """selected stills 가 0 인 episode (image gen scope 비어있음) → clean."""
    step = _make_scene_image_step(monkeypatch, expected_stills=0, found_with_disk=0, tmp_path=tmp_path)
    report = step.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"
    assert report.metadata["expected"] == 0


def test_scene_image_pipeline_silent_fallback_caught(tmp_path, monkeypatch):
    """status=completed 인데 disk file 없는 silent fallback 회귀 가드."""
    # disk 안 만들면 fp.exists() False — found 가 expected 보다 적음.
    step = _make_scene_image_step(monkeypatch, expected_stills=2, found_with_disk=2, tmp_path=None)
    report = step.verify_completion()
    assert report.is_complete is False  # disk 없음 → not complete (silent 안 됨).
    assert report.severity == "missing"


# ─────────────────────────────────────────────
# SceneConsistencyStep.verify_completion
# ─────────────────────────────────────────────

def _make_scene_consistency_step(tmp_path, monkeypatch, cp_data=None, last_result=None):
    """fresh-run path: last_result 직접 set (cp 보다 우선).
    resume path: cp_data 만 set (last_result=None).
    """
    from app.core.steps.scene_consistency_step import SceneConsistencyStep

    step = SceneConsistencyStep.__new__(SceneConsistencyStep)
    step.project_id = "P1"
    step.episode_id = "E1"
    if last_result is not None:
        step._last_execute_result = last_result

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    cp_dir = tmp_path / "P1" / "checkpoints" / "episodes" / "E1" / "scene_consistency"
    if cp_data is not None:
        cp_dir.mkdir(parents=True, exist_ok=True)
        (cp_dir / "manifest.json").write_text(json.dumps(cp_data), encoding="utf-8")
    return step


def test_scene_consistency_missing_when_no_cp(tmp_path, monkeypatch):
    step = _make_scene_consistency_step(tmp_path, monkeypatch, cp_data=None)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "missing"


def test_scene_consistency_clean_when_all_succeeded(tmp_path, monkeypatch):
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok", "fixed_elements": [{"id": "e"}]},
                {"scene_index": 2, "analysis_summary": "단일 샷 — 교차 일관성 분석 불필요", "fixed_elements": []},
            ]
        }
    }
    step = _make_scene_consistency_step(tmp_path, monkeypatch, cp_data=cp_data)
    report = step.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"


def test_scene_consistency_partial_when_some_failed(tmp_path, monkeypatch):
    """analysis_summary='분석 실패' 가 있으면 partial — silent OK 안 됨."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok", "fixed_elements": []},
                {"scene_index": 2, "analysis_summary": "분석 실패", "fixed_elements": []},
            ]
        }
    }
    step = _make_scene_consistency_step(tmp_path, monkeypatch, cp_data=cp_data)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert 2 in report.metadata["failed_scenes"]


def test_scene_consistency_silent_fallback_caught(tmp_path, monkeypatch):
    """모두 '분석 실패' (3-tier 모두 fail) → 옛 swallow 패턴 차단."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "분석 실패", "fixed_elements": []},
                {"scene_index": 2, "analysis_summary": "분석 실패", "fixed_elements": []},
            ]
        }
    }
    step = _make_scene_consistency_step(tmp_path, monkeypatch, cp_data=cp_data)
    report = step.verify_completion()
    assert report.is_complete is False
    assert "2/2 scene(s)" in report.missing[0]


# ─────────────────────────────────────────────
# SceneDetailStep.verify_completion
# ─────────────────────────────────────────────

def _make_scene_detail_step(tmp_path, monkeypatch, cp_data=None, last_result=None):
    from app.core.steps.detail_steps import SceneDetailStep
    from app.core.steps.scene_context_loader import SceneContextLoader

    step = SceneDetailStep.__new__(SceneDetailStep)
    step.project_id = "P1"
    step.episode_id = "E1"
    if last_result is not None:
        step._last_execute_result = last_result

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    # G3.2: bg-mode off — 본 fixture 는 t2i presence / fresh-run last_result 검증용 이며
    # background_prompt cp 부재 시 fail-fast 가 발동되면 안 됨.
    monkeypatch.setattr(
        "app.core.config.settings.background_mode", "off", raising=False,
    )
    # framing_scale enum SOT v1 (2026-05-15): verify_completion 의 drift check
    # 가 staging_map 안 framing_scale enum 을 helper read. non-close intent
    # (trivial sentinel = validator full / is_close_framing False) 정합.
    monkeypatch.setattr(
        SceneContextLoader,
        "_load_staging_map",
        lambda self: {
            "1_1": {"framing_scale": "medium", "camera_direction": ""},
            "2_1": {"framing_scale": "medium", "camera_direction": ""},
        },
    )
    cp_dir = tmp_path / "P1" / "checkpoints" / "episodes" / "E1" / "scene_detail"
    if cp_data is not None:
        cp_dir.mkdir(parents=True, exist_ok=True)
        (cp_dir / "manifest.json").write_text(json.dumps(cp_data), encoding="utf-8")
    return step


def _owned_sentinel_for(t2i_prompt: str) -> dict:
    """G3.2: bg-off 일 때 verify_completion 에 부착되는 trivial sentinel.

    expected_owned=[] / camera_direction='' / non-close framing 가정. 본 helper
    는 verify_completion 의 4-way drift 검증을 통과하는 valid sentinel 을 생성.
    """
    from app.core.steps._owned_helpers import build_owned_sentinel

    return build_owned_sentinel(
        owned=[],
        camera_direction="",
        t2i_prompt=t2i_prompt,
        is_close_framing=False,
        violations=[],
        owned_object_usage=[],
    )


def test_scene_detail_missing_when_no_cp(tmp_path, monkeypatch):
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=None)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "missing"


def test_scene_detail_missing_when_empty_scenes(tmp_path, monkeypatch):
    cp_data = {"data": {"scenes": []}}
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=cp_data)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "missing"


def test_scene_detail_clean_when_all_have_t2i(tmp_path, monkeypatch):
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a", "t2i_prompt": "x",
                     "owned_validation": _owned_sentinel_for("x"),
                 }]},
                {"scene_index": 2, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a", "t2i_prompt": "y",
                     "owned_validation": _owned_sentinel_for("y"),
                 }]},
            ]
        }
    }
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=cp_data)
    report = step.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"


def test_scene_detail_partial_when_some_empty_t2i(tmp_path, monkeypatch):
    """t2i_variations=[] 가 있으면 partial — silent empty 차단."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "_shot_index": 1,
                 "t2i_variations": [{"variant_label": "a"}]},
                {"scene_index": 2, "_shot_index": 1,
                 "t2i_variations": []},  # empty.
            ]
        }
    }
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=cp_data)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert (2, 1) in report.metadata["failed_indices"]


def test_scene_detail_missing_when_bg_on_loader_raises(tmp_path, monkeypatch):
    """G3.2 Wave 6 iter3: bg-on + loader AppError → severity='missing'.

    failed_indices < total 이라 평소라면 partial 인 시나리오에서, loader
    contract violation 이 발생하면 severity 가 missing 으로 escalate. loader
    violation 은 전체 batch 무효화 신호이므로 partial 보다 강한 marker 필요.
    """
    from app.core.errors import AppError
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a", "t2i_prompt": "x",
                     "owned_validation": _owned_sentinel_for("x"),
                 }]},
                {"scene_index": 2, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a", "t2i_prompt": "y",
                     "owned_validation": _owned_sentinel_for("y"),
                 }]},
            ]
        }
    }
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=cp_data)
    # bg-on 으로 override (helper 가 off 로 set 한 것을 덮어쓰기).
    monkeypatch.setattr(
        "app.core.config.settings.background_mode", "on", raising=False,
    )
    # loader 가 AppError 를 raise 하도록 stub — bg-on 분기 진입 + AppError catch.
    def _raise_loader_error(self):
        raise AppError(
            code="step.contract_violation",
            message="background_prompt cp missing — stubbed for test",
        )
    monkeypatch.setattr(
        "app.core.steps.scene_context_loader.SceneContextLoader."
        "_load_chain_bg_owned_by_shot",
        _raise_loader_error,
    )
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "missing", (
        f"loader_violations 가 partial → missing 으로 escalate 해야 함 "
        f"(got {report.severity})"
    )
    # missing message 에 loader contract violation 명시 surface.
    assert any("loader contract violation" in m for m in report.missing), (
        f"loader violation message 가 missing list 에 포함돼야 함: {report.missing}"
    )


def test_scene_detail_missing_when_all_empty_t2i(tmp_path, monkeypatch):
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "_shot_index": 1, "t2i_variations": []},
                {"scene_index": 2, "_shot_index": 1, "t2i_variations": []},
            ]
        }
    }
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=cp_data)
    report = step.verify_completion()
    assert report.is_complete is False
    # 모두 fail 면 missing (partial 보다 강한 분류).
    assert report.severity == "missing"


# ─────────────────────────────────────────────
# Codex review Critical: fresh run path — _last_execute_result 우선
# ─────────────────────────────────────────────

def test_scene_consistency_uses_last_execute_result_when_no_cp(tmp_path, monkeypatch):
    """Codex review Critical (Group 1 #3): fresh run 에서 cp 가 아직 안 만들어졌어도
    step_runner 가 보존한 _last_execute_result 로 verify 가능."""
    last_result = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok", "fixed_elements": [{"x": 1}]},
            ]
        }
    }
    # cp_data=None → cp 부재 (fresh run 시뮬레이션). last_result 만 set.
    step = _make_scene_consistency_step(tmp_path, monkeypatch, cp_data=None, last_result=last_result)
    report = step.verify_completion()
    # cp 부재여도 last_result 로 정확히 clean 분류.
    assert report.is_complete is True
    assert report.severity == "clean"


def test_scene_consistency_last_result_takes_precedence_over_cp(tmp_path, monkeypatch):
    """_last_execute_result 가 cp 보다 우선 (현재 run 결과가 정확)."""
    cp_data = {
        "data": {
            "scenes": [{"scene_index": 1, "analysis_summary": "ok", "fixed_elements": []}],
        }
    }
    # last_result 가 cp 와 다른 결과 — last_result 가 win.
    last_result = {
        "data": {
            "scenes": [{"scene_index": 1, "analysis_summary": "분석 실패", "fixed_elements": []}],
        }
    }
    step = _make_scene_consistency_step(tmp_path, monkeypatch, cp_data=cp_data, last_result=last_result)
    report = step.verify_completion()
    # last_result 의 partial 가 win — cp 의 ok 안 됨.
    assert report.is_complete is False
    assert report.severity == "partial"


def test_scene_detail_uses_last_execute_result_when_no_cp(tmp_path, monkeypatch):
    """fresh run 에서 cp 부재여도 last_result 로 verify."""
    last_result = {
        "data": {
            "scenes": [
                {"scene_index": 1, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a",
                     "owned_validation": _owned_sentinel_for(""),
                 }]},
            ]
        }
    }
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=None, last_result=last_result)
    report = step.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"


def test_scene_consistency_startswith_failure_string(tmp_path, monkeypatch):
    """Claude review IMPORTANT (Group 1 #3): _execute resume 은 startswith 매칭.
    verify 도 동일하게 startswith — '분석 실패: rate limit' 같은 prefix 형태도 catch."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok", "fixed_elements": []},
                {"scene_index": 2, "analysis_summary": "분석 실패: rate limit exceeded",
                 "fixed_elements": []},
            ]
        }
    }
    step = _make_scene_consistency_step(tmp_path, monkeypatch, cp_data=cp_data)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert 2 in report.metadata["failed_scenes"]


def test_scene_detail_metadata_keys_consistent_across_branches(tmp_path, monkeypatch):
    """Claude review IMPORTANT: empty/clean/partial 모든 분기의 metadata key 일관."""
    # empty cp.
    step1 = _make_scene_detail_step(tmp_path, monkeypatch, cp_data={"data": {"scenes": []}})
    r1 = step1.verify_completion()
    assert "total_results" in r1.metadata
    assert "failed_count" in r1.metadata
    assert "failed_indices" in r1.metadata
    # clean.
    step2 = _make_scene_detail_step(tmp_path, monkeypatch, cp_data={
        "data": {"scenes": [{"scene_index": 1, "_shot_index": 1,
                              "t2i_variations": [{"variant_label": "a"}]}]}
    })
    r2 = step2.verify_completion()
    assert "total_results" in r2.metadata
    assert "failed_count" in r2.metadata
    assert "failed_indices" in r2.metadata
    # partial — 다른 fixture 쓰기 (이미 위 test 에서 검증).


# ─────────────────────────────────────────────
# I5 cleanup (G4.1 follow-up): card-check bypass warning rate-limit
# ─────────────────────────────────────────────


def test_scene_detail_card_check_bypass_warning_emitted_once_per_step(
    tmp_path, monkeypatch, caplog,
):
    """I5 cleanup: schema_version 이 v15 (=6) 인 cp 로 verify_completion 호출 시
    `_card_check_active=False` → step 당 logger.warning 정확히 1회만 발사.
    이전 구현은 N scene 마다 redundant warning emit.
    """
    cp_data = {
        "schema_version": 6,  # 옛 v15 — manifest=7 과 mismatch.
        "data": {
            "scenes": [
                {"scene_index": 1, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a", "t2i_prompt": "x",
                     "owned_validation": _owned_sentinel_for("x"),
                 }]},
                {"scene_index": 2, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a", "t2i_prompt": "y",
                     "owned_validation": _owned_sentinel_for("y"),
                 }]},
                {"scene_index": 3, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a", "t2i_prompt": "z",
                     "owned_validation": _owned_sentinel_for("z"),
                 }]},
            ],
        },
    }
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=cp_data)
    with caplog.at_level("WARNING", logger="app.core.steps.detail_steps"):
        step.verify_completion()
    bypass_warnings = [
        rec for rec in caplog.records
        if "card check bypass" in rec.getMessage()
    ]
    assert len(bypass_warnings) == 1, (
        f"I5: bypass warning must fire exactly once per step, "
        f"got {len(bypass_warnings)} for 3 scene results"
    )
    msg = bypass_warnings[0].getMessage()
    # 한 번에 affected count + version mismatch 둘 다 보고.
    assert "schema_version=6" in msg, msg
    assert "3 scene results affected" in msg, msg


def test_scene_detail_card_check_bypass_no_warning_when_schema_matches(
    tmp_path, monkeypatch, caplog,
):
    """I5 cleanup: schema_version 이 current 와 일치하면 _card_check_active=True
    → bypass warning 미발사.
    """
    from app.core.steps.detail_steps import SCENE_DETAIL_SCHEMA_VERSION
    cp_data = {
        "schema_version": SCENE_DETAIL_SCHEMA_VERSION,  # v16 current.
        "data": {
            "scenes": [
                {"scene_index": 1, "_shot_index": 1,
                 "t2i_variations": [{
                     "variant_label": "a", "t2i_prompt": "x",
                     "owned_validation": _owned_sentinel_for("x"),
                 }]},
            ],
        },
    }
    step = _make_scene_detail_step(tmp_path, monkeypatch, cp_data=cp_data)
    with caplog.at_level("WARNING", logger="app.core.steps.detail_steps"):
        step.verify_completion()
    bypass_warnings = [
        rec for rec in caplog.records
        if "card check bypass" in rec.getMessage()
    ]
    assert bypass_warnings == [], (
        f"current-schema cp 는 bypass warning 발사 안 해야 함 (got {bypass_warnings!r})"
    )