File size: 3,182 Bytes
699f3cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from src.config.settings import MvpRuntimeSettings, OfficialRepoSettings
from src.pipeline.mvp_processor import MvpFrameProcessor, Task3OnlyProcessor
from src.pipeline.orchestrator import ProtocolOrchestrator
from src.pipeline.replay_runner import BatchManifestReplayRunner, ReplayOptions
from src.server.official_repo_batch_adapter import OfficialRepoBatchAdapter
from tests.helpers import running_mock_server
class _ExplodingMatcher:
def __init__(self) -> None:
self.last_run_info: dict[str, object] = {}
def match(self, *args, **kwargs): # type: ignore[no-untyped-def]
del args, kwargs
raise RuntimeError("forced-task3-error")
class Task3ProcessorTests(unittest.TestCase):
def test_replay_smoke_with_task3_only_processor(self) -> None:
with tempfile.TemporaryDirectory() as tmp_dir:
temp_path = Path(tmp_dir)
(temp_path / "ref-001.jpg").write_bytes(b"ref-1")
runtime_settings = MvpRuntimeSettings(task3_reference_dir=temp_path)
with running_mock_server(frame_count=3) as server:
adapter = OfficialRepoBatchAdapter(
OfficialRepoSettings(base_url=server.base_url, username="team", password="password")
)
processor = MvpFrameProcessor(runtime_settings=runtime_settings)
orchestrator = ProtocolOrchestrator(adapter, frame_processor=processor, runtime_settings=runtime_settings)
runner = BatchManifestReplayRunner(adapter, orchestrator)
summary = runner.run(ReplayOptions(max_frames=2, submit_predictions=True))
self.assertEqual(summary.frames_seen, 2)
self.assertEqual(summary.frames_submitted, 2)
def test_task3_failure_still_returns_valid_task3_only_result(self) -> None:
with tempfile.TemporaryDirectory() as tmp_dir:
temp_path = Path(tmp_dir)
(temp_path / "ref-001.jpg").write_bytes(b"ref-1")
runtime_settings = MvpRuntimeSettings(task3_reference_dir=temp_path)
processor = Task3OnlyProcessor(runtime_settings=runtime_settings)
processor.task3_matcher = _ExplodingMatcher() # type: ignore[assignment]
from src.core.frame_state import FrameEnvelope
frame = FrameEnvelope(
frame_url="http://mock/frames/2/",
image_url="/frame.jpg",
video_name="session",
translation_x=1.0,
translation_y=2.0,
translation_z=3.0,
health_status="1",
)
result = processor(frame, b"img")
self.assertEqual(result.diagnostics["task3_status"], "fallback")
self.assertNotIn("task1_status", result.diagnostics)
self.assertNotIn("task2_status", result.diagnostics)
self.assertEqual(result.detected_objects, [])
self.assertEqual(result.detected_translations, [])
self.assertEqual(result.detected_undefined_objects, [])
if __name__ == "__main__":
unittest.main()
|