"""Area #6 v1 canary — mutation=0 invariant + diagnostic emit + W1 atomicity. 5 test (spec §5.3 + §8 W3 명시): 1. test_no_mutation_invariant — entity_applied=0 / scene_applied=0 / indices=[] 2. test_sentinel_branch_not_invoked — refresh/assert helper 호출 0 (defensive) 3. test_diagnostic_emit_shape — 5 core fields strict 4. test_regeneration_required_count_matches_detected — 1:1 mapping 5. test_w1_atomicity_invariant — v6 schema active + production reader 정합 """ import json import pathlib import re from unittest.mock import MagicMock, patch import pytest from app.modules.pipeline.t2i_review import run_t2i_review @pytest.fixture(autouse=True) def _scene_branch_needs_legacy(monkeypatch): """씬 갈래를 재는 파일이라 legacy 를 못박는다 (2026-08-27). v1 기본값(`t2i_review_skip_scene_on_v1`)이 그 갈래를 건너뛰므로 설정을 안 잡으면 `test_diagnostic_emit_shape` 와 `test_regeneration_required_count_matches_detected` 가 「씬 진단 0」 으로 무너진다. 그 계약(5 core fields · 검출 1:1 매핑)은 legacy 경로에서 여전히 유효하다 — 자리를 legacy 로 고정해 살린다. """ from app.core.config import settings monkeypatch.setattr(settings, "still_recipe_mode", "off", raising=False) def _scene_detail_data(): return { "scenes": [ { "scene_index": 1, "_shot_index": 0, "t2i_variations": [ { "t2i_prompt": "BEFORE prompt unchanged", "render_prompt_card_hash": "CARD_HASH_PRE", }, ], "owned_validation": {}, }, ], } def _shot_extract(): return { "scenes": [ { "scene_index": 1, "shots": [{"shot_index": 0, "description": "shot desc"}], }, ], } def _entity_t2i_data(): return { "characters": [ {"short_id": "C01", "name": "주인공", "t2i_prompt": "Asian man, suit"}, ], "locations": [], "props": [], "completed": {}, } def _stub_full_emit(scene_id: str = "S1_L0_V0", entity_id: str = "C01"): """call_structured stub — entity batch + scene batch 모두 v1 5 core fields emit.""" def _fake(**kwargs): user_prompt = kwargs.get("user_prompt", "") if "엔티티" in user_prompt: return { "results": [ { "short_id": entity_id, "has_issues": True, "issues": [{ "id": entity_id, "scope": "entity", "issue_code": "missing_ethnicity", "action": "regen_required", "hint_diagnostic": "보통명사 인물 — 인종 형용사 추가 권장", "short_id": entity_id, }], }, ], } # scene path return { "results": [ { "item_id": scene_id, "scene_index": 1, "var_index": 0, "has_issues": True, "issues": [{ "id": scene_id, "scope": "scene", "issue_code": "awkward_translation", "action": "regen_required", "hint_diagnostic": "한국 문화권 명사 — 원어 표기 권장", "item_id": scene_id, }], }, ], } return _fake def test_no_mutation_invariant(): """canary 1: entity_applied == 0 / scene_applied == 0 / scene_applied_indices == [].""" scene_detail = _scene_detail_data() pre_t2i_prompt = scene_detail["scenes"][0]["t2i_variations"][0]["t2i_prompt"] with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_stub_full_emit(), ): result = run_t2i_review( entity_t2i_data=_entity_t2i_data(), scene_detail_data=scene_detail, entity_merge_data={"characters": [], "locations": [], "props": []}, entity_detail_data={"entity_details": {}}, shot_extract_data=_shot_extract(), shot_staging_data={}, vwr_data={"t2i_context": ""}, ) # canary contract (hard-coded invariants) assert result["entity_applied"] == 0, f"entity_applied != 0: {result['entity_applied']}" assert result["scene_applied"] == 0, f"scene_applied != 0: {result['scene_applied']}" assert result["scene_applied_indices"] == [], ( f"scene_applied_indices != []: {result['scene_applied_indices']}" ) # t2i_prompt 변경 0 post_t2i_prompt = scene_detail["scenes"][0]["t2i_variations"][0]["t2i_prompt"] assert pre_t2i_prompt == post_t2i_prompt, ( f"t2i_prompt mutated: pre={pre_t2i_prompt!r} post={post_t2i_prompt!r}" ) def test_sentinel_branch_not_invoked(): """canary 2: run_t2i_review-level helper boundary invariant. run_t2i_review 본체는 mutation=0 이므로 _refresh_scene_detail_sentinels / _assert_card_hash_unchanged 를 직접 호출하지 않는다. _execute 의 `if scene_applied > 0:` branch unreached 검증은 별도 canary carry. """ from app.core.steps import t2i_review_step scene_detail = _scene_detail_data() refresh_spy = MagicMock(wraps=t2i_review_step._refresh_scene_detail_sentinels) assert_spy = MagicMock(wraps=t2i_review_step._assert_card_hash_unchanged) save_spy = MagicMock() with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_stub_full_emit(), ), patch.object(t2i_review_step, "_refresh_scene_detail_sentinels", refresh_spy), \ patch.object(t2i_review_step, "_assert_card_hash_unchanged", assert_spy): result = run_t2i_review( entity_t2i_data=_entity_t2i_data(), scene_detail_data=scene_detail, entity_merge_data={"characters": [], "locations": [], "props": []}, entity_detail_data={"entity_details": {}}, shot_extract_data=_shot_extract(), shot_staging_data={}, vwr_data={"t2i_context": ""}, ) # run_t2i_review 본체는 helper 호출 안 함 (caller t2i_review_step._execute 가 # if branch 안에서만 호출). mutation=0 시 branch unreached → spy 호출 0. # (직접 _execute 안 부르지만 helper invocation site 가 if branch 안임을 검증.) assert refresh_spy.call_count == 0, ( f"_refresh_scene_detail_sentinels 호출 {refresh_spy.call_count}: defensive branch reached" ) assert assert_spy.call_count == 0, ( f"_assert_card_hash_unchanged 호출 {assert_spy.call_count}: defensive branch reached" ) # mutation=0 invariant 재확인 (test isolation) assert result["scene_applied"] == 0 assert result["scene_applied_indices"] == [] def test_diagnostic_emit_shape(): """canary 3: diagnostics list 의 각 entry 가 5 core fields strict 형식.""" scene_detail = _scene_detail_data() with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_stub_full_emit(), ): result = run_t2i_review( entity_t2i_data=_entity_t2i_data(), scene_detail_data=scene_detail, entity_merge_data={"characters": [], "locations": [], "props": []}, entity_detail_data={"entity_details": {}}, shot_extract_data=_shot_extract(), shot_staging_data={}, vwr_data={"t2i_context": ""}, ) assert len(result["diagnostics"]) == 2, ( f"diagnostics count != 2 (entity 1 + scene 1): {len(result['diagnostics'])}" ) core_fields = {"id", "scope", "issue_code", "action", "hint_diagnostic"} scopes_seen = set() for diag in result["diagnostics"]: # 5 core fields 모두 존재 for f in core_fields: assert f in diag, f"diagnostic missing core field {f!r}: {diag}" # action const assert diag["action"] == "regen_required", f"action != 'regen_required': {diag}" # scope enum assert diag["scope"] in ("entity", "scene"), f"scope unknown: {diag}" # hint_diagnostic non-empty assert isinstance(diag["hint_diagnostic"], str) and diag["hint_diagnostic"], ( f"hint_diagnostic empty/non-string: {diag}" ) scopes_seen.add(diag["scope"]) assert scopes_seen == {"entity", "scene"}, f"scope coverage: {scopes_seen}" def test_regeneration_required_count_matches_detected(): """canary 4: t2i_review_step._execute 의 review_summary.regeneration_required count 가 detected count 와 1:1 mapping (mutation 0 후 detected == required).""" scene_detail = _scene_detail_data() with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_stub_full_emit(), ): result = run_t2i_review( entity_t2i_data=_entity_t2i_data(), scene_detail_data=scene_detail, entity_merge_data={"characters": [], "locations": [], "props": []}, entity_detail_data={"entity_details": {}}, shot_extract_data=_shot_extract(), shot_staging_data={}, vwr_data={"t2i_context": ""}, ) # detected count assert result["entity_fixes"] == 1 assert result["scene_fixes"] == 1 # mutation=0 → detected == required (1:1) assert result["entity_fixes"] == 1, "entity regen_required != detected" assert result["scene_fixes"] == 1, "scene regen_required != detected" def test_w1_atomicity_invariant(): """canary 5: spec §5.3 + Codex Round 4 Finding 1 — v6 schema active 와 production code reader pattern 정합. mixed window (v6 schema + 옛 target/suggestion reader) 절대 0. """ from app.modules.prompt_loader import load_schema # (a) v6 active schema = 5 core fields required strict for stem in ("scene_schema", "entity_schema"): schema = load_schema("t2i_review", stem) issue_obj_schema = schema["properties"]["results"]["items"]["properties"]["issues"]["items"] required = set(issue_obj_schema["required"]) assert {"id", "scope", "issue_code", "action", "hint_diagnostic"}.issubset(required), ( f"{stem} 5 core fields not required: missing={{'id','scope','issue_code','action','hint_diagnostic'}} - required={required}" ) # legacy target/suggestion field 폐기 검증 properties = set(issue_obj_schema["properties"].keys()) assert "target" not in properties, f"{stem} legacy target field 잔존" assert "suggestion" not in properties, f"{stem} legacy suggestion field 잔존" # (b) production code reader pattern 0 (mixed window 검출) # Codex Round 1 Important #3 fix-up: comma-default form 도 잡히도록 close paren \) 폐기. # Path resolution: BACKEND_ROOT 기반 (parents[2] = backend/). pytest CWD=backend/ # 와 독립적으로 동작 (codebase 일관 패턴, plan amend 6 정합). BACKEND_ROOT = pathlib.Path(__file__).resolve().parents[2] src = (BACKEND_ROOT / "app/modules/pipeline/t2i_review.py").read_text() forbidden_reader_patterns = [ r'iss\["target"\]', r'iss\["suggestion"\]', r'(?:iss|fix)\.get\(\s*["\']target["\']', # open-ended: catches both .get("target") and .get("target", "") r'(?:iss|fix)\.get\(\s*["\']suggestion["\']', r'fix\["target"\]', r'fix\["suggestion"\]', ] for pat in forbidden_reader_patterns: assert re.search(pat, src) is None, ( f"production reader pattern {pat!r} 잔존 (W1 atomicity 위반)" ) # (c) production code mutation site 0 mutation_patterns = [ r"old\.replace\(target,\s*suggestion\)", r"old_prompt\.replace\(target,\s*suggestion\)", r"\.replace\(target,\s*suggestion\)", r"target\s+in\s+old\b", r"target\s+in\s+old_prompt\b", ] for pat in mutation_patterns: assert re.search(pat, src) is None, ( f"production mutation site {pat!r} 잔존 (W1 atomicity 위반)" ) # (d) manifest schema_version 검증 from app.core.step_manifest import STEP_MANIFEST t2i_row = STEP_MANIFEST["t2i_review"] assert t2i_row.get("schema_version") == 1, ( f"t2i_review manifest schema_version != 1: {t2i_row.get('schema_version')!r}" )