"""B (2026-07-02) — outdoor 관련샷 prev 완성프레임 의무첨부. primary_location 구조 fallback(short_id→UUID 역매핑) + required 진단 매트릭스 + consumer 로더 flag 게이트 + persistence scene_prev_frame resolve. 전부 결정론 — 프레임 품질/실발동은 canary+육안. """ from __future__ import annotations import json from unittest.mock import MagicMock from app.core.config import settings from app.core.steps.outdoor_site_layout_step import load_outdoor_prev_frame_context from app.modules.pipeline.outdoor_site_layout_plan import ( build_outdoor_prev_frame_diag, primary_location_uuid_by_scene, resolve_outdoor_history_fallback_uuid, ) from app.services.scene_persistence_service import ScenePersistenceService # ─────────────────── primary_location_uuid_by_scene ─────────────────── def _lookup(): return { "uuid-l11": {"entity_type": "location", "short_id": "L11"}, "uuid-l05": {"entity_type": "location", "short_id": "L05"}, "uuid-c01": {"entity_type": "character", "short_id": "C01"}, } def test_uuid_map_outdoor_only_and_reverse_join(): out = primary_location_uuid_by_scene( {15: "L11", 12: "L05", 3: "L99"}, # L05=indoor(집합 밖), L99=역매핑 없음 {"L11", "L99"}, _lookup(), ) assert out == {15: "uuid-l11"} def test_uuid_map_ambiguous_short_id_excluded(): lk = _lookup() lk["uuid-l11-dup"] = {"entity_type": "location", "short_id": "L11"} out = primary_location_uuid_by_scene({15: "L11"}, {"L11"}, lk) assert out == {} # 같은 short_id 2+ → 모호 — 제외 (결정론) def test_history_fallback_requires_history_entry(): m = {15: "uuid-l11"} hist = {"uuid-l11": (b"png", {"id": "still-1"})} assert resolve_outdoor_history_fallback_uuid( 15, primary_loc_uuid_by_scene=m, location_scene_history=hist) == "uuid-l11" assert resolve_outdoor_history_fallback_uuid( 15, primary_loc_uuid_by_scene=m, location_scene_history={}) is None assert resolve_outdoor_history_fallback_uuid( 9, primary_loc_uuid_by_scene=m, location_scene_history=hist) is None # ─────────────────── required 진단 매트릭스 ─────────────────── def _diag(**over): base = dict( scene_index=15, shot_index=5, outdoor_loc_sids={"L11"}, primary_location_by_scene={15: "L11"}, ve_location_sids=set(), chain_bg_attached=False, ref_usage="", has_dep=True, has_same_scene_prior=False, resolved=False, bytes_source_kind="none", source_still_id=None, ) base.update(over) return build_outdoor_prev_frame_diag(**base) def test_diag_missing_case_full_fields(): d = _diag() assert d is not None assert d["lane"] == "outdoor" assert d["previous_frame_required"] is True assert d["resolved"] is False assert d["reason_if_missing"] == "previous_frame_required_missing" assert d["required_signals"] == {"dep": True, "same_scene_prior": False} assert d["source_selection"] is None def test_diag_resolved_records_source(): d = _diag(resolved=True, bytes_source_kind="location_history", source_still_id="still-7", has_dep=False, has_same_scene_prior=True) assert d is not None assert d["resolved"] is True and d["reason_if_missing"] is None assert d["source_selection"] == "location_history" assert d["source_still_id"] == "still-7" def test_diag_not_outdoor_none(): assert _diag(primary_location_by_scene={15: "L05"}) is None # L05 ∉ outdoor # frame-visible VE 가 outdoor 면 primary 무관하게 대상 assert _diag(primary_location_by_scene={}, ve_location_sids={"L11"}) is not None def test_diag_excluded_cases_none(): assert _diag(chain_bg_attached=True) is None # plate 실재 assert _diag(ref_usage="zoom_in_detail") is None # zoom 별도 정책 assert _diag(has_dep=False, has_same_scene_prior=False) is None # 신호 없음 # close 는 close×ref_usage matrix 계약이 prev-shot bg 합성 금지 — 의무 아님 assert _diag(close_framing=True) is None # ─────────────────── consumer 로더 ─────────────────── def _write_cp(base, step, data): d = base / step d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text( json.dumps({"status": "completed", "data": data}), encoding="utf-8") def test_loader_flag_off_empty(monkeypatch): monkeypatch.setattr(settings, "outdoor_prev_frame_required_enabled", False, raising=False) assert load_outdoor_prev_frame_context("p", "e") == {} def test_loader_reads_structural_fields(monkeypatch, tmp_path): monkeypatch.setattr(settings, "outdoor_prev_frame_required_enabled", True, raising=False) monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False) base = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" _write_cp(base, "background_classify", {"building_groups": [ {"members": [ {"loc_id": "L11", "is_indoor": False}, {"loc_id": "L05", "is_indoor": True}, ]}, ]}) _write_cp(base, "scene_director", {"scenes": [ {"scene_index": 15, "primary_location": "L11"}, {"scene_index": 12, "primary_location": "L05"}, ]}) ctx = load_outdoor_prev_frame_context("p1", "e1") assert ctx["outdoor_loc_sids"] == {"L11"} assert ctx["primary_location_by_scene"] == {15: "L11", 12: "L05"} def test_loader_no_outdoor_locs_empty(monkeypatch, tmp_path): monkeypatch.setattr(settings, "outdoor_prev_frame_required_enabled", True, raising=False) monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False) base = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" _write_cp(base, "background_classify", {"building_groups": [ {"members": [{"loc_id": "L05", "is_indoor": True}]}, ]}) assert load_outdoor_prev_frame_context("p1", "e1") == {} # ─────────────────── persistence scene_prev_frame resolve ─────────────────── def _svc_with_first_returns(first_returns): instance = ScenePersistenceService.__new__(ScenePersistenceService) db = MagicMock() q = MagicMock() db.query.return_value = q q.filter.return_value = q q.order_by.return_value = q q.first.side_effect = first_returns instance._db = db instance._project_id = "p1" return instance def test_resolve_scene_prev_frame_role(): svc = _svc_with_first_returns([("prev-scene-uuid",)]) ids, refs, remaining = svc._resolve_prev_frame_asset_ids([{ "role": "previous_shot_continuity", "label": "BG prev", "pipeline_role": "scene_prev_frame", "source_still_id": "still-prev", }], "e1") assert ids == ["prev-scene-uuid"] assert refs[0]["pipeline_role"] == "scene_prev_frame" assert refs[0]["resolved_from_unresolved"] is True assert remaining == []