"""run_t2i_review end-to-end producer → consumer 검증 — Area #6 v1. v1 (2026-05-18+): blind substring mutation 폐기 + 5 core fields structured diagnostic emit. mutation=0 invariant + diagnostic 누적 검증. """ from unittest.mock import 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`, 기본 True). 설정을 안 잡으면 LLM 호출이 아예 없어 여기 다섯 건이 「호출 0」으로 무너진다. ★그렇다고 지우면 안 된다 — 이들이 지키는 계약(item_id prefix · schema 위반 fail-fast · 일시 실패 전파 · mutation 0)은 legacy 경로에서 **여전히 유효하다.** 그 경로는 t2i 산문을 이미지에 실제로 쓴다. 그래서 계약을 살리고 자리를 legacy 로 고정한다. v1 이 그 갈래를 건너뛰는 것과 `scene_review_status` 는 `tests/unit/test_v1_lean_authoring.py` 가 따로 잰다. """ 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": "ORIGINAL prompt"}, ], "owned_validation": {}, }, ], } def _shot_extract(): return { "scenes": [ { "scene_index": 1, "shots": [ {"shot_index": 0, "description": "shot desc"}, ], }, ], } def _empty_entity_inputs(): return dict( entity_t2i={"characters": [], "locations": [], "props": [], "completed": {}}, entity_merge={"characters": [], "locations": [], "props": []}, entity_detail={"entity_details": {}}, vwr={"t2i_context": ""}, ) def _stub_scene_call_returns(item_id: str, issue_code: str, hint: str): """call_structured stub — v1 5 core fields schema 정합 result.""" def _fake(**kwargs): return { "results": [ { "item_id": item_id, "scene_index": 1, "var_index": 0, "has_issues": True, "issues": [ { "id": item_id, "scope": "scene", "issue_code": issue_code, "action": "regen_required", "hint_diagnostic": hint, "item_id": item_id, }, ], }, ], } return _fake def test_run_t2i_review_end_to_end_emits_diagnostic_no_mutation(): """v1: producer→diagnostic emit 끝까지 정확 propagate + mutation 0 invariant. 이전: scene fixture t2i_prompt 가 in-place mutate. 이후: t2i_prompt 변경 0 (canary), diagnostic 누적 only. """ scene_detail = _scene_detail_data() inputs = _empty_entity_inputs() fake_call = _stub_scene_call_returns( item_id="S1_L0_V0", issue_code="awkward_translation", hint="원어 표기 권장 — 한국 문화권 명사", ) with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=fake_call, ): result = run_t2i_review( entity_t2i_data=inputs["entity_t2i"], scene_detail_data=scene_detail, entity_merge_data=inputs["entity_merge"], entity_detail_data=inputs["entity_detail"], shot_extract_data=_shot_extract(), vwr_data=inputs["vwr"], ) # canary: t2i_prompt 변경 0 (mutation=0 invariant) assert ( scene_detail["scenes"][0]["t2i_variations"][0]["t2i_prompt"] == "ORIGINAL prompt" ) assert result["scene_applied"] == 0 assert result["scene_applied_indices"] == [] # diagnostic 누적 assert result["scene_fixes"] == 1 assert len(result["diagnostics"]) == 1 diag = result["diagnostics"][0] assert diag["id"] == "S1_L0_V0" assert diag["scope"] == "scene" assert diag["issue_code"] == "awkward_translation" assert diag["action"] == "regen_required" assert "원어" in diag["hint_diagnostic"] def test_run_t2i_review_raises_on_garbage_result_missing_item_id(): """schema 가 item_id required 인데 LLM 이 garbage 반환 시 (item_id 누락) fail-fast.""" from app.core.errors import AppError scene_detail = _scene_detail_data() inputs = _empty_entity_inputs() def _fake_no_item_id(**kwargs): return { "results": [ { # item_id 누락 — schema 위반 시뮬 "scene_index": 1, "var_index": 0, "has_issues": True, "issues": [ { "id": "S1_L0_V0", "scope": "scene", "issue_code": "awkward_translation", "action": "regen_required", "hint_diagnostic": "x", "item_id": "S1_L0_V0", }, ], }, ], } with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_fake_no_item_id, ): with pytest.raises(AppError) as exc_info: run_t2i_review( entity_t2i_data=inputs["entity_t2i"], scene_detail_data=scene_detail, entity_merge_data=inputs["entity_merge"], entity_detail_data=inputs["entity_detail"], shot_extract_data=_shot_extract(), vwr_data=inputs["vwr"], ) assert exc_info.value.code == "t2i_review.missing_item_id" def test_run_t2i_review_raises_on_schema_validation_error(): """call_structured 가 SchemaValidationError raise 시 fail-fast (broad except 흡수 금지).""" from app.modules.llm.safety import SchemaValidationError scene_detail = _scene_detail_data() inputs = _empty_entity_inputs() def _fake_schema_failure(**kwargs): if "씬" in kwargs.get("user_prompt", ""): raise SchemaValidationError( "Local schema validation failed for step=t2i_review at " "path=results/0/issues/0/id: missing" ) return {"results": []} with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_fake_schema_failure, ): with pytest.raises(SchemaValidationError): run_t2i_review( entity_t2i_data=inputs["entity_t2i"], scene_detail_data=scene_detail, entity_merge_data=inputs["entity_merge"], entity_detail_data=inputs["entity_detail"], shot_extract_data=_shot_extract(), vwr_data=inputs["vwr"], ) def test_run_t2i_review_raises_on_transient_batch_failure(): """transient exception → batch_failed AppError (fail-fast, partial success 금지).""" from app.core.errors import AppError scene_detail = _scene_detail_data() inputs = _empty_entity_inputs() def _fake_transient(**kwargs): if "씬" in kwargs.get("user_prompt", ""): raise RuntimeError("transient network error") return {"results": []} with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_fake_transient, ): with pytest.raises(AppError) as exc_info: run_t2i_review( entity_t2i_data=inputs["entity_t2i"], scene_detail_data=scene_detail, entity_merge_data=inputs["entity_merge"], entity_detail_data=inputs["entity_detail"], shot_extract_data=_shot_extract(), vwr_data=inputs["vwr"], ) assert exc_info.value.code == "t2i_review.batch_failed" def test_run_t2i_review_raises_on_entity_batch_failure(): """B2 patch 보존: entity batch transient failure → entity_batch_failed AppError.""" from app.core.errors import AppError inputs = _empty_entity_inputs() inputs["entity_t2i"] = { "characters": [ {"short_id": "C01", "name": "주인공", "t2i_prompt": "한 남자의 클로즈업"}, ], "locations": [], "props": [], "completed": {}, } scene_detail = _scene_detail_data() def _fake_entity_failure(**kwargs): if "엔티티" in kwargs.get("user_prompt", ""): raise RuntimeError("transient entity LLM failure") return {"results": []} with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_fake_entity_failure, ): with pytest.raises(AppError) as exc_info: run_t2i_review( entity_t2i_data=inputs["entity_t2i"], scene_detail_data=scene_detail, entity_merge_data=inputs["entity_merge"], entity_detail_data=inputs["entity_detail"], shot_extract_data=_shot_extract(), vwr_data=inputs["vwr"], ) assert exc_info.value.code == "t2i_review.entity_batch_failed" def test_run_t2i_review_user_prompt_includes_item_id_prefix(): """user_prompt 에 [item_id: ...] prefix 가 포함되어 LLM 이 echo 가능 — v0 동일 보존.""" scene_detail = _scene_detail_data() inputs = _empty_entity_inputs() captured_prompts = [] def _fake_capture(**kwargs): captured_prompts.append(kwargs.get("user_prompt", "")) return {"results": []} with patch( "app.modules.pipeline.t2i_review.call_structured", side_effect=_fake_capture, ): run_t2i_review( entity_t2i_data=inputs["entity_t2i"], scene_detail_data=scene_detail, entity_merge_data=inputs["entity_merge"], entity_detail_data=inputs["entity_detail"], shot_extract_data=_shot_extract(), vwr_data=inputs["vwr"], ) assert any("[item_id: S1_L0_V0]" in p for p in captured_prompts), ( f"user_prompt 에 item_id prefix 누락: {captured_prompts}" )