"""G3.2 Phase 7 Task 21 — consumer wiring integration tests.

본 테스트는 G3.2 Phase 1~6 의 producer→loader→consumer 흐름이 한 시나리오 안에서
정확히 wiring 되어 있는지 검증한다. 단위 테스트가 각 helper 를 검증하는 것과
달리, integration 테스트는 다음 8 시나리오를 end-to-end (mocked LLM) 로 cover:

  1. close framing 3-block skip (guide / camera / owned 모두 skip).
  2. non-close inject (owned block 정상 prepend).
  3. sentinel drift detection (cp 직접 편집 → verify_completion drift 신호).
  4. contract_violation status (judge 위반 → variation 마킹 + no retry).
  5. analysis-only block (R5-I3) — bg-on + scene_detail 직전 dep 미완료 → check_gate 차단.
  6. bg-mode-off → entire owned wiring no-op (judge 미호출, sentinel trivial).
  7. schema isolation — 옛 v4 cp / partial v5 cp → loader fail-fast.
  8. 옛 v4 / partial v5 cp scenarios (#7 변형).

LLM 호출 (`run_owned_judge` 및 `call_structured`) 은 모두 mock — 실제 API 미사용.
"""
from __future__ import annotations

from typing import Any, Callable, Dict, List, Tuple
from unittest.mock import MagicMock, patch

import pytest

from app.core.errors import AppError
from app.core.steps._owned_helpers import (
    OWNED_VALIDATOR_CLOSE_SKIP,
    OWNED_VALIDATOR_FULL,
    build_owned_sentinel,
    compute_camera_direction_hash,
    compute_owned_hash,
    compute_t2i_prompt_hash,
)
from app.core.steps.detail_steps import (
    SceneDetailStep,
    _build_phase2_prepend_blocks,
    _user_edited_owned_contract_violated,
)
from app.core.steps.scene_context_loader import SceneContextLoader


# ──────────────────────────────────────────────────────────────────────────
# Helpers
# ──────────────────────────────────────────────────────────────────────────


def _make_scene_detail_step(tmp_path, monkeypatch, *, bg_mode="off"):
    """SceneDetailStep skeleton — DB / cp 의존성 우회."""
    monkeypatch.setattr(
        "app.core.config.settings.projects_dir", str(tmp_path), raising=False,
    )
    monkeypatch.setattr(
        "app.core.config.settings.background_mode", bg_mode, raising=False,
    )
    step = SceneDetailStep.__new__(SceneDetailStep)
    step.project_id = "P1"
    step.episode_id = "E1"
    step.project_config = {}
    return step


class _FakeRunner:
    """SceneContextLoader 가 요구하는 최소 runner skeleton."""

    project_id = "P1"
    episode_id = "E1"

    def __init__(self, cps: Dict[str, Any]):
        self._cps = cps
        self.db = MagicMock()
        self.db.execute.return_value.fetchall.return_value = []
        self.project_config = None

    def _load_prev_checkpoint(self, step_id: str):
        return self._cps.get(step_id)

    def build_opik_metadata(self):
        return {}


# ──────────────────────────────────────────────────────────────────────────
# Scenario 1: close framing 3-block skip
# ──────────────────────────────────────────────────────────────────────────


class TestCloseFramingThreeBlockSkip:
    """close framing bool flag → guide / camera_meta / owned 모두 skip.

    framing_scale enum SOT v1 (2026-05-15): `_build_phase2_prepend_blocks` signature
    가 `camera_direction: str` → `is_close_framing: bool` 변경. production caller
    가 helper (`get_framing_scale_or_raise`) 로 staging.framing_scale enum read
    후 `== FRAMING_CLOSE` 결과 전달. 본 test 는 helper 결과 단일 (True) case
    검증 — parametrize camera_direction 의 5 keyword case 는 의미 X.
    """

    def test_close_framing_skips_all_three_blocks(self):
        """3 토글 모두 enabled + 3 블록 데이터 모두 존재 + is_close_framing=True → 모두 skip."""
        out = _build_phase2_prepend_blocks(
            si=1, shi=1,
            essence_by_shot={(1, 1): ["essence_a"]},
            chain_bg_guide_by_shot={(1, 1): "do not redraw walls"},
            chain_bg_camera_meta_by_shot={(1, 1): {
                "camera_position": "near 1",
                "camera_height": "1.6m",
                "lens_hint": "35mm",
            }},
            chain_bg_owned_by_shot={(1, 1): ["door", "window"]},
            shot_essence_enabled=True,
            chain_bg_guide_enabled=True,
            chain_bg_camera_meta_enabled=True,
            is_close_framing=True,
        )
        # essence 는 close framing 무관 (Rule J 와 별개) — 출력에 포함됨.
        assert "essence_a" in out
        # G3.2 contract: chain_bg 3 블록 모두 close framing skip.
        assert "do not redraw walls" not in out
        assert "near 1" not in out
        assert "door" not in out
        assert "window" not in out


# ──────────────────────────────────────────────────────────────────────────
# Scenario 2: non-close inject — English canonical owned 정상 prepend
# ──────────────────────────────────────────────────────────────────────────


class TestNonCloseOwnedInject:
    """non-close framing + owned 1+ entries → owned block prepend."""

    def test_owned_block_appears_with_english_canonical_items(self):
        out = _build_phase2_prepend_blocks(
            si=2, shi=3,
            essence_by_shot={},
            chain_bg_guide_by_shot={},
            chain_bg_owned_by_shot={(2, 3): ["TV", "door", "window"]},
            shot_essence_enabled=False,
            chain_bg_guide_enabled=False,
            is_close_framing=False,  # non-close (framing_scale enum SOT v1)
        )
        # owned block header 가 포함.
        assert "chain_bg에 이미 그려진 객체" in out
        # 각 ASCII canonical entry 가 bullet 으로 출력.
        assert "- TV" in out
        assert "- door" in out
        assert "- window" in out

    def test_no_inject_when_owned_empty(self):
        """owned [] 면 block 자체 미포함."""
        out = _build_phase2_prepend_blocks(
            si=1, shi=1,
            essence_by_shot={},
            chain_bg_guide_by_shot={},
            chain_bg_owned_by_shot={(1, 1): []},
            shot_essence_enabled=False,
            chain_bg_guide_enabled=False,
            is_close_framing=False,  # non-close (framing_scale enum SOT v1)
        )
        assert "chain_bg에 이미 그려진 객체" not in out


# ──────────────────────────────────────────────────────────────────────────
# Scenario 3: sentinel drift detection
# ──────────────────────────────────────────────────────────────────────────


class TestSentinelDriftDetection:
    """variation 의 t2i_prompt / owned / camera_direction 변경 시 drift 검출.

    `_user_edited_owned_contract_violated` 가 reuse path 의 1차 가드. 동일 4-way
    drift 검사가 ``verify_completion`` 에도 적용 (round 5 BLOCKING 3, 4).
    """

    def _base_variation(self):
        return {
            "t2i_prompt": "wide shot of door and window",
            "owned_validation": build_owned_sentinel(
                owned=["door", "window"],
                camera_direction="WIDE",
                t2i_prompt="wide shot of door and window",
                is_close_framing=False,
                violations=[],
                owned_object_usage=[  # C2 v1: owned token 전부 absent echo
                    {"owned_token": "door", "usage_kind": "absent", "source_phrase": ""},
                    {"owned_token": "window", "usage_kind": "absent", "source_phrase": ""},
                ],
            ),
        }

    def test_clean_baseline_passes(self):
        var = self._base_variation()
        violated = _user_edited_owned_contract_violated(
            var,
            expected_owned=["door", "window"],
            expected_camera_direction="WIDE",
            is_close=False,
        )
        assert violated is False

    def test_t2i_prompt_edit_flagged(self):
        var = self._base_variation()
        var["t2i_prompt"] = "wide shot of door, window, and lamp"
        violated = _user_edited_owned_contract_violated(
            var,
            expected_owned=["door", "window"],
            expected_camera_direction="WIDE",
            is_close=False,
        )
        assert violated is True

    def test_owned_drift_flagged(self):
        var = self._base_variation()
        violated = _user_edited_owned_contract_violated(
            var,
            expected_owned=["door"],  # window dropped
            expected_camera_direction="WIDE",
            is_close=False,
        )
        assert violated is True

    def test_camera_direction_drift_flagged(self):
        var = self._base_variation()
        violated = _user_edited_owned_contract_violated(
            var,
            expected_owned=["door", "window"],
            expected_camera_direction="MEDIUM",  # was WIDE
            is_close=False,
        )
        assert violated is True

    def test_validator_type_drift_flagged_when_close_changes(self):
        """sentinel 은 full validator 인데 현재 close framing 으로 평가 → drift."""
        var = self._base_variation()
        violated = _user_edited_owned_contract_violated(
            var,
            expected_owned=["door", "window"],
            expected_camera_direction="WIDE",
            is_close=True,  # caller 가 close framing 으로 평가
        )
        assert violated is True

    def test_missing_sentinel_flagged(self):
        var = {"t2i_prompt": "x"}
        violated = _user_edited_owned_contract_violated(
            var,
            expected_owned=[],
            expected_camera_direction="",
            is_close=False,
        )
        assert violated is True


# ──────────────────────────────────────────────────────────────────────────
# Scenario 4: contract_violation status
# ──────────────────────────────────────────────────────────────────────────


class TestContractViolationStatus:
    """judge 가 violations 반환 → sentinel.violations 보존 + caller status 마킹.

    `run_owned_judge` 만 mock — 실제 LLM 호출 없음.
    """

    def test_judge_returns_violations_marks_sentinel(self):
        """judge violations → build_owned_sentinel 의 violations 필드 보존."""
        from app.core.steps._owned_judge import run_owned_judge

        fake_call = MagicMock(return_value={
            "violations": [
                {
                    "owned_object": "door",
                    "violating_phrase": "open door visible",
                    "reason": "owned 객체 새로 묘사",
                },
            ],
        })
        violations = run_owned_judge(
            t2i_prompt="open door visible behind the actor",
            owned=["door", "window"],
            owned_object_usage=[  # C2 v1 (W3): judge v4 cross-check echo input
                {"owned_token": "door", "usage_kind": "redraw",
                 "source_phrase": "open door visible"},
                {"owned_token": "window", "usage_kind": "absent", "source_phrase": ""},
            ],
            camera_direction="WIDE",
            call_structured_fn=fake_call,
        )
        assert len(violations) == 1
        assert violations[0]["owned_object"] == "door"

        sentinel = build_owned_sentinel(
            owned=["door", "window"],
            camera_direction="WIDE",
            t2i_prompt="open door visible behind the actor",
            is_close_framing=False,
            violations=violations,
            owned_object_usage=[  # C2 v1: owned token 전부 absent echo
                {"owned_token": "door", "usage_kind": "absent", "source_phrase": ""},
                {"owned_token": "window", "usage_kind": "absent", "source_phrase": ""},
            ],
        )
        # violations 보존 + validator=FULL.
        assert sentinel["violations"] == violations
        assert sentinel["validator"] == OWNED_VALIDATOR_FULL

    def test_judge_no_violations_returns_empty_list(self):
        from app.core.steps._owned_judge import run_owned_judge

        fake_call = MagicMock(return_value={"violations": []})
        violations = run_owned_judge(
            t2i_prompt="actor stands in foreground",
            owned=["door"],
            owned_object_usage=[  # C2 v1 (W3): judge v4 cross-check echo input
                {"owned_token": "door", "usage_kind": "absent", "source_phrase": ""},
            ],
            camera_direction="WIDE",
            call_structured_fn=fake_call,
        )
        assert violations == []

    def test_judge_skipped_when_owned_empty(self):
        """owned [] → LLM 호출 자체 skip (cost 절감)."""
        from app.core.steps._owned_judge import run_owned_judge

        fake_call = MagicMock()
        violations = run_owned_judge(
            t2i_prompt="anything",
            owned=[],
            owned_object_usage=[],  # C2 v1 (W3): owned [] → echo [] (full coverage)
            camera_direction="WIDE",
            call_structured_fn=fake_call,
        )
        assert violations == []
        fake_call.assert_not_called()


# ──────────────────────────────────────────────────────────────────────────
# Scenario 5: analysis-only block (R5-I3) — gate 차단
# ──────────────────────────────────────────────────────────────────────────


class TestAnalysisOnlyBlockGate:
    """bg-on + scene_detail 의 background_prompt dep 미완료 → check_gate 차단.

    StepRunner.check_gate 가 manifest 의 depends_on 을 순회하며
    ``_get_step_run(dep)`` 가 None 이거나 status != completed/not_applicable
    이면 ``gate.blocked`` raise.

    회귀 가드 — manifest 에서 background_prompt dep 가 누락되거나 partial cascade
    contract (allow_partial_downstream=False) 가 깨지면 본 테스트 fail.
    """

    def test_scene_detail_depends_on_background_prompt(self):
        """manifest 검증 — background_prompt 는 scene_detail 의 dep."""
        from app.core.step_manifest import get_depends_on

        deps = get_depends_on("scene_detail")
        assert "background_prompt" in deps, (
            "G3.2: scene_detail manifest 에 background_prompt dep 누락 — "
            "owned list 를 unmet dep 상태에서 소비할 위험."
        )

    def test_check_gate_raises_when_background_prompt_missing(self, tmp_path, monkeypatch):
        """background_prompt step_run 가 DB 에 없으면 check_gate raise.

        Wave 6 MINOR fix: dead ``step.manifest = ...`` 제거 — check_gate 는
        STEP_MANIFEST module-level (get_depends_on(self.step_id)) 를 읽지
        self.manifest 를 보지 않음. test_scene_detail_depends_on_background_prompt
        가 manifest 의 dep 존재를 별도로 검증.
        """
        step = _make_scene_detail_step(tmp_path, monkeypatch, bg_mode="on")
        step.step_id = "scene_detail"
        # 모든 dep 의 step_run 를 None 반환 → blocked.
        step.db = MagicMock()
        step.db.execute.return_value.fetchone.return_value = None

        with pytest.raises(AppError) as exc_info:
            step.check_gate()
        assert exc_info.value.code == "gate.blocked"

    def test_check_gate_passes_when_background_prompt_completed(self, tmp_path, monkeypatch):
        """background_prompt step_run status='completed' 면 check_gate 통과.

        Wave 6 MINOR fix: dead ``step.manifest = ...`` 제거 (위와 동일 이유).
        """
        from types import SimpleNamespace

        step = _make_scene_detail_step(tmp_path, monkeypatch, bg_mode="on")
        step.step_id = "scene_detail"
        step.db = MagicMock()
        # Block B T0 (plan v2.1.3): _get_step_run 가 dict 반환 — 내부 fetchone()
        # row 는 attribute access 지원 row-like (status / run_id / 등).
        step.db.execute.return_value.fetchone.return_value = SimpleNamespace(
            status="completed",
            run_id="r-test",
            started_at=None,
            completed_count=1,
            applicable_count=1,
            recovery_count=0,
            updated_at=None,
            # 락 소유자 신원 (alembic 010) — 구 행은 전부 None 이다.
            owner_host=None,
            owner_pid=None,
            owner_boot_id=None,
            heartbeat_at=None,
            cancel_requested_at=None,
        )
        # 예외 없이 통과.
        step.check_gate()

    def test_scene_detail_allow_partial_downstream_false(self):
        """G3.2 Round 2 #6: contract_violation partial cascade stop."""
        from app.core.step_manifest import get_manifest_dict

        m = get_manifest_dict("scene_detail")
        assert m.get("allow_partial_downstream") is False, (
            "G3.2: scene_detail allow_partial_downstream False 가 깨짐 — "
            "owned judge violations 시 downstream cascade 가 무방비로 진행."
        )


# ──────────────────────────────────────────────────────────────────────────
# Scenario 6: bg-mode-off → entire owned wiring no-op
# ──────────────────────────────────────────────────────────────────────────


class TestBgModeOffNoOp:
    """bg-mode=off → loader 가 빈 매핑 / judge 호출 안 됨 / verify 통과."""

    def test_loader_returns_empty_when_bg_off(self, tmp_path, monkeypatch):
        """bg-off + bp_cp 부재 → loader 빈 dict (no raise)."""
        monkeypatch.setattr(
            "app.core.config.settings.background_mode", "off", raising=False,
        )
        runner = _FakeRunner({})  # bp_cp 부재
        loader = SceneContextLoader(runner)
        assert loader._load_chain_bg_owned_by_shot() == {}

    def test_loader_returns_empty_when_bg_off_with_old_v4_cp(self, tmp_path, monkeypatch):
        """bg-off + 옛 v4 cp 잔존 → loader 통과 (off 에선 schema 검사 skip)."""
        monkeypatch.setattr(
            "app.core.config.settings.background_mode", "off", raising=False,
        )
        runner = _FakeRunner({"background_prompt": {
            "schema_version": 1,
            "data": {"backgrounds": {"bg1": {"status": "ok"}}},
        }})
        loader = SceneContextLoader(runner)
        assert loader._load_chain_bg_owned_by_shot() == {}

    def test_judge_skipped_when_owned_empty_in_bg_off_path(self):
        """bg-off → owned 항상 [] → judge 호출 자체 skip."""
        from app.core.steps._owned_judge import run_owned_judge

        fake_call = MagicMock()
        violations = run_owned_judge(
            t2i_prompt="any",
            owned=[],  # bg-off path 결과
            owned_object_usage=[],  # C2 v1 (W3): owned [] → echo [] (full coverage)
            camera_direction="WIDE",
            call_structured_fn=fake_call,
        )
        assert violations == []
        fake_call.assert_not_called()

    def test_verify_completion_clean_with_trivial_sentinel(self, tmp_path, monkeypatch):
        """bg-off path 에서 producer 가 생성하는 trivial sentinel 은 verify 통과."""
        from app.core.integrity_report import CompletionReport

        step = _make_scene_detail_step(tmp_path, monkeypatch, bg_mode="off")
        # framing_scale enum SOT v1 (2026-05-15): verify_completion 의 drift check
        # 가 staging_map 안 framing_scale enum 을 helper read. patch
        # SceneContextLoader._load_staging_map → {"1_1": {"framing_scale":
        # "medium", ...}} (non-close intent, trivial sentinel.validator=full 정합).
        monkeypatch.setattr(
            SceneContextLoader,
            "_load_staging_map",
            lambda self: {"1_1": {"framing_scale": "medium", "camera_direction": ""}},
        )
        # producer 가 _analyze_one 에서 부착하는 trivial sentinel 시뮬레이션.
        trivial_sentinel = build_owned_sentinel(
            owned=[], camera_direction="",
            t2i_prompt="x",
            is_close_framing=False, violations=[],
            owned_object_usage=[],
        )
        step._last_execute_result = {
            "data": {
                "scenes": [
                    {"scene_index": 1, "_shot_index": 1,
                     "t2i_variations": [{
                         "t2i_prompt": "x",
                         "owned_validation": trivial_sentinel,
                     }]},
                ]
            }
        }
        report = step.verify_completion()
        assert isinstance(report, CompletionReport)
        assert report.is_complete is True
        assert report.severity == "clean"


# ──────────────────────────────────────────────────────────────────────────
# Scenario 7 + 8: schema isolation — 옛 v4 / partial v5 cp loader fail-fast
# ──────────────────────────────────────────────────────────────────────────


class TestSchemaIsolationAndPartialCp:
    """bg-on path 의 loader 가 schema_version<2 / partial cp 차단 (fail-fast)."""

    @pytest.fixture
    def bg_on(self, monkeypatch):
        monkeypatch.setattr(
            "app.core.config.settings.background_mode", "on", raising=False,
        )

    def test_old_v4_cp_raises(self, bg_on):
        """schema_version=1 (옛 v4) → contract_violation."""
        runner = _FakeRunner({"background_prompt": {
            "schema_version": 1,
            "data": {"backgrounds": {"bg1": {"status": "ok"}}},
        }})
        loader = SceneContextLoader(runner)
        with pytest.raises(AppError) as exc_info:
            loader._load_chain_bg_owned_by_shot()
        assert exc_info.value.code == "step.contract_violation"

    def test_partial_v5_cp_missing_owned_raises(self, bg_on):
        """schema=2 인데 ok bg entry 에 owned 부재 → contract_violation."""
        runner = _FakeRunner({"background_prompt": {
            "schema_version": 2,
            "data": {"backgrounds": {
                "bg_ok": {
                    "status": "ok",
                    "objects_owned_by_background": ["door"],
                    "spec": {"applies_to_shots": ["S1_Shot1"]},
                },
                "bg_partial": {"status": "ok"},  # owned 누락
            }},
        }})
        loader = SceneContextLoader(runner)
        with pytest.raises(AppError) as exc_info:
            loader._load_chain_bg_owned_by_shot()
        assert exc_info.value.code == "step.contract_violation"

    def test_partial_v5_cp_non_ascii_owned_raises(self, bg_on):
        """schema=2 + owned 에 non-ASCII (한글) 포함 → round 7 BLOCKING 1 raise."""
        runner = _FakeRunner({"background_prompt": {
            "schema_version": 2,
            "data": {"backgrounds": {
                "bg1": {
                    "status": "ok",
                    "objects_owned_by_background": ["문"],  # 한국어
                    "spec": {"applies_to_shots": ["S1_Shot1"]},
                },
            }},
        }})
        loader = SceneContextLoader(runner)
        with pytest.raises(AppError) as exc_info:
            loader._load_chain_bg_owned_by_shot()
        assert exc_info.value.code == "step.contract_violation"

    def test_bg_on_no_cp_raises_silent_block(self, bg_on):
        """bg-on + cp None → silent {} 차단 (round 4 BLOCKING 1)."""
        runner = _FakeRunner({})
        loader = SceneContextLoader(runner)
        with pytest.raises(AppError):
            loader._load_chain_bg_owned_by_shot()

    def test_valid_v5_cp_passes(self, bg_on):
        """schema=2 + 모든 ok entry 에 owned 1+ entries → loader 정상."""
        runner = _FakeRunner({"background_prompt": {
            "schema_version": 2,
            "data": {"backgrounds": {
                "bg1": {
                    "status": "ok",
                    "objects_owned_by_background": ["door", "window"],
                    "spec": {"applies_to_shots": ["S1_Shot1", "S1_Shot2"]},
                },
            }},
        }})
        loader = SceneContextLoader(runner)
        result = loader._load_chain_bg_owned_by_shot()
        assert result == {(1, 1): ["door", "window"], (1, 2): ["door", "window"]}