| 1 | """Tests for scripts/lib/transcribe.py — caption-free transcription fallback (U6).""" |
| 2 | |
| 3 | from unittest import mock |
| 4 | |
| 5 | from lib import health, transcribe |
| 6 | |
| 7 | |
| 8 | class TestPrerequisites: |
| 9 | def test_missing_ffmpeg_degrades(self): |
| 10 | with mock.patch.object(transcribe.shutil, "which", return_value=None): |
| 11 | result = transcribe.transcribe_media("https://x/v", {"GROQ_API_KEY": "k"}) |
| 12 | assert result.ok is False |
| 13 | assert "ffmpeg" in result.reason |
| 14 | assert result.health.state == health.MISSING |
| 15 | |
| 16 | def test_no_provider_key_degrades(self): |
| 17 | with mock.patch.object(transcribe.shutil, "which", return_value="/usr/bin/ffmpeg"): |
| 18 | result = transcribe.transcribe_media("https://x/v", {}) |
| 19 | assert result.ok is False |
| 20 | assert "provider" in result.reason |
| 21 | assert result.health.state == health.MISSING |
| 22 | |
| 23 | def test_is_available(self): |
| 24 | with mock.patch.object(transcribe.shutil, "which", return_value="/usr/bin/ffmpeg"): |
| 25 | assert transcribe.is_available({"GROQ_API_KEY": "k"}) is True |
| 26 | assert transcribe.is_available({}) is False |
| 27 | |
| 28 | |
| 29 | class TestTranscribeFlow: |
| 30 | def _patches(self, chunks, post_side_effect): |
| 31 | return [ |
| 32 | mock.patch.object(transcribe.shutil, "which", return_value="/usr/bin/ffmpeg"), |
| 33 | mock.patch.object(transcribe, "_acquire_audio", return_value="/tmp/audio.mp3"), |
| 34 | mock.patch.object(transcribe, "_chunk_audio", return_value=chunks), |
| 35 | mock.patch.object(transcribe, "_post_audio", side_effect=post_side_effect), |
| 36 | mock.patch.object(transcribe.shutil, "rmtree"), |
| 37 | mock.patch.object(transcribe.tempfile, "mkdtemp", return_value="/tmp/wd"), |
| 38 | ] |
| 39 | |
| 40 | def test_under_limit_single_chunk(self): |
| 41 | with mock.patch.object(transcribe.shutil, "which", return_value="/usr/bin/ffmpeg"), \ |
| 42 | mock.patch.object(transcribe, "_acquire_audio", return_value="/tmp/audio.mp3"), \ |
| 43 | mock.patch.object(transcribe, "_chunk_audio", return_value=["/tmp/audio.mp3"]), \ |
| 44 | mock.patch.object(transcribe, "_post_audio", return_value="hello world"), \ |
| 45 | mock.patch.object(transcribe.shutil, "rmtree"), \ |
| 46 | mock.patch.object(transcribe.tempfile, "mkdtemp", return_value="/tmp/wd"): |
| 47 | result = transcribe.transcribe_media("https://x/v", {"GROQ_API_KEY": "k"}) |
| 48 | assert result.ok is True |
| 49 | assert result.text == "hello world" |
| 50 | assert result.chunks == 1 |
| 51 | assert result.provider == "groq" |
| 52 | |
| 53 | def test_over_limit_chunks_joined_in_order(self): |
| 54 | chunks = ["/tmp/wd/chunk_000.mp3", "/tmp/wd/chunk_001.mp3"] |
| 55 | with mock.patch.object(transcribe.shutil, "which", return_value="/usr/bin/ffmpeg"), \ |
| 56 | mock.patch.object(transcribe, "_acquire_audio", return_value="/tmp/audio.mp3"), \ |
| 57 | mock.patch.object(transcribe, "_chunk_audio", return_value=chunks), \ |
| 58 | mock.patch.object(transcribe, "_post_audio", side_effect=["part one", "part two"]), \ |
| 59 | mock.patch.object(transcribe.shutil, "rmtree"), \ |
| 60 | mock.patch.object(transcribe.tempfile, "mkdtemp", return_value="/tmp/wd"): |
| 61 | result = transcribe.transcribe_media("https://x/v", {"GROQ_API_KEY": "k"}) |
| 62 | assert result.ok is True |
| 63 | assert result.text == "part one\npart two" |
| 64 | assert result.chunks == 2 |
| 65 | |
| 66 | def test_provider_fallback_on_chunk(self): |
| 67 | # groq raises, openai succeeds -> fallback used. |
| 68 | def post(provider, path, key, timeout): |
| 69 | if provider == "groq": |
| 70 | raise RuntimeError("groq 500") |
| 71 | return "via openai" |
| 72 | with mock.patch.object(transcribe.shutil, "which", return_value="/usr/bin/ffmpeg"), \ |
| 73 | mock.patch.object(transcribe, "_acquire_audio", return_value="/tmp/audio.mp3"), \ |
| 74 | mock.patch.object(transcribe, "_chunk_audio", return_value=["/tmp/audio.mp3"]), \ |
| 75 | mock.patch.object(transcribe, "_post_audio", side_effect=post), \ |
| 76 | mock.patch.object(transcribe.shutil, "rmtree"), \ |
| 77 | mock.patch.object(transcribe.tempfile, "mkdtemp", return_value="/tmp/wd"): |
| 78 | result = transcribe.transcribe_media( |
| 79 | "https://x/v", {"GROQ_API_KEY": "k", "OPENAI_API_KEY": "o"}) |
| 80 | assert result.ok is True |
| 81 | assert result.text == "via openai" |
| 82 | assert result.provider == "openai" |
| 83 | |
| 84 | def test_all_providers_fail_degrades(self): |
| 85 | with mock.patch.object(transcribe.shutil, "which", return_value="/usr/bin/ffmpeg"), \ |
| 86 | mock.patch.object(transcribe, "_acquire_audio", return_value="/tmp/audio.mp3"), \ |
| 87 | mock.patch.object(transcribe, "_chunk_audio", return_value=["/tmp/audio.mp3"]), \ |
| 88 | mock.patch.object(transcribe, "_post_audio", side_effect=RuntimeError("boom")), \ |
| 89 | mock.patch.object(transcribe.shutil, "rmtree"), \ |
| 90 | mock.patch.object(transcribe.tempfile, "mkdtemp", return_value="/tmp/wd"): |
| 91 | result = transcribe.transcribe_media("https://x/v", {"GROQ_API_KEY": "k"}) |
| 92 | assert result.ok is False |
| 93 | assert "all providers failed" in result.reason |
| 94 |