from unittest.mock import MagicMock, patch from app.core.steps.background_master_plan_step import ( BackgroundMasterPlanStep, SCHEMA_VERSION, PROMPT_VERSION, ) def test_step_disabled_when_mode_off(): with patch("app.core.config.settings.background_mode", "off"): step = BackgroundMasterPlanStep.__new__(BackgroundMasterPlanStep) 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 def test_step_skips_prev_shot_only_groups(): """kind=prev_shot_ref 그룹은 master plan 호출 skip.""" classify_data = { "data": {"building_groups": [ {"group_id": "bg_x", "kind": "chain_bg", "members": [{"loc_id": "L01"}]}, {"group_id": "bg_y", "kind": "prev_shot_ref", "members": [{"loc_id": "L02"}]}, ]} } with patch("app.core.config.settings.background_mode", "on"): step = BackgroundMasterPlanStep.__new__(BackgroundMasterPlanStep) step.project_id = "p"; step.episode_id = "e"; step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) step._load_prev_checkpoint = MagicMock(side_effect=lambda sid: { "background_classify": classify_data, "scene_save": {"data": {"segments": []}}, "shot_validator": {"data": {"scenes": []}}, "shot_selection": {"data": {"scenes": []}}, "visual_world_rules": {"data": {"rules_text": ""}}, }.get(sid)) # T4-fix3: location_profiles build 도 mock — DB query 회피. step._load_location_profiles = MagicMock(return_value={ "L01": {"kind": "single_space", "allowed_space_keys": ["main"]}, }) with patch("app.core.steps.background_master_plan_step.run_background_master_plan") as mock_run: mock_run.return_value = { "group_id": "bg_x", "rationale_summary": "", "floor_plans": [], "backgrounds": [], "gen_order": [] } result = step._execute() # bg_y는 skip되어 master plan 호출 안 됨 called_groups = [c.kwargs["expected_group_id"] for c in mock_run.call_args_list] assert "bg_y" not in called_groups def _make_step(classify_data, profiles): step = BackgroundMasterPlanStep.__new__(BackgroundMasterPlanStep) step.project_id = "p"; step.episode_id = "e"; step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) step._load_prev_checkpoint = MagicMock(side_effect=lambda sid: { "background_classify": classify_data, "scene_save": {"data": {"segments": []}}, "shot_validator": {"data": {"scenes": []}}, "shot_selection": {"data": {"scenes": []}}, "visual_world_rules": {"data": {"rules_text": ""}}, }.get(sid)) step._load_location_profiles = MagicMock(return_value=profiles) return step def test_step_pre_filters_members_without_location_profile(): """A-fix (W21B-w5): profile 없는 member(저빈도 generic location, entity_filter 제거)는 LLM 호출 전 제외 — 같은 그룹의 profiled member 는 보존되어 그룹이 완료된다. 재현: bg_rooftop_villa 가 L05(옥탑방, profile 있음) + L07(빌라 마당, profile 없음)을 묶고 있을 때, 이전엔 L07 fp 가 validate_master_plan_raw_intent 에서 SemanticKeyError → 그룹 통째 유실. pre-filter 로 L07 만 떨구고 L05 는 살린다. """ classify_data = { "data": {"building_groups": [ {"group_id": "bg_z", "kind": "chain_bg", "members": [ {"loc_id": "L05", "label": "interior", "is_indoor": True, "shot_count": 17}, {"loc_id": "L07", "label": "yard", "is_indoor": False, "shot_count": 0}, ]}, ]} } profiles = {"L05": {"kind": "single_space", "allowed_space_keys": ["main"]}} # L07 없음 with patch("app.core.config.settings.background_mode", "on"): step = _make_step(classify_data, profiles) with patch("app.core.steps.background_master_plan_step.run_background_master_plan") as mock_run, \ patch("app.modules.pipeline.background_master_plan.build_master_plan_user_prompt") as mock_prompt: mock_prompt.return_value = "PROMPT" mock_run.return_value = { "group_id": "bg_z", "rationale_summary": "", "floor_plans": [], "backgrounds": [], "gen_order": [], } result = step._execute() # 그룹은 1회 호출됐고 group_loc_ids 에서 L07 제외, L05 유지 assert mock_run.call_count == 1 kwargs = mock_run.call_args.kwargs assert "L07" not in kwargs["group_loc_ids"] assert "L05" in kwargs["group_loc_ids"] # prompt builder 도 필터된 members 만 받음 (LLM 이 L07 fp 생성하지 않도록) member_locs = [m["loc_id"] for m in mock_prompt.call_args.kwargs["group"]["members"]] assert member_locs == ["L05"] # 그룹은 더 이상 SemanticKeyError 로 failed 되지 않음 (A-fix 핵심). # (status 가 ok 가 아닐 수 있는 이유: mock plan 의 backgrounds=[] 라 후속 scope_filter # 가 all_filtered 처리 — pre-filter 와 무관한 별개 단계. 여기선 failed 아님만 확인.) assert result["failed_count"] == 0 assert result["data"]["plans"]["bg_z"]["status"] != "failed" diag = result["data"]["diagnostics"]["location_profile_pre_filter"] assert "bg_z" in diag assert any(fm["loc_id"] == "L07" for fm in diag["bg_z"]["filtered_members"]) def test_step_all_members_filtered_group_skipped_not_failed(): """모든 member 가 profile 부재면 그룹은 skip(all_filtered) — failure 아님, LLM 미호출.""" classify_data = { "data": {"building_groups": [ {"group_id": "bg_q", "kind": "chain_bg", "members": [ {"loc_id": "L07", "label": "yard", "is_indoor": False, "shot_count": 0}, ]}, ]} } profiles = {"L09": {"kind": "single_space", "allowed_space_keys": ["main"]}} # 그룹과 무관 with patch("app.core.config.settings.background_mode", "on"): step = _make_step(classify_data, profiles) with patch("app.core.steps.background_master_plan_step.run_background_master_plan") as mock_run: result = step._execute() assert mock_run.call_count == 0 assert result["failed_count"] == 0 assert result["data"]["plans"]["bg_q"]["status"] == "all_filtered"