"""Tests for BackgroundPromptStep — Phase 7 Step 5 / Phase 8 v2 step entry.""" from __future__ import annotations from unittest.mock import MagicMock, patch from app.core.steps.background_prompt_step import ( PROMPT_VERSION, BackgroundPromptStep, _build_fp_lookup_for_bg, ) def test_step_disabled_when_mode_off(): """background_mode=off → not_applicable + empty data.""" with patch("app.core.config.settings.background_mode", "off"): step = BackgroundPromptStep.__new__(BackgroundPromptStep) step.project_id = "p" step.episode_id = "e" step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) result = step._execute() assert result["applicable_count"] == 0 assert result["completed_count"] == 0 assert result["failed_count"] == 0 assert result["data"] == {} # ----- Phase 8 v2 ----- def test_bg_prompt_step_prompt_version_bumped_to_v2(): """PROMPT_VERSION 갱신 history. v5 (G3.2): 5.202605032354 — Rule 12 objects_owned_by_background contract 추가. D6 v6 (2026-05-09): 6.202605091200 — bg_id pattern → BG_ID_RE + state_label_raw / sub_location_label 통합 (D6 raw intent 정합) + runtime enum injection. """ assert PROMPT_VERSION == "6.202605091200" def test_build_fp_lookup_for_bg_resolves_numbered_and_camera(): """v2 helper: bg_id → (numbered_elements list, camera entry) flat 매핑. master_plan 의 backgrounds[].depends_on_fp[0] 으로 fp_id 매핑 후 fp_prompt cp 의 fp.numbered_elements 와 fp.camera_recommendations[bg_id] 를 lookup 한다. """ fp_prompt_cp = { "data": { "floor_plans": { "fp_a": { "numbered_elements": [ { "number": 1, "label": "wardrobe", "category": "furniture", "position_hint": "north wall", }, { "number": 2, "label": "window", "category": "opening", "position_hint": "east wall", }, ], "camera_recommendations": [ { "bg_id": "bg_x_day", "sub_location": "loc_x", "camera_position": "near 1", "camera_height": "1.6m", "lens_hint": "35mm", }, { "bg_id": "bg_x_night", "sub_location": "loc_x", "camera_position": "near 2", "camera_height": "1.6m", "lens_hint": "35mm", }, ], } } } } master_plan_cp = { "data": { "plans": { "g1": { "plan": { "backgrounds": [ { "bg_id": "bg_x_day", "depends_on_fp": ["fp_a"], }, { "bg_id": "bg_x_night", "depends_on_fp": ["fp_a"], }, { # depends_on_fp 빈 → 매핑에서 제외 "bg_id": "bg_orphan", "depends_on_fp": [], }, ] } } } } } numbered_lookup, camera_lookup = _build_fp_lookup_for_bg( fp_prompt_cp=fp_prompt_cp, master_plan_cp=master_plan_cp ) assert "bg_x_day" in numbered_lookup assert "bg_x_night" in numbered_lookup assert "bg_orphan" not in numbered_lookup # depends_on_fp 빈 → 제외 assert numbered_lookup["bg_x_day"][0]["label"] == "wardrobe" assert numbered_lookup["bg_x_day"] == numbered_lookup["bg_x_night"] assert camera_lookup["bg_x_day"]["camera_position"] == "near 1" assert camera_lookup["bg_x_night"]["camera_position"] == "near 2" def test_build_fp_lookup_handles_missing_cps_gracefully(): """v2: cp 가 None / 빈 dict 라도 빈 lookup 두 개 반환.""" a, b = _build_fp_lookup_for_bg(fp_prompt_cp={}, master_plan_cp={}) assert a == {} assert b == {} a2, b2 = _build_fp_lookup_for_bg( fp_prompt_cp={"data": {}}, master_plan_cp={"data": {}} ) assert a2 == {} assert b2 == {} def test_build_fp_lookup_handles_missing_fp_entry(): """v2: master_plan 이 참조하는 fp_id 가 fp_prompt cp 에 없으면 빈 list.""" fp_prompt_cp = {"data": {"floor_plans": {}}} master_plan_cp = { "data": { "plans": { "g1": { "plan": { "backgrounds": [ {"bg_id": "bg_x", "depends_on_fp": ["fp_missing"]} ] } } } } } numbered, camera = _build_fp_lookup_for_bg( fp_prompt_cp=fp_prompt_cp, master_plan_cp=master_plan_cp ) assert numbered["bg_x"] == [] # camera 없으면 entry 자체가 없음 (lookup.get(bg_id) 패턴 호환) assert "bg_x" not in camera def test_bg_prompt_step_injects_numbered_and_camera_into_module_call(): """v2 통합: step 이 fp_prompt cp 를 로드해서 bg 별로 numbered + camera 를 build_bg_user_prompt 에 inject 한다. """ plans_cp = { "data": { "plans": { "g1": { "status": "ok", "plan": { "backgrounds": [ { "bg_id": "bg_a", "loc_id": "L01", "sub_location": "living", "state_label": "day_normal", "depends_on_fp": ["fp_a"], "depends_on_bg": [], "applies_to_shots": ["S01_Shot1"], } ] }, } } } } fp_render_cp = { "data": { "floor_plans": { "fp_a": {"status": "ok", "png_path": "/tmp/fp_a.png"} } } } fp_prompt_cp = { "data": { "floor_plans": { "fp_a": { "numbered_elements": [ { "number": 1, "label": "wardrobe", "category": "furniture", "position_hint": "north wall", } ], "camera_recommendations": [ { "bg_id": "bg_a", "sub_location": "living", "camera_position": "near 1", "camera_height": "1.6m", "lens_hint": "35mm", } ], } } } } rules_cp = {"data": {"rules_text": "rules"}} scene_save_cp = {"data": {"segments": []}} cps = { "background_master_plan": plans_cp, "floor_plan_render": fp_render_cp, "floor_plan_prompt": fp_prompt_cp, "scene_save": scene_save_cp, "shot_validator": None, "shot_selection": None, "visual_world_rules": rules_cp, } captured: dict = {} def fake_build(*args, **kwargs): captured["numbered_elements"] = kwargs.get("numbered_elements") captured["camera_recommendation"] = kwargs.get("camera_recommendation") return "USER_PROMPT" def fake_run(**kwargs): return { "bg_id": kwargs["expected_bg_id"], "t2i_prompt": "x" * 60, "ref_guide": "g", "shot_guides": [ {"shot_id": s, "guide_text": "h"} for s in kwargs["applies_to_shots"] ], # G3.2: producer cp 가 owned list 를 carry — _execute 가 dict 에서 직접 read. "objects_owned_by_background": ["door", "window"], } step = BackgroundPromptStep.__new__(BackgroundPromptStep) step.project_id = "p" step.episode_id = "e" step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) with patch("app.core.config.settings.background_mode", "on"), patch.object( step, "_load_prev_checkpoint", side_effect=lambda sid: cps.get(sid) ), patch( "app.core.steps.background_prompt_step.build_bg_user_prompt", side_effect=fake_build, ), patch( "app.core.steps.background_prompt_step.run_background_prompt", side_effect=fake_run, ), patch( "app.core.steps.background_master_plan_step._select_scenes_for_group", return_value=([], []), ): result = step._execute() assert result["completed_count"] == 1 assert result["failed_count"] == 0 assert "bg_a" in result["data"]["backgrounds"] # build_bg_user_prompt 가 numbered + camera 두 인자를 모두 받아야 함 assert captured["numbered_elements"] is not None assert captured["numbered_elements"][0]["label"] == "wardrobe" assert captured["camera_recommendation"]["camera_position"] == "near 1" def test_bg_prompt_step_partial_reports_actual_ok_count(): """C1-fix (W21B-w5): 일부 bg 만 실패하면 completed_count 는 실제 ok 수, applicable_count 는 전체 수로 보고 → step_runner 가 'partial' 도출 (이전엔 completed=1 if failed==0 else 0 이진화 → 1개 실패에 status='failed' → dispatch 즉시 break → downstream 전체 stop). """ plans_cp = { "data": { "plans": { "g1": { "status": "ok", "plan": { "backgrounds": [ { "bg_id": "bg_a", "loc_id": "L01", "sub_location": "living", "state_label": "day_normal", "depends_on_fp": ["fp_a"], "depends_on_bg": [], "applies_to_shots": ["S01_Shot1"], }, { "bg_id": "bg_b", "loc_id": "L01", "sub_location": "living", "state_label": "night_normal", "depends_on_fp": ["fp_a"], "depends_on_bg": [], "applies_to_shots": ["S02_Shot1"], }, ] }, } } } } fp_render_cp = {"data": {"floor_plans": {"fp_a": {"status": "ok", "png_path": "/tmp/fp_a.png"}}}} fp_prompt_cp = {"data": {"floor_plans": {"fp_a": {"numbered_elements": [], "camera_recommendations": []}}}} cps = { "background_master_plan": plans_cp, "floor_plan_render": fp_render_cp, "floor_plan_prompt": fp_prompt_cp, "scene_save": {"data": {"segments": []}}, "shot_validator": None, "shot_selection": None, "visual_world_rules": {"data": {"rules_text": "rules"}}, } from app.core.steps.background_prompt_step import BackgroundPromptError def fake_run(**kwargs): bid = kwargs["expected_bg_id"] if bid == "bg_b": raise BackgroundPromptError("bg_b boom") return { "bg_id": bid, "t2i_prompt": "x" * 60, "ref_guide": "g", "shot_guides": [{"shot_id": s, "guide_text": "h"} for s in kwargs["applies_to_shots"]], "objects_owned_by_background": ["door"], } step = BackgroundPromptStep.__new__(BackgroundPromptStep) step.project_id = "p"; step.episode_id = "e"; step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) with patch("app.core.config.settings.background_mode", "on"), patch.object( step, "_load_prev_checkpoint", side_effect=lambda sid: cps.get(sid) ), patch( "app.core.steps.background_prompt_step.build_bg_user_prompt", side_effect=lambda *a, **k: "USER_PROMPT", ), patch( "app.core.steps.background_prompt_step.run_background_prompt", side_effect=fake_run, ), patch( "app.core.steps.background_master_plan_step._select_scenes_for_group", return_value=([], []), ): result = step._execute() assert result["applicable_count"] == 2 assert result["completed_count"] == 1 # 이전 이진화면 0 assert result["failed_count"] == 1 # step_runner status 도출과 동일한 식 → 'partial' (failed>0, completed>0) completed, total, failed = ( result["completed_count"], result["applicable_count"], result["failed_count"] ) final_status = "completed" if failed == 0 else ("partial" if completed > 0 else "failed") assert final_status == "partial" assert result["data"]["backgrounds"]["bg_a"]["status"] == "ok" assert result["data"]["backgrounds"]["bg_b"]["status"] == "failed"