"""End-to-end mock pipeline test: exercises the full worker flow without real APIs.""" import asyncio import os import pytest import tempfile from pathlib import Path os.environ["MOCK_MODE"] = "true" from src.config import EnvConfig from src.db import MockDB, VideoTask, WorkerStats from src.r2_client import R2Client from src.providers.aistudio import AIStudioProvider from src.providers.openrouter import OpenRouterProvider from src.pipeline import Pipeline from src.batch_cycle import BatchCycleEngine from src.providers.base import TranscriptionRequest class TestMockPipelineE2E: """Full end-to-end test using mock providers and mock R2.""" def _make_config(self) -> EnvConfig: return EnvConfig( mock_mode=True, worker_id="test-worker-001", gpu_type="mock-gpu", ) def test_full_pipeline_single_video(self): config = self._make_config() db = MockDB() db.seed_videos([VideoTask(video_id="mock_video_001", language="te", segment_count=5)]) r2 = R2Client(config) primary = AIStudioProvider(api_key="fake-key-1", mock_mode=True) fallback = AIStudioProvider(api_key="fake-key-2", mock_mode=True) stats = WorkerStats() pipeline = Pipeline( config=config, db=db, r2=r2, primary_provider=primary, fallback_provider=fallback, worker_id="test-worker-001", stats=stats, ) task = asyncio.get_event_loop().run_until_complete( db.claim_video("test-worker-001") ) assert task is not None assert task.video_id == "mock_video_001" success = asyncio.get_event_loop().run_until_complete( pipeline.process_video(task) ) assert success is True assert stats.segments_sent > 0 assert stats.segments_completed > 0 assert stats.batches_completed >= 1 assert db.video_queue[0].status == "done" assert len(db.results) > 0 def test_pipeline_no_fallback(self): config = self._make_config() db = MockDB() db.seed_videos([VideoTask(video_id="mock_video_002", language="hi")]) r2 = R2Client(config) primary = AIStudioProvider(api_key="fake-key-1", mock_mode=True) stats = WorkerStats() pipeline = Pipeline( config=config, db=db, r2=r2, primary_provider=primary, fallback_provider=None, worker_id="test-worker-002", stats=stats, ) task = asyncio.get_event_loop().run_until_complete( db.claim_video("test-worker-002") ) success = asyncio.get_event_loop().run_until_complete( pipeline.process_video(task) ) assert success is True assert stats.segments_completed > 0 def test_pipeline_no_videos_available(self): config = self._make_config() db = MockDB() # empty queue task = asyncio.get_event_loop().run_until_complete( db.claim_video("test-worker") ) assert task is None class TestMockBatchCycle: def test_batch_cycle_with_mock_provider(self): primary = AIStudioProvider(api_key="fake", mock_mode=True) engine = BatchCycleEngine( primary_provider=primary, fallback_provider=None, worker_id="test-worker", video_id="test-video", ) requests = [ TranscriptionRequest( segment_id=f"seg_{i:03d}", audio_base64="AAAA", language_code="te", original_file=f"seg_{i:03d}.flac", ) for i in range(10) ] result = asyncio.get_event_loop().run_until_complete( engine.run_batch( requests=requests, expected_language="te", audio_durations={f"seg_{i:03d}": 5.0 for i in range(10)}, trim_metas={}, ) ) assert result.segments_sent == 10 assert result.segments_returned == 10 assert result.segments_error == 0 assert result.cache_hits > 0 assert len(result.transcription_records) == 10 assert all(r["quality_score"] > 0 for r in result.transcription_records) def test_batch_cycle_with_fallback(self): primary = AIStudioProvider(api_key="fake", mock_mode=True) fallback = OpenRouterProvider(api_key="fake", mock_mode=True) engine = BatchCycleEngine( primary_provider=primary, fallback_provider=fallback, worker_id="test-worker", video_id="test-video", ) requests = [ TranscriptionRequest( segment_id="seg_001", audio_base64="AAAA", language_code="te", original_file="seg_001.flac", ) ] result = asyncio.get_event_loop().run_until_complete( engine.run_batch(requests, "te", {"seg_001": 5.0}, {}) ) assert result.segments_returned == 1 class TestMockDBOperations: def test_worker_registration(self): db = MockDB() asyncio.get_event_loop().run_until_complete( db.register_worker("w1", "aistudio", "RTX 4090", {"version": "1.0"}) ) assert "w1" in db.workers assert db.workers["w1"]["status"] == "online" def test_heartbeat_update(self): db = MockDB() asyncio.get_event_loop().run_until_complete( db.register_worker("w1", "aistudio", "RTX 4090", {}) ) stats = WorkerStats(segments_sent=100, segments_completed=95) asyncio.get_event_loop().run_until_complete( db.update_heartbeat("w1", stats) ) assert db.workers["w1"]["total_segments_sent"] == 100 def test_claim_and_release(self): db = MockDB() db.seed_videos([VideoTask(video_id="v1", language="te")]) task = asyncio.get_event_loop().run_until_complete(db.claim_video("w1")) assert task.video_id == "v1" assert db.video_queue[0].status == "claimed" asyncio.get_event_loop().run_until_complete(db.release_video("v1")) assert db.video_queue[0].status == "pending" def test_mark_done(self): db = MockDB() db.seed_videos([VideoTask(video_id="v1", language="te")]) asyncio.get_event_loop().run_until_complete(db.claim_video("w1")) asyncio.get_event_loop().run_until_complete(db.mark_video_done("v1")) assert db.video_queue[0].status == "done" def test_worker_offline(self): db = MockDB() asyncio.get_event_loop().run_until_complete( db.register_worker("w1", "aistudio", "RTX 4090", {}) ) asyncio.get_event_loop().run_until_complete(db.set_worker_offline("w1")) assert db.workers["w1"]["status"] == "offline" class TestMockR2: def test_mock_tar_creation_and_extraction(self): config = EnvConfig(mock_mode=True) r2 = R2Client(config) with tempfile.TemporaryDirectory() as tmpdir: work_dir = Path(tmpdir) tar_path = r2.download_tar("test_video", work_dir) assert tar_path.exists() extracted = r2.extract_tar(tar_path, "test_video") assert extracted.video_id == "test_video" assert len(extracted.segment_paths) > 0 assert extracted.metadata.get("video_id") == "test_video" def test_mock_pack_and_upload(self): config = EnvConfig(mock_mode=True) r2 = R2Client(config) with tempfile.TemporaryDirectory() as tmpdir: work_dir = Path(tmpdir) tar_path = r2.download_tar("test_video", work_dir) extracted = r2.extract_tar(tar_path, "test_video") result_tar = r2.pack_results_tar( video_id="test_video", work_dir=extracted.work_dir, segment_paths=extracted.segment_paths, transcription_jsons={"seg1.flac": {"transcription": "test"}}, metadata=extracted.metadata, ) assert result_tar.exists() # Upload should not raise in mock mode r2.upload_tar(result_tar, "test_video")