"""W20E2: shot-aware BG readiness report tests. Aggregator over W20E1 preflight + operator-supplied target facts. Pure no-network, no-write, no-LLM, no-VLM, no-image-gen tests. """ from __future__ import annotations import ast import inspect import json from types import SimpleNamespace from app.modules.pipeline.shot_aware_bg_preflight import ( SCOPE_FULL_FRESH_PROJECT_E2E, SCOPE_SHOT_AWARE_BG_E2E, SCOPE_SHOT_AWARE_BG_PLAN_ONLY, ShotAwareBgPreflightRequest, evaluate_shot_aware_bg_preflight, ) from app.modules.pipeline.shot_aware_bg_readiness import ( ShotAwareBgReadinessTarget, build_shot_aware_bg_readiness_report, ) # ───────────────────────── helpers ───────────────────────── def _settings_at_shot_aware_happy_path(**overrides) -> SimpleNamespace: base = dict( background_mode="on", floor_plan_prompt_version="6", background_prompt_version="7", background_render_reference_mode="shot_aware_plan", base_location_dossier_enabled=True, floor_plan_geometry_readback_enabled=True, shot_aware_bg_render_plan_enabled=True, floor_plan_vlm_readback_real_provider_enabled=False, shot_aware_bg_render_plan_real_provider_enabled=False, ) base.update(overrides) return SimpleNamespace(**base) def _request_plan_only(**overrides) -> ShotAwareBgPreflightRequest: base = dict( requested_scope=SCOPE_SHOT_AWARE_BG_PLAN_ONLY, approve_real_vlm=False, approve_real_shot_aware_planner_llm=False, approve_image_generation=False, image_call_cap=0, approve_full_fresh_project_e2e=False, ) base.update(overrides) return ShotAwareBgPreflightRequest(**base) def _request_e2e_with_image(**overrides) -> ShotAwareBgPreflightRequest: base = dict( requested_scope=SCOPE_SHOT_AWARE_BG_E2E, approve_real_vlm=False, approve_real_shot_aware_planner_llm=False, approve_image_generation=True, image_call_cap=3, approve_full_fresh_project_e2e=False, ) base.update(overrides) return ShotAwareBgPreflightRequest(**base) def _request_full_with_image(**overrides) -> ShotAwareBgPreflightRequest: base = dict( requested_scope=SCOPE_FULL_FRESH_PROJECT_E2E, approve_real_vlm=False, approve_real_shot_aware_planner_llm=False, approve_image_generation=True, image_call_cap=10, approve_full_fresh_project_e2e=True, ) base.update(overrides) return ShotAwareBgPreflightRequest(**base) # ───────────────────────── happy path ───────────────────────── def test_plan_only_happy_path_report_is_json_serializable(): report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_plan_only(), ) assert report["ok"] is True, report assert report["preflight"]["failures"] == [] assert report["readiness_blockers"] == [] serialized = json.dumps(report) restored = json.loads(serialized) assert restored == report def test_real_api_call_and_write_counts_always_zero(): report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_plan_only(), ) assert report["real_api_call_counts"] == { "image": 0, "llm": 0, "vlm": 0, } assert report["write_counts"] == { "db": 0, "image_asset": 0, "projects_checkpoint": 0, } # ───────────────────────── preflight failures aggregate ───────────────────────── def test_e2e_with_default_approvals_fails_through_preflight(): """e2e scope with image_generation NOT approved must fail closed via W20E1 preflight (surfaced as preflight failure, not a readiness blocker — those are target-fact related).""" request = ShotAwareBgPreflightRequest( requested_scope=SCOPE_SHOT_AWARE_BG_E2E, approve_real_vlm=False, approve_real_shot_aware_planner_llm=False, approve_image_generation=False, image_call_cap=0, approve_full_fresh_project_e2e=False, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=request, ) assert report["ok"] is False pf_codes = [f["code"] for f in report["preflight"]["failures"]] assert "image_generation_not_approved_for_e2e" in pf_codes def test_full_with_default_approvals_fails_through_preflight(): request = ShotAwareBgPreflightRequest( requested_scope=SCOPE_FULL_FRESH_PROJECT_E2E, approve_real_vlm=False, approve_real_shot_aware_planner_llm=False, approve_image_generation=False, image_call_cap=0, approve_full_fresh_project_e2e=False, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=request, ) assert report["ok"] is False pf_codes = [f["code"] for f in report["preflight"]["failures"]] assert "full_fresh_project_e2e_out_of_scope" in pf_codes def test_preflight_failures_serialized_as_plain_dicts(): request = ShotAwareBgPreflightRequest( requested_scope=SCOPE_SHOT_AWARE_BG_E2E, approve_real_vlm=False, approve_real_shot_aware_planner_llm=False, approve_image_generation=False, image_call_cap=0, approve_full_fresh_project_e2e=False, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=request, ) for f in report["preflight"]["failures"]: assert set(f.keys()) == {"code", "message", "selector_path"} assert isinstance(f["code"], str) assert isinstance(f["message"], str) assert f["selector_path"] is None or isinstance( f["selector_path"], str ) # ───────────────────────── target readiness blockers ───────────────────────── def test_target_planning_doc_missing_creates_blocker_for_e2e(): target = ShotAwareBgReadinessTarget( project_id="p-001", episode_id="e-002", planning_doc_present=False, planning_doc_checkpoint_present=True, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_e2e_with_image(), target=target, ) assert report["ok"] is False blocker_codes = [b["code"] for b in report["readiness_blockers"]] assert "planning_doc_missing_for_e2e_scope" in blocker_codes assert report["target"]["project_id"] == "p-001" assert report["target"]["episode_id"] == "e-002" assert report["target"]["planning_doc_present"] is False assert report["target"]["planning_doc_checkpoint_present"] is True def test_target_planning_doc_missing_creates_blocker_for_full(): target = ShotAwareBgReadinessTarget( project_id="p-001", episode_id="e-002", planning_doc_present=False, planning_doc_checkpoint_present=True, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_full_with_image(), target=target, ) assert report["ok"] is False blocker_codes = [b["code"] for b in report["readiness_blockers"]] assert "planning_doc_missing_for_e2e_scope" in blocker_codes def test_target_planning_doc_checkpoint_missing_creates_blocker_for_e2e(): target = ShotAwareBgReadinessTarget( project_id="p-001", episode_id="e-002", planning_doc_present=True, planning_doc_checkpoint_present=False, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_e2e_with_image(), target=target, ) assert report["ok"] is False blocker_codes = [b["code"] for b in report["readiness_blockers"]] assert "planning_doc_checkpoint_missing_for_e2e_scope" in blocker_codes def test_target_facts_do_not_create_blockers_for_plan_only_scope(): """plan-only scope does not consume the planning doc; missing-doc facts must not create blockers there (those facts are only meaningful for the image-generating render scopes).""" target = ShotAwareBgReadinessTarget( project_id="p-001", episode_id="e-002", planning_doc_present=False, planning_doc_checkpoint_present=False, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_plan_only(), target=target, ) assert report["ok"] is True, report assert report["readiness_blockers"] == [] def test_no_target_means_no_blockers_and_target_dict_has_none_fields(): """When target is omitted, no readiness blockers fire (operator hasn't claimed the facts either way) and the report's target dict reports None per field so downstream consumers can detect 'unspecified' instead of confusing it with False.""" report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_e2e_with_image(), ) assert report["readiness_blockers"] == [] assert report["target"] == { "project_id": None, "episode_id": None, "planning_doc_present": None, "planning_doc_checkpoint_present": None, } def test_target_planning_doc_present_true_creates_no_blocker(): target = ShotAwareBgReadinessTarget( project_id="p-001", episode_id="e-002", planning_doc_present=True, planning_doc_checkpoint_present=True, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_e2e_with_image(), target=target, ) assert report["ok"] is True, report assert report["readiness_blockers"] == [] def test_multiple_readiness_blockers_aggregate_without_raising(): target = ShotAwareBgReadinessTarget( project_id="p-001", episode_id="e-002", planning_doc_present=False, planning_doc_checkpoint_present=False, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_e2e_with_image(), target=target, ) assert report["ok"] is False blocker_codes = [b["code"] for b in report["readiness_blockers"]] assert "planning_doc_missing_for_e2e_scope" in blocker_codes assert "planning_doc_checkpoint_missing_for_e2e_scope" in blocker_codes def test_readiness_blockers_serialized_as_plain_dicts(): target = ShotAwareBgReadinessTarget( planning_doc_present=False, planning_doc_checkpoint_present=False, ) report = build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_e2e_with_image(), target=target, ) for b in report["readiness_blockers"]: assert set(b.keys()) == {"code", "message"} assert isinstance(b["code"], str) assert isinstance(b["message"], str) and b["message"] # ───────────────────────── budget_summary preserved exactly ───────────────────────── def test_budget_summary_preserved_exactly_from_w20e1(): """W20E2 must not reimplement budget math; it must pass the W20E1 budget_summary through unchanged.""" settings = _settings_at_shot_aware_happy_path() request = _request_e2e_with_image() expected = evaluate_shot_aware_bg_preflight( settings=settings, request=request ).budget_summary report = build_shot_aware_bg_readiness_report( settings=settings, request=request ) assert report["preflight"]["budget_summary"] == expected # ───────────────────────── W20E1 called exactly once ───────────────────────── def test_w20e1_preflight_called_exactly_once(monkeypatch): """The aggregator must call W20E1's evaluator exactly once — no retry, no re-entry. Patching the module-level binding catches a second call.""" from app.modules.pipeline import shot_aware_bg_readiness as mod call_count = {"n": 0} original = mod.evaluate_shot_aware_bg_preflight def counting(*, settings, request): call_count["n"] += 1 return original(settings=settings, request=request) monkeypatch.setattr(mod, "evaluate_shot_aware_bg_preflight", counting) build_shot_aware_bg_readiness_report( settings=_settings_at_shot_aware_happy_path(), request=_request_e2e_with_image(), target=ShotAwareBgReadinessTarget( planning_doc_present=True, planning_doc_checkpoint_present=True, ), ) assert call_count["n"] == 1 # ───────────────────────── purity / no-real-provider safety ───────────────────────── def _collect_imported_module_names(module) -> list[str]: src = inspect.getsource(module) tree = ast.parse(src) names: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: names.append(alias.name) elif isinstance(node, ast.ImportFrom): if node.module: names.append(node.module) return names def test_module_imports_no_forbidden_provider_libs(): import app.modules.pipeline.shot_aware_bg_readiness as mod imported = {n.split(".")[0] for n in _collect_imported_module_names(mod)} forbidden = { "litellm", "openai", "fal_client", "fal", "google", "anthropic", "PIL", "requests", "httpx", } leaked = imported & forbidden assert not leaked, ( f"shot_aware_bg_readiness imports forbidden module(s) " f"{sorted(leaked)} — must stay pure" ) def test_module_only_app_import_is_w20e1_preflight(): """The only ``app.*`` import allowed is W20E1's preflight module. Anything else risks pulling provider code in transitively.""" import app.modules.pipeline.shot_aware_bg_readiness as mod imported = _collect_imported_module_names(mod) app_imports = [n for n in imported if n.startswith("app")] assert app_imports == [ "app.modules.pipeline.shot_aware_bg_preflight" ], ( f"shot_aware_bg_readiness must import only the W20E1 preflight " f"module from app.*, got {app_imports!r}" )