"""Phase 7 background-redesign integration smoke test. Exercises all 6 steps in documented order with mocked LLM (`call_structured`) and mocked `render_one_*` image helpers, verifying: 1. Each step's ``_execute()`` runs without raising. 2. The pipeline preserves Phase 6 shape — final ``background_render`` output is ``data.groups[bg_id]`` with the keys ``status, location_id, png_path, t2i_prompt, shot_guides, shot_ids, parent_id, ref_used``. 3. Master-plan invariant: same ``sub_location`` reuses the same ``fp_id`` (here ``fp_living`` is referenced by both backgrounds). 4. Steps execute in documented order classify → master_plan → fp_prompt → fp_render → bg_prompt → bg_render. Pattern follows existing unit tests (``tests/core/test_background_master_plan_step.py``): ``Step.__new__`` to bypass ``__init__`` + ``_load_prev_checkpoint`` mocked from a shared dict that progresses as each step finishes. """ from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock, patch import pytest 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.background_render_step import BackgroundRenderStep 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.""" step = cls.__new__(cls) step.project_id = project_id step.episode_id = episode_id step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) # MagicMock db so that _register_image_assets best-effort UPSERT doesn't # crash on missing attribute. EntityCanon query returns [] → UPSERT skips. 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 사용. # __init__ 우회 시 attribute 부재 → warning. tmp_path 기반 빈 dir 주입 (manifest.json # 없으므로 fresh start, return {}). 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). 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 smoke test") 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_phase7_e2e_smoke(tmp_path: Path) -> None: project_id = "p_smoke" episode_id = "e_smoke" # ── Upstream stubs (entity_*/scene_save/shot_*/visual_world_rules) ── 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."} ] } } 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, } call_order: list = [] # Redirect projects_dir so render PNGs land in tmp_path 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.config.settings.background_render_reference_mode", "legacy" ), 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() call_order.append("background_classify") assert classify_out["completed_count"] == 1 groups = classify_out["data"]["building_groups"] assert len(groups) == 1 assert groups[0]["kind"] == "chain_bg" 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 도 새 형식. # day intent 가 먼저 (loc/time_phase 정렬 — day < dusk) → L01B01. "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() call_order.append("background_master_plan") assert plan_out["completed_count"] == 1 plans_map = plan_out["data"]["plans"] assert "bg_x" in plans_map assert plans_map["bg_x"]["status"] == "ok" # Invariant: same sub_location → same fp_id bgs = plans_map["bg_x"]["plan"]["backgrounds"] sub_to_fp = {bg["sub_location"]: bg["depends_on_fp"][0] for bg in bgs} assert sub_to_fp == {"living_room": "fp_living"} prev_cp["background_master_plan"] = plan_out # ── Step 3: floor_plan_prompt ── fp_prompt_step = _make_step( FloorPlanPromptStep, project_id, episode_id, prev_cp, cp_root=tmp_path, ) with patch( "app.core.steps.floor_plan_prompt_step.run_floor_plan_prompt" ) as mock_fp_prompt: mock_fp_prompt.return_value = { "fp_id": "fp_living", "t2i_prompt": "Top-down architectural floor plan of a small living room. " * 3, "key_elements": ["sofa", "tv stand", "window"], } fp_prompt_out = fp_prompt_step._execute() call_order.append("floor_plan_prompt") fp_prompts = fp_prompt_out["data"]["floor_plans"] assert "fp_living" in fp_prompts assert fp_prompts["fp_living"]["status"] == "ok" assert fp_prompts["fp_living"]["t2i_prompt"] prev_cp["floor_plan_prompt"] = fp_prompt_out # ── Step 4: floor_plan_render ── 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") # minimal PNG header bytes 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() call_order.append("floor_plan_render") fp_renders = fp_render_out["data"]["floor_plans"] assert fp_renders["fp_living"]["status"] == "ok" png_path = Path(fp_renders["fp_living"]["png_path"]) assert png_path.exists() prev_cp["floor_plan_render"] = fp_render_out # Current background_prompt selector (v7+) consumes the deterministic # overlay checkpoint. This smoke keeps overlay generation out of scope # and provides the minimal valid upstream shape. prev_cp["floor_plan_overlay_payload"] = _overlay_payload_for_smoke( "L01B01", "L01B02", ) # ── Step 5: background_prompt ── bg_prompt_step = _make_step( BackgroundPromptStep, project_id, episode_id, prev_cp, cp_root=tmp_path, ) # D6: expected_bg_id 가 handshake 후 assigned ID (L01B01/L01B02). day 는 # B01, dusk 는 B02 (intent sort 순서). bg_prompts_returned = { "L01B01": { "bg_id": "L01B01", "t2i_prompt": "Wide living room, soft daylight, photoreal." * 2, "ref_guide": "use floor plan layout", "shot_guides": [ {"shot_id": "S01_Shot1", "guide": "establishing wide, daylight"} ], # G3.2: producer cp 가 owned list 를 carry — schema=2. "objects_owned_by_background": ["sofa", "window"], }, "L01B02": { "bg_id": "L01B02", "t2i_prompt": "Same living room, dusk warm tones, photoreal." * 2, "ref_guide": "match prior bg framing", "shot_guides": [ {"shot_id": "S01_Shot2", "guide": "wide, dusk amber"} ], "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 = lambda **kwargs: bg_prompts_returned[ kwargs["expected_bg_id"] ] bg_prompt_out = bg_prompt_step._execute() call_order.append("background_prompt") bg_prompts = bg_prompt_out["data"]["backgrounds"] assert {"L01B01", "L01B02"}.issubset(bg_prompts.keys()) assert all(bg_prompts[bid]["status"] == "ok" for bid in bg_prompts) prev_cp["background_prompt"] = bg_prompt_out # ── Step 6: background_render ── bg_render_step = _make_step( BackgroundRenderStep, project_id, episode_id, prev_cp, cp_root=tmp_path, ) def _fake_render_bg(*, out_path, bg_id, fp_path=None, prior_bg_paths=None, **kwargs): out_path.write_bytes(b"\x89PNG\r\n\x1a\n") ref_used = "text_only" refs_total = (1 if fp_path is not None and fp_path.exists() else 0) + sum( 1 for p in (prior_bg_paths or []) if p is not None and p.exists() ) if refs_total == 1 and fp_path is not None and fp_path.exists(): ref_used = "fp_only" elif refs_total >= 1: ref_used = f"refs_{refs_total}" return { "status": "ok", "attempts": 1, "strategies": [], "ref_used": ref_used, "final_block_reason": None, "png_path": str(out_path), } with patch( "app.core.steps.background_render_step._resolve_openai_client", return_value=MagicMock(), ), patch( "app.modules.pipeline.background_render.render_one_background", side_effect=_fake_render_bg, ): bg_render_out = bg_render_step._execute() call_order.append("background_render") # ── Phase 6 compatibility shape ── # P0-1 (W21B-w6): per-bg count — 2 bg (L01B01, L01B02) 모두 status=ok + # png_path → completed=2 (이전 이진화 `1 if failed==0 else 0` 면 1). assert bg_render_out["completed_count"] == 2 assert bg_render_out["failed_count"] == 0 groups_out = bg_render_out["data"]["groups"] required_keys = { "status", "location_id", "png_path", "t2i_prompt", "shot_guides", "shot_ids", "parent_id", "ref_used", } # D6: assign_bg_ids 가 deterministic L##B## 부여. (L01, day, quiet) → L01B01, # (L01, dusk, quiet) → L01B02 (intent sort: day < dusk → B01 먼저). for bid in ("L01B01", "L01B02"): assert bid in groups_out, f"missing {bid} in data.groups" entry = groups_out[bid] missing = required_keys - entry.keys() assert not missing, f"{bid} missing required keys: {missing}" assert entry["status"] == "ok" assert entry["location_id"] == "L01" assert Path(entry["png_path"]).exists() assert entry["t2i_prompt"] # parent_id wiring (DAG): dusk depends on day assert groups_out["L01B01"]["parent_id"] == "" assert groups_out["L01B02"]["parent_id"] == "L01B01" # shot_ids → applies_to_shots assert groups_out["L01B01"]["shot_ids"] == ["S01_Shot1"] assert groups_out["L01B02"]["shot_ids"] == ["S01_Shot2"] # ── Documented step order ── assert call_order == [ "background_classify", "background_master_plan", "floor_plan_prompt", "floor_plan_render", "background_prompt", "background_render", ]