"""OutdoorPlaceCanonStep 결정론 테스트 (W22 ②) — 게이트/헬퍼/위임/격리. 이미지·VLM 콜은 전부 monkeypatch — 스텝의 조립·분기·UPSERT 위임만 검증. """ from unittest.mock import MagicMock, patch from app.core.steps.outdoor_place_canon_step import ( SCHEMA_VERSION, OutdoorPlaceCanonStep, collect_scale_block, pick_style_fp_path, ) def _make_step(): step = OutdoorPlaceCanonStep.__new__(OutdoorPlaceCanonStep) step.project_id = "SAMPLE_PROJECT" step.episode_id = "SAMPLE_EPISODE" step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) step.db = MagicMock() return step def _patch_settings(mode="on", flag=True): return ( patch("app.core.config.settings.background_mode", mode), patch("app.core.config.settings.outdoor_direct_compose_enabled", flag, create=True), ) # ── 게이트 ─────────────────────────────────────────────────────────── def test_gate_mode_off(): p1, p2 = _patch_settings(mode="off") with p1, p2: result = _make_step()._execute() assert result["applicable_count"] == 0 assert result["schema_version"] == SCHEMA_VERSION def test_gate_flag_off(): p1, p2 = _patch_settings(flag=False) with p1, p2: result = _make_step()._execute() assert result["data"] == {"groups": {}} def test_no_spec_groups_returns_empty(): step = _make_step() step._load_prev_checkpoint = lambda sid: ( {"data": {"groups": {"g1": {"skipped": "no scenes"}}}} if sid == "outdoor_place_spec" else None ) p1, p2 = _patch_settings() with p1, p2: result = step._execute() assert result["applicable_count"] == 0 # ── 헬퍼 ───────────────────────────────────────────────────────────── def test_collect_scale_block_filters_by_group_and_dedups(): cp = {"data": {"floor_plans": { "fp1": {"group_id": "g1", "scale_guideline": "three storeys"}, "fp2": {"group_id": "g1", "scale_guideline": "three storeys"}, # 중복 "fp3": {"group_id": "g2", "scale_guideline": "one storey"}, "fp4": {"group_id": "g1", "scale_guideline": ""}, }}} assert collect_scale_block(cp, "g1") == "three storeys" assert collect_scale_block(cp, "g3") == "" assert collect_scale_block(None, "g1") == "" def test_pick_style_fp_prefers_same_group_then_first_ok(tmp_path): a = tmp_path / "a.png"; a.write_bytes(b"x") b = tmp_path / "b.png"; b.write_bytes(b"x") render_cp = {"data": {"floor_plans": { "fp_a": {"status": "ok", "png_path": str(a)}, "fp_b": {"status": "ok", "png_path": str(b)}, "fp_c": {"status": "failed", "png_path": ""}, }}} prompt_cp = {"data": {"floor_plans": { "fp_a": {"group_id": "g_other"}, "fp_b": {"group_id": "g_mine"}, }}} assert pick_style_fp_path(render_cp, prompt_cp, "g_mine") == b # 같은 그룹 fp 없으면 fp_id 정렬 첫 ok assert pick_style_fp_path(render_cp, prompt_cp, "g_none") == a assert pick_style_fp_path(None, None, "g") is None # ── 위임 / 격리 ────────────────────────────────────────────────────── def _cps(tmp_path=None): return { "outdoor_place_spec": {"data": {"groups": { "g1": {"spec": {"layout_narration_en": "x", "zone_labels_en": ["Y"], "items": []}, "outdoor_loc_ids": ["L01"], "scene_indices": [3]}, "g_skip": {"skipped": "no scenes"}, }}}, "floor_plan_prompt": {"data": {"floor_plans": {}}}, "floor_plan_render": {"data": {"floor_plans": {}}}, } def _run(step, fake_run): p1, p2 = _patch_settings() with p1, p2, patch( "app.modules.pipeline.outdoor_place_canon.run_outdoor_place_canon_group", side_effect=fake_run, ), patch( "app.core.creator_corrections.project_corrections_block", return_value="", ), patch( "app.services.image_capture.context.generation_context", ) as gen_ctx: gen_ctx.return_value.__enter__ = MagicMock(return_value=None) gen_ctx.return_value.__exit__ = MagicMock(return_value=False) return step._execute() def test_delegates_only_spec_groups_and_upserts(): step = _make_step() step._load_prev_checkpoint = lambda sid: _cps().get(sid) step._make_nb2_fn = MagicMock(return_value=lambda *a, **k: b"png") step._make_gpt_edit_fn = MagicMock(return_value=lambda *a, **k: b"png") step._upsert_canon_asset = MagicMock(side_effect=["master-id", "map-id"]) captured = {} def fake_run(**kwargs): captured.update(kwargs) return { "status": "ok", "master_png_path": "/tmp/m.png", "map_png_path": "/tmp/p.png", "selected_candidate": "A", "judge_totals": {"A": 15}, "issues_count": 0, "issues": [], "fix_applied": False, "style_fp_attached": False, "photo_prompt": "PP", "map_prompt": "MP", } result = _run(step, fake_run) # skipped 그룹 제외, spec 그룹만 applicable assert result["applicable_count"] == 1 assert result["completed_count"] == 1 and result["failed_count"] == 0 g = result["data"]["groups"]["g1"] assert g["master_asset_id"] == "master-id" assert g["map_asset_id"] == "map-id" assert captured["group_id"] == "g1" # UPSERT 2회: master 는 input 없음, map 은 master lineage calls = step._upsert_canon_asset.call_args_list assert calls[0].kwargs["variant_type"] == "master" assert calls[0].kwargs["input_image_ids"] == [] assert calls[1].kwargs["variant_type"] == "map" assert calls[1].kwargs["input_image_ids"] == ["master-id"] assert calls[0].kwargs["pipeline_role"] == "outdoor_canon_photo" assert calls[1].kwargs["pipeline_role"] == "outdoor_canon_map" step.db.commit.assert_called_once() def test_group_failure_isolated(): step = _make_step() step._load_prev_checkpoint = lambda sid: _cps().get(sid) step._make_nb2_fn = MagicMock(return_value=lambda *a, **k: b"png") step._make_gpt_edit_fn = MagicMock(return_value=lambda *a, **k: b"png") step._upsert_canon_asset = MagicMock() def fake_run(**kwargs): raise RuntimeError("SAMPLE 체인 실패") result = _run(step, fake_run) assert result["failed_count"] == 1 and result["completed_count"] == 0 assert result["data"]["groups"]["g1"]["status"] == "failed" step._upsert_canon_asset.assert_not_called() # ── 접수 불명은 재구매로 이어지지 않는다 ──────────────────────────── # # 두 그리는 함수 다 `except Exception` 으로 감싼 재시도 루프를 갖고 있다. # 클라이언트가 「보냈는지 모른다」로 멈춘 것을 그 자리에서 삼키면, 클라이언트 # 안에서 막은 중복 결제가 한 층 위에서 그대로 열린다. # - nb2 : attempt 루프 한 겹 (2026-08-26 자체 리뷰 MEDIUM-4) # - map : 모델 × attempt 두 겹 (2026-08-26 Codex 재리뷰 BLOCK) # # ★ 두 시험 다 「예외가 나갔다」만 보지 않고 **부른 횟수가 1** 인지를 잰다. # 전파만 확인하면 `raise last_exc` 로 끝나는 코드도 같이 통과한다. def test_nb2_does_not_resend_on_submission_unknown(): from app.modules.llm.image_send_state import ImageSubmissionUnknown import pytest step = _make_step() calls = [] class _FakeClient: def set_context(self, **kw): return self def generate_image(self, prompt, **kw): calls.append(prompt) raise ImageSubmissionUnknown("보냈는지 모른다", cause="wall_timeout") with patch("app.modules.llm.gemini_image_client.GeminiImageClient", return_value=_FakeClient()), \ patch("app.core.image_call_budget.reserve_current_call", MagicMock()): fn = step._make_nb2_fn() with pytest.raises(ImageSubmissionUnknown): fn("PROMPT") assert len(calls) == 1, f"접수 불명 뒤에 다시 보냈다: {len(calls)}회" def test_map_edit_does_not_resend_on_submission_unknown(): from app.modules.llm.image_send_state import ImageSubmissionUnknown import pytest step = _make_step() calls = [] def _fake_draw(**kwargs): calls.append(kwargs.get("model")) raise ImageSubmissionUnknown("보냈는지 모른다", cause="wall_timeout") with patch("app.modules.pipeline.marker_map_engine.draw_marker_map_via_model", _fake_draw), \ patch("app.modules.pipeline.marker_map_engine.marker_map_models", lambda: ["nb2", "gpt-image-2"]), \ patch("app.core.steps.outdoor_place_canon_step._resolve_openai_client", MagicMock()), \ patch("app.core.image_call_budget.reserve_current_call", MagicMock()): fn = step._make_gpt_edit_fn() with pytest.raises(ImageSubmissionUnknown): fn("PROMPT", []) # 모델 2개 × attempt 2회 = 최대 4번 살 수 있던 자리다. assert len(calls) == 1, f"접수 불명 뒤에 다시 보냈다: {calls}"