"""CP-5 Phase 2 — ImageRunner hardening tests (post-build-review absorption). Each test covers exactly one of the 7 bug classes flagged in `consultations/recoil/cp4-post-build-review/gemini_round_1.md` so that future regressions surface as a named failure rather than as a downstream symptom. The cases here are net-new vs the CP-4 `test_image_runner.py` suite — they exercise edges the CP-4 suite does not (NaN cost, Path coercion, success-without-output, KeyboardInterrupt, force-rebind, KeyError UX). """ import math import pathlib import sys from types import SimpleNamespace sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent.parent.parent)) from recoil.core.paths import ensure_pipeline_importable # noqa: E402 ensure_pipeline_importable() import pytest # noqa: E402 from recoil.pipeline.core.registry import ( # noqa: E402 MODALITY_IMAGE_T2I, get_runner, register_runner, ) from recoil.pipeline.core.runners._shared import _failure_metadata_production as _failure_metadata # noqa: E402 from recoil.pipeline.core.runners.image_runner import ( # noqa: E402 ImageRunner, _step_result_to_run_result, ) def _step_result(**overrides): """Build a SimpleNamespace shaped like the real StepResult contract.""" base = dict( shot_id="X", success=True, final_state="keyframe_generated", output_path="/tmp/x.png", cost_usd=0.04, error=None, take_index=0, gate_verdict=None, model="nbp", pipeline="still", ) base.update(overrides) return SimpleNamespace(**base) # ── Bug 1.1 — Path → str coercion ──────────────────────────────────── def test_adapter_coerces_path_to_str(): sr = _step_result(output_path=pathlib.Path("/tmp/p.png")) r = _step_result_to_run_result(sr, "X", MODALITY_IMAGE_T2I) assert isinstance(r.output_path, str) assert r.output_path == "/tmp/p.png" # ── Bug 1.2 — NaN sanitization ─────────────────────────────────────── def test_adapter_sanitizes_nan_cost(): sr = _step_result(cost_usd=float("nan")) r = _step_result_to_run_result(sr, "X", MODALITY_IMAGE_T2I) assert r.metadata["cost_usd"] == 0.0 assert not ( isinstance(r.metadata["cost_usd"], float) and math.isnan(r.metadata["cost_usd"]) ) def test_adapter_handles_none_cost(): sr = _step_result(cost_usd=None) r = _step_result_to_run_result(sr, "X", MODALITY_IMAGE_T2I) assert r.metadata["cost_usd"] == 0.0 # ── Bug 2.1 (REJECTED) — extra payload keys silently dropped by .get() ── def test_run_ignores_unknown_payload_keys(): """Bug 2.1 was rejected — runners use explicit `payload.get("key")`. Extra keys are silently dropped at the dict-lookup boundary; we verify they never reach the StepRunner. NO `inspect.signature` filtering. """ seen: dict = {} class _SR: def execute_keyframe(self, **kw): seen.update(kw) return _step_result() runner = ImageRunner(_SR()) r = runner.run( { "shot_id": "X", "prompt": "p", "model": "nbp", "aspect_ratio": "9_16", "client_id": "should_be_ignored", "webhook_url": "should_be_ignored", "extra_orch_metadata": {"foo": "bar"}, } ) assert "client_id" not in seen assert "webhook_url" not in seen assert "extra_orch_metadata" not in seen assert r.success is True # ── Bug 2.2 — required-key surfacing + metadata always populated ───── def test_run_missing_required_keys_returns_documented_error(): runner = ImageRunner(SimpleNamespace(execute_keyframe=lambda **kw: _step_result())) r = runner.run({"shot_id": "X"}) assert r.success is False assert "missing required keys" in r.error assert r.metadata == _failure_metadata() # ── Bug 6.1 — exception path metadata populated ────────────────────── def test_exception_path_populates_metadata(): class _Boom: def execute_keyframe(self, **kw): raise RuntimeError("upstream API down") r = ImageRunner(_Boom()).run({"shot_id": "X", "prompt": "p", "model": "nbp", "aspect_ratio": "9_16"}) assert r.success is False assert "RuntimeError" in r.error assert "upstream API down" in r.error for key in ( "final_state", "cost_usd", "gate_verdict", "take_index", "model", "pipeline", ): assert key in r.metadata assert r.metadata["final_state"] == "failed" assert r.metadata["cost_usd"] == 0.0 # ── Bug 6.2 — KeyboardInterrupt propagates (BaseException not caught) ── def test_keyboard_interrupt_propagates(): class _Interrupt: def execute_keyframe(self, **kw): raise KeyboardInterrupt() with pytest.raises(KeyboardInterrupt): ImageRunner(_Interrupt()).run({"shot_id": "X", "prompt": "p", "model": "nbp", "aspect_ratio": "9_16"}) # ── Bug 6.3 — success without output coerced to failure ────────────── def test_success_without_output_path_coerced_to_failure(): sr = _step_result(success=True, output_path=None) r = _step_result_to_run_result(sr, "X", MODALITY_IMAGE_T2I) assert r.success is False assert "no output_path" in (r.error or "") def test_success_without_output_path_preserves_existing_error(): """If StepRunner already attached an error string, keep it — don't overwrite.""" sr = _step_result(success=True, output_path=None, error="upstream timeout") r = _step_result_to_run_result(sr, "X", MODALITY_IMAGE_T2I) assert r.success is False assert r.error == "upstream timeout" # ── Bug 3.1 — force=True allows rebinding ──────────────────────────── def test_register_runner_force_rebinds(): class _A: modality = "test_x" def run(self, p): pass class _B: modality = "test_x" def run(self, p): pass register_runner("test_x", _A()) with pytest.raises(KeyError): register_runner("test_x", _B()) # force=False default register_runner("test_x", _B(), force=True) assert get_runner("test_x").__class__.__name__ == "_B" def test_register_runner_idempotent_same_instance_without_force(): """Re-registering the SAME instance is allowed even with force=False.""" class _A: modality = "test_y" def run(self, p): pass inst = _A() register_runner("test_y", inst) register_runner("test_y", inst) # no raise — same instance assert get_runner("test_y") is inst # ── KeyError UX (Bug 5.1) — bootstrap hint present ─────────────────── def test_get_runner_keyerror_mentions_bootstrap(): with pytest.raises(KeyError) as exc: get_runner("nonexistent") msg = str(exc.value) assert "nonexistent" in msg assert "register_runner" in msg or "register_default_runners" in msg