"""SceneGenerationCoordinator 단위 테스트 — W5 F22 Phase B.25. SceneImageService에서 분리된 9개 bound method의 로직을 검증. scene_image_service는 delegation smoke test만 (self._coord 호출). Fixture `coord`는 SceneGenerationCoordinator.__new__로 생성한 뒤 최소 필수 속성 (_db, _project_id, _actor_id, _logger, _persistence_svc, _reference_svc, _validation_svc, _variation_svc, _provenance_svc)을 MagicMock으로 주입. """ from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock import pytest from app.services.scene_generation_coordinator import SceneGenerationCoordinator @pytest.fixture def coord() -> SceneGenerationCoordinator: instance = SceneGenerationCoordinator.__new__(SceneGenerationCoordinator) instance._db = MagicMock() instance._project_id = "p1" instance._actor_id = "u1" instance._logger = MagicMock() instance._persistence_svc = MagicMock() instance._reference_svc = MagicMock() instance._validation_svc = MagicMock() instance._variation_svc = MagicMock() instance._provenance_svc = MagicMock() return instance # ────────────────────────────────────────────────────────────────────── # _finalize_and_track_primary — W5 F22 Phase B.22.9 # ────────────────────────────────────────────────────────────────────── def test_finalize_and_track_primary_happy_path(coord, tmp_path): """best_idx 범위 내 + file exists → 4개 dict 업데이트 + set_primary + mark_completed.""" coord._persistence_svc = MagicMock() img_path = tmp_path / "primary.png" img_path.write_bytes(b"IMG") saved_asset_ids = ["a1", "a2", "a3"] var_results_list = [ {"file_path": str(tmp_path / "x.png")}, {"file_path": str(img_path)}, # best_idx=1 {"file_path": str(tmp_path / "y.png")}, ] still_data = { "id": "still-uuid", "scene_index": 5, "shot_index": 2, } scene_cp = MagicMock() scene_results = {} scene_paths = {} scene_paths_by_id = {} primary_path = coord._finalize_and_track_primary( best_idx=1, saved_asset_ids=saved_asset_ids, var_results_list=var_results_list, still_data=still_data, si=5, scene_cp=scene_cp, scene_results_by_index=scene_results, scene_paths_by_index=scene_paths, scene_paths_by_index_by_id=scene_paths_by_id, ) coord._persistence_svc.set_primary_asset.assert_called_once_with("a2") assert primary_path == img_path assert scene_results[5] == var_results_list[1] assert scene_paths[5] == img_path assert scene_paths_by_id["still-uuid"] == img_path scene_cp.mark_completed.assert_called_once_with("still-uuid", { "asset_ids": saved_asset_ids, "primary_id": "a2", "primary_path": str(img_path), }) def test_finalize_and_track_primary_best_idx_out_of_bounds(coord, tmp_path): """best_idx가 범위 밖 → saved_asset_ids[0] + var_results_list[0] fallback.""" coord._persistence_svc = MagicMock() img = tmp_path / "first.png"; img.write_bytes(b"F") saved = ["a1", "a2"] var = [{"file_path": str(img)}, {"file_path": str(tmp_path / "x.png")}] still_data = {"id": "s1", "scene_index": 1, "shot_index": 1} scene_cp = MagicMock() results = {} primary_path = coord._finalize_and_track_primary( best_idx=99, saved_asset_ids=saved, var_results_list=var, still_data=still_data, si=1, scene_cp=scene_cp, scene_results_by_index=results, scene_paths_by_index={}, scene_paths_by_index_by_id={}, ) coord._persistence_svc.set_primary_asset.assert_called_once_with("a1") assert primary_path == img # primary_result fallback = var[0] assert results[1] == var[0] # mark_completed primary_id = a1 (saved[0]) called_args = scene_cp.mark_completed.call_args assert called_args[0][1]["primary_id"] == "a1" def test_finalize_and_track_primary_file_missing_skips_path_maps(coord, tmp_path): """primary_path not exists → scene_results_by_index만 갱신, path_maps는 불변.""" coord._persistence_svc = MagicMock() missing = tmp_path / "missing.png" # write_bytes 호출 안 함 saved = ["a1"] var = [{"file_path": str(missing)}] still_data = {"id": "s1", "scene_index": 3, "shot_index": 4} scene_cp = MagicMock() paths_by_idx = {} paths_by_id = {} results = {} primary_path = coord._finalize_and_track_primary( best_idx=0, saved_asset_ids=saved, var_results_list=var, still_data=still_data, si=3, scene_cp=scene_cp, scene_results_by_index=results, scene_paths_by_index=paths_by_idx, scene_paths_by_index_by_id=paths_by_id, ) assert primary_path == missing assert results[3] == var[0] # 무조건 기록 assert paths_by_idx == {} # file exists 조건 실패 → 갱신 안 됨 assert paths_by_id == {} # mark_completed는 여전히 호출 (primary_path str로 기록) scene_cp.mark_completed.assert_called_once() def test_finalize_and_track_primary_missing_still_id_uses_empty_string(coord, tmp_path): """still_data.id 없으면 '' 키로 scene_paths_by_index_by_id 저장 + mark_completed still_id=''.""" coord._persistence_svc = MagicMock() img = tmp_path / "p.png"; img.write_bytes(b"P") saved = ["a1"] var = [{"file_path": str(img)}] still_data = {"scene_index": 1, "shot_index": 1} # id 없음 scene_cp = MagicMock() paths_by_id = {} coord._finalize_and_track_primary( best_idx=0, saved_asset_ids=saved, var_results_list=var, still_data=still_data, si=1, scene_cp=scene_cp, scene_results_by_index={}, scene_paths_by_index={}, scene_paths_by_index_by_id=paths_by_id, ) assert paths_by_id[""] == img # mark_completed still_id="" (fallback) scene_cp.mark_completed.assert_called_once() assert scene_cp.mark_completed.call_args[0][0] == "" # ────────────────────────────────────────────────────────────────────── # _finalize_single_scene — W5 F22 Phase B.23.4 # ────────────────────────────────────────────────────────────────────── def test_finalize_single_scene_full_pipeline(coord, monkeypatch): """lineage 생성 + save + log + image_to_dict 순서 검증.""" coord._persistence_svc = MagicMock() # mock build_lineage_fields lineage_called = {} def _fake_build_lineage(db, pid, gen_type, **kwargs): lineage_called.update({"db": db, "pid": pid, "gen_type": gen_type, **kwargs}) return {"prompt_type": kwargs.get("prompt_type")} import app.services.scene_generation_coordinator as sis monkeypatch.setattr(sis, "build_lineage_fields", _fake_build_lineage) fake_asset = MagicMock() coord._persistence_svc.save_single_scene_asset.return_value = fake_asset itd_called = {} def _fake_image_to_dict(a): itd_called["asset"] = a return {"id": "dict-id"} monkeypatch.setattr(sis, "image_to_dict", _fake_image_to_dict) scene_result = {"id": "sr-1", "file_path": "/tmp/a.png"} visible = [{"id": "e1"}, {"id": "e2"}] result = coord._finalize_single_scene( scene_result=scene_result, visible_entities=visible, still_id="still-1", ip="1.2.3.4", ) assert lineage_called["gen_type"] == "scene_image_generator" assert lineage_called["ref_entity_ids"] == ["e1", "e2"] assert lineage_called["prompt_type"] == "original" # save 호출 coord._persistence_svc.save_single_scene_asset.assert_called_once_with( scene_result, {"prompt_type": "original"}, ) # activity log coord._logger.log.assert_called_once() log_kwargs = coord._logger.log.call_args.kwargs assert log_kwargs["action"] == "image.generate_single_scene" assert log_kwargs["resource_type"] == "image" assert log_kwargs["resource_id"] == "sr-1" assert log_kwargs["detail"] == {"still_id": "still-1"} assert log_kwargs["ip_address"] == "1.2.3.4" # image_to_dict 반환값 assert itd_called["asset"] is fake_asset assert result == {"id": "dict-id"} def test_finalize_single_scene_empty_visible_entities(coord, monkeypatch): """visible_entities 비어있으면 ref_entity_ids=[]로 lineage 생성.""" coord._persistence_svc = MagicMock() captured = {} def _fake_build_lineage(db, pid, gen_type, **kwargs): captured.update(kwargs) return {"lineage": "x"} import app.services.scene_generation_coordinator as sis monkeypatch.setattr(sis, "build_lineage_fields", _fake_build_lineage) monkeypatch.setattr(sis, "image_to_dict", lambda a: {"ok": True}) coord._persistence_svc.save_single_scene_asset.return_value = MagicMock() result = coord._finalize_single_scene( scene_result={"id": "sr"}, visible_entities=[], still_id="s", ip=None, ) assert captured["ref_entity_ids"] == [] assert result == {"ok": True} # ────────────────────────────────────────────────────────────────────── # _build_single_scene_prompt_and_refs — W5 F22 Phase B.23.5 # ────────────────────────────────────────────────────────────────────── # Task 2 (single-vs-batch reference contract fix, commit pending): 기존 3 test # (full_path / no_variations_uses_fallback / build_exception_fallback) 는 helper # 도입 전 inline-flow 검증. _build_single_scene_prompt_and_refs 가 이제 # build_scene_attached_refs helper 호출 form 으로 변경되어 contract 가 다름. # # 동등 검증은 다음 신규 test 로 이관: # - tests/services/test_single_scene_uses_helper.py (4 tests — helper 호출 + 9 fields 전달) # - tests/services/test_build_scene_attached_refs.py (7 tests — helper 5a~5d 분기 + v2 contract) # spec: docs/superpowers/specs/2026-05-08-single-batch-reference-contract-design.md §4.1 # plan: docs/superpowers/plans/2026-05-08-single-batch-reference-contract-implementation.md Task 2 # ────────────────────────────────────────────────────────────────────── # _maybe_generate_variation_slot — W5 F22 Phase B.24.1 # ────────────────────────────────────────────────────────────────────── def test_maybe_generate_variation_slot_none_type_returns_none(coord): """variation_{slot}_type이 없거나 'none'이면 None 반환.""" still = MagicMock() still.variation_a_type = None coord._variation_svc = MagicMock() result = coord._maybe_generate_variation_slot( slot="a", still=still, i2i_editor=MagicMock(), original_bytes=b"x", original_id="img1", scene_dir=MagicMock(), episode_id="ep", lineage={}, ) assert result is None coord._variation_svc._generate_variation_i2i.assert_not_called() def test_maybe_generate_variation_slot_none_string_returns_none(coord): """variation_{slot}_type == 'none' 문자열도 스킵.""" still = MagicMock() still.variation_b_type = "none" coord._variation_svc = MagicMock() result = coord._maybe_generate_variation_slot( slot="b", still=still, i2i_editor=MagicMock(), original_bytes=b"x", original_id="img1", scene_dir=MagicMock(), episode_id="ep", lineage={}, ) assert result is None def test_maybe_generate_variation_slot_calls_i2i_with_dynamic_fields(coord): """slot='a'일 때 variation_a_* 필드가 올바른 키워드 인자로 전달.""" still = MagicMock() still.variation_a_type = "angle" still.variation_a_angle = '{"h":10}' still.variation_a_color = None coord._variation_svc = MagicMock() coord._variation_svc._generate_variation_i2i.return_value = {"id": "new"} result = coord._maybe_generate_variation_slot( slot="a", still=still, i2i_editor="EDITOR", original_bytes=b"orig", original_id="o1", scene_dir="DIR", episode_id="ep", lineage={"k": "v"}, ) assert result == {"id": "new"} called_kwargs = coord._variation_svc._generate_variation_i2i.call_args.kwargs assert called_kwargs["variant_label"] == "variant_a" assert called_kwargs["var_type"] == "angle" assert called_kwargs["angle_json"] == '{"h":10}' assert called_kwargs["color_prompt"] is None assert called_kwargs["original_bytes"] == b"orig" assert called_kwargs["original_id"] == "o1" assert called_kwargs["scene_dir"] == "DIR" assert called_kwargs["episode_id"] == "ep" assert called_kwargs["lineage"] == {"k": "v"} assert called_kwargs["still"] is still assert called_kwargs["i2i_editor"] == "EDITOR" def test_maybe_generate_variation_slot_b_dispatches_to_b_fields(coord): """slot='b'일 때는 variation_b_* 필드 사용 (getattr dynamic dispatch).""" still = MagicMock() still.variation_b_type = "color" still.variation_b_angle = None still.variation_b_color = "warm orange tone" coord._variation_svc = MagicMock() coord._variation_svc._generate_variation_i2i.return_value = {"id": "b1"} coord._maybe_generate_variation_slot( slot="b", still=still, i2i_editor="E", original_bytes=b"o", original_id="oid", scene_dir="D", episode_id="ep", lineage={}, ) kwargs = coord._variation_svc._generate_variation_i2i.call_args.kwargs assert kwargs["variant_label"] == "variant_b" assert kwargs["var_type"] == "color" assert kwargs["color_prompt"] == "warm orange tone" assert kwargs["angle_json"] is None # ────────────────────────────────────────────────────────────────────── # _finalize_scene_with_variations — W5 F22 Phase B.24.2 # ────────────────────────────────────────────────────────────────────── def test_finalize_variations_recommended_A_sets_variant_primary(coord, monkeypatch): """recommended='A' + variant_a 있으면 A를 primary로 설정.""" import app.services.scene_generation_coordinator as sis primary_calls = [] def _fake_set_primary(db, pid, img_id, still_id): primary_calls.append((img_id, still_id)) monkeypatch.setattr(sis, "_set_variant_primary_helper", _fake_set_primary) still = MagicMock() still.recommended_variant = "A" result = coord._finalize_scene_with_variations( still=still, still_id="s1", original_result={"id": "orig1"}, original_id="orig1", variant_a_result={"id": "a1"}, variant_b_result={"id": "b1"}, ip="1.2.3.4", ) assert primary_calls == [("a1", "s1")] coord._db.commit.assert_called_once() assert result["recommended"] == "A" def test_finalize_variations_recommended_B_sets_b_primary(coord, monkeypatch): """recommended='B' + variant_b 있으면 B를 primary로 설정.""" import app.services.scene_generation_coordinator as sis primary_calls = [] monkeypatch.setattr(sis, "_set_variant_primary_helper", lambda db, pid, img, sid: primary_calls.append(img)) still = MagicMock(); still.recommended_variant = "B" coord._finalize_scene_with_variations( still=still, still_id="s1", original_result={"id": "o"}, original_id="o", variant_a_result={"id": "a"}, variant_b_result={"id": "b"}, ip=None, ) assert primary_calls == ["b"] def test_finalize_variations_recommended_original_skips_primary(coord, monkeypatch): """recommended='original' (or default)이면 primary 변경 안 함.""" import app.services.scene_generation_coordinator as sis primary_calls = [] monkeypatch.setattr(sis, "_set_variant_primary_helper", lambda db, pid, img, sid: primary_calls.append(img)) still = MagicMock(); still.recommended_variant = None # → 'original' fallback result = coord._finalize_scene_with_variations( still=still, still_id="s", original_result={"id": "o"}, original_id="o", variant_a_result={"id": "a"}, variant_b_result={"id": "b"}, ip=None, ) assert primary_calls == [] assert result["recommended"] == "original" def test_finalize_variations_recommended_A_without_a_result_skips(coord, monkeypatch): """recommended='A'지만 variant_a_result=None이면 primary 설정 스킵.""" import app.services.scene_generation_coordinator as sis primary_calls = [] monkeypatch.setattr(sis, "_set_variant_primary_helper", lambda db, pid, img, sid: primary_calls.append(img)) still = MagicMock(); still.recommended_variant = "A" coord._finalize_scene_with_variations( still=still, still_id="s", original_result={"id": "o"}, original_id="o", variant_a_result=None, variant_b_result={"id": "b"}, ip=None, ) assert primary_calls == [] # A result 없으므로 스킵 def test_finalize_variations_log_detail_and_return_structure(coord, monkeypatch): """activity log detail + 반환 dict 구조 검증.""" import app.services.scene_generation_coordinator as sis monkeypatch.setattr(sis, "_set_variant_primary_helper", lambda db, pid, img, sid: None) still = MagicMock(); still.recommended_variant = "A" result = coord._finalize_scene_with_variations( still=still, still_id="still1", original_result={"id": "orig", "file_path": "/x.png"}, original_id="orig", variant_a_result={"id": "va"}, variant_b_result=None, ip="9.9.9.9", ) # 로그 detail 검증 log_kwargs = coord._logger.log.call_args.kwargs assert log_kwargs["action"] == "still.generate_with_variations" assert log_kwargs["resource_type"] == "still" assert log_kwargs["resource_id"] == "still1" assert log_kwargs["detail"]["original_id"] == "orig" assert log_kwargs["detail"]["variant_a_id"] == "va" assert log_kwargs["detail"]["variant_b_id"] is None # None→None 유지 assert log_kwargs["detail"]["recommended"] == "A" assert log_kwargs["ip_address"] == "9.9.9.9" # 반환 dict 구조 assert result["original"] == {"id": "orig", "file_path": "/x.png"} assert result["variant_a"] == {"id": "va"} assert result["variant_b"] is None assert result["recommended"] == "A" # ────────────────────────────────────────────────────────────────────── # T4 D5 §4.6 — _generate_variation_in_loop validator wiring (B2 fix) # ────────────────────────────────────────────────────────────────────── def test_T4_variation_loop_calls_validator_after_full_prompt_build(coord, tmp_path): """T4 (B2 architectural fix): variation loop 안 validator 가 _build_final_scene_prompt 직후, generate_and_validate_scene 직전에 호출됨. variation 별 _full_prompt 가 다르므로 phantom guard 도 variation-specific. """ from unittest.mock import patch, MagicMock from app.services import scene_generation_coordinator as sgc captured = {} mock_gen = MagicMock(return_value={ "file_path": str(tmp_path / "x.png"), "generation_model": "gemini-test", "validation": {}, }) def _spy_validate(rpc, labeled_refs, attached_meta, prompt, *, is_close_framing, chain_bg_lookup=None, reference_phrase_kinds=None): captured["validator_called"] = True captured["prompt"] = prompt captured["attached_meta"] = attached_meta captured["chain_bg_lookup"] = chain_bg_lookup # validator 호출 시점에 generate_and_validate_scene 은 아직 미호출이어야 함 captured["gen_called_at_validate_time"] = mock_gen.call_count with patch.object(sgc, "_build_final_scene_prompt", return_value="FINAL_PROMPT_FOR_GEMINI"), \ patch("app.core.ref_contract_validator.validate_attached_refs", side_effect=_spy_validate), \ patch("app.modules.pipeline.scene_image_pipeline.generate_and_validate_scene", mock_gen): # Area #11 v1 W3: labeled_refs + attached_meta kwarg → payload (LabeledRefPayload). from app.services.prompt_service import make_labeled_ref_payload payload = make_labeled_ref_payload( labeled_refs=[("Image 1: ...", b"x")], ref_roles=["character_ref"], ref_role_metadata=[{}], attached_meta=[("character_outlook", "C01O02")], ) result = coord._generate_variation_in_loop( still_data={"scene_index": 8, "shot_index": 4, "id": "s1"}, var_t2i="var t2i prompt", var_theme="t1", var_theme_label="lbl", visible_entities=[], payload=payload, rpc={"asset_requirements": {"required_refs": [], "readiness_policy": None}}, is_close_framing=False, chain_bg_lookup=lambda _: None, reference_phrase_kinds=[], # Area #5 W3: sidecar gemini_client=MagicMock(), sanitizer=MagicMock(), validator=None, scene_dir=tmp_path, cached_style_context="", cached_entity_text_map={}, world_guide={}, episode_id="ep1", ) assert captured.get("validator_called") is True assert captured["prompt"] == "FINAL_PROMPT_FOR_GEMINI" assert captured["attached_meta"] == [("character_outlook", "C01O02")] # validator 호출 시점에 Gemini 는 아직 호출 안 됨 (B2 위치 binding) assert captured["gen_called_at_validate_time"] == 0 # validator 통과 후 Gemini 1회 호출 assert mock_gen.call_count == 1 assert result is not None def test_T4_variation_loop_blocks_gemini_on_ref_contract_violation(coord, tmp_path): """T4 (B2) + 2026-05-10 Fix C: validator RefContractError → Gemini 호출 X. Fix C (2026-05-10): variation 결과를 None silent skip 하지 않고 typed failure dict 로 반환 — `_failure_reason` + `_failure_detail` 보존하여 cp.failed 메시지로 직렬화 가능. 기존 None 시그니처는 cp 가 generic "all variations failed" 만 남겨 24 shot deterministic 실패 root cause 가시성 차단했음. """ from unittest.mock import patch, MagicMock from app.services import scene_generation_coordinator as sgc from app.core.ref_contract_validator import RefContractError mock_gen = MagicMock() with patch.object(sgc, "_build_final_scene_prompt", return_value="FINAL_PROMPT"), \ patch("app.core.ref_contract_validator.validate_attached_refs", side_effect=RefContractError("required character_outlook 'C01O02' missing")), \ patch("app.modules.pipeline.scene_image_pipeline.generate_and_validate_scene", mock_gen): # Area #11 v1 W3: labeled_refs + attached_meta kwarg → payload (LabeledRefPayload). from app.services.prompt_service import make_labeled_ref_payload payload = make_labeled_ref_payload( labeled_refs=[("Image 1: ...", b"x")], ref_roles=["character_ref"], ref_role_metadata=[{}], attached_meta=[("character", "C01")], # base fallback — C01O02 요구 만족 X ) result = coord._generate_variation_in_loop( still_data={"scene_index": 8, "shot_index": 4, "id": "s1"}, var_t2i="var t2i", var_theme="t1", var_theme_label="lbl", visible_entities=[], payload=payload, rpc={"asset_requirements": {"required_refs": [ {"kind": "character_outlook", "id": "C01O02", "policy": "required"}, ]}}, is_close_framing=False, chain_bg_lookup=lambda _: None, reference_phrase_kinds=[], # Area #5 W3: sidecar (step 6 통과, step 1 required_refs 위반으로 RefContractError 발생) gemini_client=MagicMock(), sanitizer=MagicMock(), validator=None, scene_dir=tmp_path, cached_style_context="", cached_entity_text_map={}, world_guide={}, episode_id="ep1", ) # Fix C: typed failure dict (None 아님) — caller 가 cp.failed 에 reason 직렬화 가능. assert isinstance(result, dict) assert result.get("_failure_reason") == "REF_CONTRACT_VIOLATION" assert "C01O02" in result.get("_failure_detail", "") assert mock_gen.call_count == 0 # Gemini 호출 차단 # ────────────────────────────────────────────────────────────────────── # FINDING 9 Cat3 — batch-path render_prompt_card ordering # ────────────────────────────────────────────────────────────────────── def test_batch_path_passes_required_refs_to_resolver(coord, monkeypatch, tmp_path): """FINDING 9 Cat3: batch path 가 render_prompt_card.required_refs 를 resolve_refs_for_prompt_set 호출 시점에 전달하는지 검증. 버그: rpc resolution(still_data["render_prompt_card"] inject)이 resolver 호출보다 뒤에 있어 _patch_a_required_refs_batch 가 항상 [] 를 수신 → batch-path prop attach 100% broken. still_data 는 DB shape (render_prompt_card 필드 없음) — resolver 는 manifest lookup 결과를 봐야 한다. """ import app.services.scene_generation_coordinator as sgc class _StopAfterResolver(Exception): pass rpc = { "asset_requirements": { "required_refs": [{"kind": "prop", "id": "P10", "policy": "required"}], }, } monkeypatch.setattr(sgc, "lookup_render_prompt_card", lambda **kw: rpc) captured = {} def _capture_resolver(*a, **kw): captured["required_refs"] = kw.get("required_refs") raise _StopAfterResolver coord._reference_svc.resolve_refs_for_prompt_set = _capture_resolver coord._reference_svc.detect_state_variant_sids = MagicMock(return_value={}) still_data = { "id": "1b7bc30b", "scene_index": 18, "shot_index": 15, "visible_entities_json": "[]", "t2i_variations": [{"t2i_prompt": "a prompt"}], } with pytest.raises(_StopAfterResolver): coord._generate_scene_in_loop( si=0, stills=[still_data], entity_lookup={}, scene_paths_by_index_by_id={}, location_scene_history={}, staging_map={}, scene_ref_image_map={}, background_chain_bg_map={}, dep_detail_map={}, gemini_client=MagicMock(), sanitizer=MagicMock(), validator=None, scene_dir=tmp_path, cached_style_context="", cached_entity_text_map={}, world_guide={}, episode_id="ep1", ) assert captured["required_refs"] == [ {"kind": "prop", "id": "P10", "policy": "required"}, ]