"""Phase 8 prompt-redesign integration smoke test. Mirrors the Phase 7 ``test_phase7_e2e_smoke`` pattern but focuses on the Phase 8 v2 data flow: - Step 3 (``floor_plan_prompt``) emits ``numbered_elements`` + ``camera_recommendations`` and persists them in ``data.floor_plans[fp_id]``. - Step 5 (``background_prompt``) loads those fields from the floor-plan checkpoint and injects them into the user prompt of ``run_background_prompt``. The composed user prompt MUST contain a ``numbered_elements_block`` line and a ``camera_recommendation_block`` line that both reference the matching ``bg_id``. Render steps (Step 4 / Step 6) are out of scope — Task 7 already covers the size default in ``test_background_render.py``. We stop after Step 5 to keep this test fast and focused on the prompt data flow. """ from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock, patch from app.core.steps.background_classify_step import BackgroundClassifyStep from app.core.steps.background_master_plan_step import BackgroundMasterPlanStep from app.core.steps.background_prompt_step import BackgroundPromptStep from app.core.steps.floor_plan_prompt_step import FloorPlanPromptStep from app.core.steps.floor_plan_render_step import FloorPlanRenderStep def _make_step( cls, project_id: str, episode_id: str, prev_cp_map: dict, cp_root: Path | None = None, ): """Build a step instance bypassing __init__, wired to the shared cp map. Same pattern used by ``test_phase7_e2e.py``. """ step = cls.__new__(cls) step.project_id = project_id step.episode_id = episode_id step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) step.db = MagicMock() step.db.query.return_value.filter.return_value.all.return_value = [] step.db.query.return_value.filter_by.return_value.first.return_value = None # D6: master_plan_step 의 _load_prev_background_catalog 가 self._cp_dir 사용. if cp_root is not None: step._cp_dir = cp_root / project_id / "checkpoints" / "episodes" / episode_id / cls.__name__ def _load(sid): # Phase 8.1: world_guide / visual_world_rules 누락도 None 반환 (graceful). # Helper(visual_context_helper)가 빈 dict 처리하므로 smoke flow 무영향. if sid not in prev_cp_map: if sid in {"world_guide", "visual_world_rules"}: return None raise KeyError( f"unexpected _load_prev_checkpoint('{sid}') in phase8 smoke" ) return prev_cp_map[sid] step._load_prev_checkpoint = MagicMock(side_effect=_load) return step def _overlay_payload_for_smoke(*bg_ids: str) -> dict: """Minimal W19B-1 overlay checkpoint shape required by bg_prompt v7+.""" return { "data": { "overlays": { bid: { "bg_id": bid, "base_markers_to_reference": [], "transient_markers_to_describe": [], "ignored_state_overlay_markers": [], "clean_background_expected": True, } for bid in bg_ids } } } def test_phase8_floor_plan_to_background_flow_with_numbered_camera( tmp_path: Path, ) -> None: """fp_prompt v2 → bg_prompt_step end-to-end: numbered_elements + camera_recommendations 가 fp 체크포인트에서 bg user prompt 까지 흐른다. """ project_id = "p_phase8" episode_id = "e_phase8" # ── Upstream stubs ── entity_merge = { "data": {"locations": [{"short_id": "L01", "name": "Living"}]} } entity_detail = { "data": { "locations": [ { "short_id": "L01", "kind": "indoor", "building_group": "x", "description": "living area", } ] } } scene_save = { "data": { "segments": [ { "scene_index": 1, "heading": "INT. LIVING - DAY", "text": "A scene unfolds.", } ] } } shot_validator = { "data": { "scenes": [ { "scene_index": 1, "shots": [ { "shot_index": 1, "description": "wide of living room day", "location_id": "L01", }, { "shot_index": 2, "description": "wide of living room dusk", "location_id": "L01", }, ], } ] } } shot_selection = { "data": { "scenes": [{"scene_index": 1, "selected_shot_indices": [1, 2]}] } } visual_world_rules = {"data": {"rules_text": "minimal smoke rules"}} scene_director = { "data": { "scenes": [ { "scene_index": 1, "primary_location": "L01", "scene_type": "normal", "present_entity_ids": ["L01"], } ] } } prev_cp: dict = { "entity_merge": entity_merge, "entity_detail": entity_detail, "scene_save": scene_save, "shot_validator": shot_validator, "shot_selection": shot_selection, "visual_world_rules": visual_world_rules, "scene_director": scene_director, } with patch( "app.core.config.settings.projects_dir", str(tmp_path) ), patch("app.core.config.settings.background_mode", "on"), patch( "app.core.config.settings.background_prompt_version", "6" ), patch( "app.core.steps.background_master_plan_step.BackgroundMasterPlanStep._load_location_profiles", return_value={ "L01": {"kind": "single_space", "allowed_space_keys": ["main"]}, }, ): # ── Step 1: background_classify ── classify_step = _make_step( BackgroundClassifyStep, project_id, episode_id, prev_cp, cp_root=tmp_path, ) with patch( "app.modules.pipeline.background_classify.run_background_classify" ) as mock_classify: mock_classify.return_value = { "building_groups": [ { "group_id": "bg_x", "anchor_loc": "L01", "kind": "chain_bg", "members": [ { "loc_id": "L01", "label": "Living", "shot_count": 2, "is_indoor": True, } ], "rationale": "single indoor cluster", } ] } classify_out = classify_step._execute() prev_cp["background_classify"] = classify_out # ── Step 2: background_master_plan ── plan_step = _make_step( BackgroundMasterPlanStep, project_id, episode_id, prev_cp, cp_root=tmp_path, ) # D6 raw intent shape — assign_bg_ids 가 (loc_id, space_key, time_phase, # state_class) 4튜플 sem_key 로 deterministic L##B## 부여. handshake 가 # backgrounds[*].bg_id slot 을 새 ID 로 치환 (raw bg_id 는 placeholder). master_plan = { "group_id": "bg_x", "rationale_summary": "single cluster, 2 dayparts", "floor_plans": [ { "fp_id": "fp_living", "loc_id": "L01", "space_key_hint": "main", "sub_location": "living_room", "scope": "single living room", "depends_on_fp": [], } ], "backgrounds": [ { "bg_id": "cb_living_day", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "quiet", "surface_role": "interior_room", "state_label_raw": "day", "sub_location": "living_room", "sub_location_label": "living_room", "depends_on_fp": ["fp_living"], "depends_on_bg": [], "applies_to_shots": ["S01_Shot1"], }, { "bg_id": "cb_living_dusk", "loc_id": "L01", "space_key_hint": "main", "time_phase": "dusk", "state_class": "quiet", "surface_role": "interior_room", "state_label_raw": "dusk", "sub_location": "living_room", "sub_location_label": "living_room", "depends_on_fp": ["fp_living"], # production-like: LLM raw intent 의 depends_on_bg 도 새 형식. "depends_on_bg": ["L01B01"], "applies_to_shots": ["S01_Shot2"], }, ], "gen_order": [ "fp_living", "cb_living_day", "cb_living_dusk", ], } with patch( "app.core.steps.background_master_plan_step.run_background_master_plan" ) as mock_plan: mock_plan.return_value = master_plan plan_out = plan_step._execute() prev_cp["background_master_plan"] = plan_out # ── Step 3: floor_plan_prompt v2 — numbered_elements + camera_recommendations ── fp_prompt_step = _make_step( FloorPlanPromptStep, project_id, episode_id, prev_cp, cp_root=tmp_path, ) fp_v2_result = { "fp_id": "fp_living", "t2i_prompt": ( "Top-down architectural floor plan with numbered markers " "for furniture and openings, clean line work." ), "key_elements": ["wardrobe", "sliding window", "low table"], "numbered_elements": [ { "number": 1, "label": "main entrance", "category": "opening", "position_hint": "south wall, center", }, { "number": 2, "label": "wardrobe", "category": "furniture", "position_hint": "north wall, far end", }, { "number": 3, "label": "sliding window", "category": "opening", "position_hint": "east wall", }, ], # D6: fp_prompt 가 master_plan handshake 후 catalog 의 새 ID 로 # camera_recommendations[*].bg_id 를 작성. day → L01B01, dusk → L01B02. "camera_recommendations": [ { "bg_id": "L01B01", "sub_location": "living_room", "camera_position": ( "near number 1 (entrance), facing diagonally toward " "number 2 (wardrobe) and number 3 (sliding window)" ), "camera_height": "eye-level standing 1.6m", "lens_hint": "35mm wide angle", "framing_notes": ( "include numbers 2 and 3 prominently, soft daylight" ), }, { "bg_id": "L01B02", "sub_location": "living_room", "camera_position": ( "near number 1 (entrance), facing diagonally toward " "number 2 (wardrobe) and number 3 (sliding window)" ), "camera_height": "eye-level standing 1.6m", "lens_hint": "35mm wide angle", "framing_notes": ( "amber dusk light through number 3" ), }, ], } with patch( "app.core.steps.floor_plan_prompt_step.run_floor_plan_prompt" ) as mock_fp_prompt: mock_fp_prompt.return_value = fp_v2_result fp_prompt_out = fp_prompt_step._execute() # ── Verify Step 3 persists v2 fields ── fp_floor_plans = fp_prompt_out["data"]["floor_plans"] assert "fp_living" in fp_floor_plans fp_persisted = fp_floor_plans["fp_living"] assert fp_persisted["status"] == "ok" assert "numbered_elements" in fp_persisted, ( "Step 3 must persist numbered_elements in the checkpoint" ) assert "camera_recommendations" in fp_persisted, ( "Step 3 must persist camera_recommendations in the checkpoint" ) # numbered list shape preserved nums_persisted = fp_persisted["numbered_elements"] assert isinstance(nums_persisted, list) and len(nums_persisted) == 3 assert {e["number"] for e in nums_persisted} == {1, 2, 3} assert {e["category"] for e in nums_persisted} <= { "furniture", "opening", "prop", "plot_device", "area", } # camera entries match both bg_ids in master_plan (D6 assigned) cam_bg_ids = { c["bg_id"] for c in fp_persisted["camera_recommendations"] } assert cam_bg_ids == {"L01B01", "L01B02"} prev_cp["floor_plan_prompt"] = fp_prompt_out # ── Step 4: floor_plan_render (mocked PNG) ── fp_render_step = _make_step( FloorPlanRenderStep, project_id, episode_id, prev_cp, cp_root=tmp_path, ) from app.modules.pipeline.floor_plan_render import ( FloorPlanRenderResult, ) def _fake_render_fp(*, out_path, fp_id, **kwargs): out_path.write_bytes(b"\x89PNG\r\n\x1a\n") return FloorPlanRenderResult( fp_id=fp_id, status="ok", png_path=str(out_path), attempts=1, ref_used="text_only", error="", ) with patch( "app.core.steps.floor_plan_render_step._resolve_openai_client", return_value=MagicMock(), ), patch( "app.modules.pipeline.floor_plan_render.render_one_floor_plan", side_effect=_fake_render_fp, ): fp_render_out = fp_render_step._execute() prev_cp["floor_plan_render"] = fp_render_out # Current background_prompt selector (v7+) consumes the deterministic # overlay checkpoint. This smoke is focused on numbered/camera injection, # so provide the minimal valid upstream overlay shape. prev_cp["floor_plan_overlay_payload"] = _overlay_payload_for_smoke( "L01B01", "L01B02", ) # ── Step 5: background_prompt — capture user_prompt for inject verification ── bg_prompt_step = _make_step( BackgroundPromptStep, project_id, episode_id, prev_cp, cp_root=tmp_path, ) captured_user_prompts: dict = {} def _fake_run_bg_prompt(**kwargs): bid = kwargs["expected_bg_id"] captured_user_prompts[bid] = kwargs["user_prompt"] return { "bg_id": bid, "t2i_prompt": ( "Eye-level cinematic interior photograph of a small " "living room, daylight, photoreal, 16:9 cinematic aspect " "ratio." ), "ref_guide": "use floor plan layout", "shot_guides": [ { # D6: bid 가 assigned ID. day=L01B01, dusk=L01B02. "shot_id": ( "S01_Shot1" if bid == "L01B01" else "S01_Shot2" ), "guide_text": "wide framing", } ], # G3.2: producer cp 가 owned list 를 carry — schema=2. "objects_owned_by_background": ["sofa", "window"], } with patch( "app.core.steps.background_prompt_step.run_background_prompt" ) as mock_bg_prompt: mock_bg_prompt.side_effect = _fake_run_bg_prompt bg_prompt_out = bg_prompt_step._execute() # ── Verify Step 5 injected fp v2 fields into the bg user prompt ── bg_prompts = bg_prompt_out["data"]["backgrounds"] # D6: assigned bg_ids (L01B01=day, L01B02=dusk). assert {"L01B01", "L01B02"}.issubset(bg_prompts.keys()) assert all(bg_prompts[bid]["status"] == "ok" for bid in bg_prompts) for bid in ("L01B01", "L01B02"): up = captured_user_prompts[bid] # numbered_elements_block — full list serialized verbatim assert "1. main entrance [opening] — south wall, center" in up, ( f"{bid} user_prompt missing numbered marker 1" ) assert "2. wardrobe [furniture] — north wall, far end" in up, ( f"{bid} user_prompt missing numbered marker 2" ) assert "3. sliding window [opening] — east wall" in up, ( f"{bid} user_prompt missing numbered marker 3" ) # camera_recommendation_block — bg_id specific entry assert "camera_position:" in up, ( f"{bid} user_prompt missing camera_position line" ) assert "camera_height: eye-level standing 1.6m" in up, ( f"{bid} user_prompt missing camera_height line" ) assert "lens_hint: 35mm wide angle" in up, ( f"{bid} user_prompt missing lens_hint line" ) assert "framing_notes:" in up, ( f"{bid} user_prompt missing framing_notes line" ) # numbered ref carried verbatim into camera_position assert "near number 1" in up, ( f"{bid} user_prompt missing numbered reference inside camera" ) # bg-specific framing_notes routed correctly (no cross-contamination) assert "soft daylight" in captured_user_prompts["L01B01"] assert "amber dusk light" in captured_user_prompts["L01B02"] assert "soft daylight" not in captured_user_prompts["L01B02"] assert "amber dusk light" not in captured_user_prompts["L01B01"]