"""Tests for visual validation hooks wired into elements.py and step_runner.py. Mocks all critic classes at their lazy-import sites so no Gemini API calls are made. Validates the Phase 1-3 integration from BUILD_SPEC_VALIDATION_INTEGRATION.md. """ import logging from pathlib import Path from unittest.mock import MagicMock, patch import pytest from recoil.core.critic import CriticResult, Dimension, Outcome, Severity from recoil.pipeline._lib.elements import ElementManager # --------------------------------------------------------------------------- # Helpers — build mock CriticResult objects # --------------------------------------------------------------------------- def _passing_result(critic_name: str = "test") -> CriticResult: return CriticResult( critic_name=critic_name, outcome=Outcome.PASS, dimensions=[ Dimension(name="CHECK_A", severity=Severity.HARD, passed=True), ], ) def _failing_result(critic_name: str = "test") -> CriticResult: return CriticResult( critic_name=critic_name, outcome=Outcome.FAIL, dimensions=[ Dimension( name="LIMB_COUNT", severity=Severity.HARD, passed=False, message="Expected 4 limbs, found 6", ), ], ) # --------------------------------------------------------------------------- # Element ref validation (lib/elements.py) # --------------------------------------------------------------------------- class TestElementRefValidation: """Tests for _build_element_entry ref handling in ElementManager. ElementManager no longer accepts validate_refs — ref validation was moved out of the element builder. These tests verify _build_element_entry correctly builds or rejects element entries based on ref availability. """ def _make_manager(self) -> ElementManager: return ElementManager(mode="inline") def _fake_ref(self, tmp_path: Path, name: str = "hero.png") -> Path: """Create a minimal valid PNG so _to_data_uri can read it.""" # Minimal 1x1 white PNG (magic bytes + minimal IHDR) import base64 png_b64 = ( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAC0lEQVQI" "12NgAAIABQABNjN9GQAAAAlwSFlzAAAWJQAAFiUBSVIk8AAAAA0lEQVQI12P4" "z8BQDwAEgAF/QualzQAAAABJRU5ErkJggg==" ) p = tmp_path / name p.write_bytes(base64.b64decode(png_b64)) return p def test_element_ref_validation_skips_bad_ref(self, tmp_path): """_build_element_entry returns None when given empty refs list.""" mgr = self._make_manager() result = mgr._build_element_entry([]) assert result is None, ( "Empty refs should cause _build_element_entry to return None" ) def test_element_ref_validation_passes_good_ref(self, tmp_path): """Valid ref image produces a proper element entry.""" mgr = self._make_manager() ref = self._fake_ref(tmp_path) result = mgr._build_element_entry([ref]) assert result is not None, "Good ref should produce a valid element entry" assert "frontal_image_url" in result assert "reference_image_urls" in result def test_element_ref_validation_disabled(self, tmp_path): """ElementManager builds entries without running any critic.""" mgr = self._make_manager() ref = self._fake_ref(tmp_path) with patch( "recoil.pipeline._lib.critics.ref_image_critic.RefImageCritic", ) as MockClass: result = mgr._build_element_entry([ref]) assert result is not None, "Element should be built without critic" assert "frontal_image_url" in result MockClass.assert_not_called() # --------------------------------------------------------------------------- # StepRunner frame validation (orchestrator/step_runner.py) # --------------------------------------------------------------------------- class TestStepRunnerValidation: """Tests for StartFrameCritic and VideoFrameCritic hooks in execute_video.""" @pytest.fixture def mock_store(self): store = MagicMock() store.update_shot = MagicMock() return store @pytest.fixture def mock_paths(self, tmp_path): paths = MagicMock() paths.video_dir = tmp_path / "video" paths.video_dir.mkdir(parents=True, exist_ok=True) paths.project = "test_project" return paths @pytest.fixture def runner(self, mock_store, mock_paths): from recoil.execution.step_runner import StepRunner return StepRunner( store=mock_store, paths=mock_paths, validate_frames=True, ) def test_start_frame_validation_blocks_and_logs( self, runner, mock_store, tmp_path, caplog ): """Phase 2.5 Task 6: hard-fail start frame BLOCKS video submit and routes to video_semantic_failed (no API call).""" start_frame = tmp_path / "start.png" start_frame.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100) mock_critic = MagicMock() mock_critic.run.return_value = ( str(start_frame), _failing_result("start_frame"), ) mock_client = MagicMock() mock_gen_result = MagicMock() mock_gen_result.success = False mock_gen_result.error = "test abort" mock_gen_result.cost = 0.0 mock_client.wait_for_job.return_value = mock_gen_result with ( patch( "recoil.pipeline._lib.critics.start_frame_critic.StartFrameCritic", return_value=mock_critic, ), patch( "recoil.execution.step_runner.get_client", return_value=mock_client, ), caplog.at_level(logging.ERROR, logger="recoil.execution.step_runner"), ): result = runner.execute_video( shot_id="S01E01_001", prompt="test prompt", model="kling-v3", start_frame=start_frame, ) # Verify ERROR was logged with HARD FAIL error_msgs = [r.message for r in caplog.records if r.levelno == logging.ERROR] assert any("HARD FAIL" in m or "Start frame" in m for m in error_msgs), ( f"Expected 'HARD FAIL' / 'Start frame' error, got: {error_msgs}" ) mock_critic.run.assert_called_once_with(str(start_frame)) # API client should NOT have been called — start frame blocked mock_client.submit.assert_not_called() # Result should be video_semantic_failed assert result.final_state == "video_semantic_failed" def test_video_validation_blocks_and_logs( self, runner, mock_store, tmp_path, caplog ): """Phase 2.5 Task 7: hard-fail video frame BLOCKS pipeline and routes to video_semantic_failed (after generation, before approval).""" mock_vf_critic = MagicMock() mock_vf_critic.run.return_value = ( "fake_video.mp4", _failing_result("video_frame"), ) # Mock a successful video generation so we reach the video validation hook mock_client = MagicMock() mock_gen_result = MagicMock() mock_gen_result.success = True mock_gen_result.error = None mock_gen_result.cost = 0.05 mock_gen_result.video_url = "https://example.com/video.mp4" mock_gen_result.video_bytes = b"fake video data" mock_client.wait_for_job.return_value = mock_gen_result # Mock _save_video to return a fake path saved_video = tmp_path / "video" / "S01E01_002_take001.mp4" saved_video.parent.mkdir(parents=True, exist_ok=True) saved_video.write_bytes(b"fake") with ( patch( "recoil.pipeline._lib.critics.video_frame_critic.VideoFrameCritic", return_value=mock_vf_critic, ), patch( "recoil.execution.video_model_client.VideoModelClient", return_value=mock_client, ), patch.object( runner, "_save_video", return_value=saved_video, ), caplog.at_level(logging.ERROR, logger="recoil.execution.step_runner"), ): result = runner.execute_video( shot_id="S01E01_002", prompt="test prompt", model="kling-v3", ) # Verify ERROR was logged with HARD FAIL error_msgs = [r.message for r in caplog.records if r.levelno == logging.ERROR] assert any("HARD FAIL" in m or "Video" in m for m in error_msgs), ( f"Expected 'HARD FAIL' / 'Video' error, got: {error_msgs}" ) mock_vf_critic.run.assert_called_once_with(str(saved_video)) # Result should be video_semantic_failed assert result.final_state == "video_semantic_failed"