| 1 | """Tests for scripts/lib/transcribe.py — caption-free transcription fallback (U6).""" |
| 2 | |
| 3 | from unittest import mock |
| 4 | |
| 5 | from lib import health, http, 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 | |
| 95 | |
| 96 | class TestPostAudioUserAgent: |
| 97 | """_post_audio must send a real User-Agent, not Python-urllib/3.x. |
| 98 | |
| 99 | Groq's Cloudflare edge 403s the default urllib fingerprint with |
| 100 | "error code: 1010" (#983); a plain-urllib UA is a liability on every |
| 101 | provider this helper serves. |
| 102 | """ |
| 103 | |
| 104 | class _Resp: |
| 105 | def __enter__(self): |
| 106 | return self |
| 107 | |
| 108 | def __exit__(self, *exc): |
| 109 | return False |
| 110 | |
| 111 | def read(self): |
| 112 | return b'{"text": "hello"}' |
| 113 | |
| 114 | def _capture_headers(self, tmp_path): |
| 115 | captured = {} |
| 116 | |
| 117 | def _capture(req, timeout=None): |
| 118 | # urllib stores header keys capitalized ("User-agent"). |
| 119 | captured.update({k.lower(): v for k, v in req.headers.items()}) |
| 120 | return self._Resp() |
| 121 | |
| 122 | audio = tmp_path / "audio.mp3" |
| 123 | audio.write_bytes(b"fake-audio") |
| 124 | with mock.patch("urllib.request.urlopen", side_effect=_capture): |
| 125 | transcribe._post_audio("groq", str(audio), "key", timeout=5.0) |
| 126 | return captured |
| 127 | |
| 128 | def test_user_agent_is_canonical_skill_ua(self, tmp_path): |
| 129 | captured = self._capture_headers(tmp_path) |
| 130 | assert "user-agent" in captured |
| 131 | assert captured["user-agent"] == http.USER_AGENT |
| 132 | |
| 133 | def test_user_agent_not_urllib_default(self, tmp_path): |
| 134 | captured = self._capture_headers(tmp_path) |
| 135 | assert not captured["user-agent"].startswith("Python-urllib") |
| 136 | |
| 137 | def test_user_agent_sent_for_openai_too(self, tmp_path): |
| 138 | captured = {} |
| 139 | |
| 140 | def _capture(req, timeout=None): |
| 141 | captured.update({k.lower(): v for k, v in req.headers.items()}) |
| 142 | return self._Resp() |
| 143 | |
| 144 | audio = tmp_path / "audio.mp3" |
| 145 | audio.write_bytes(b"fake-audio") |
| 146 | with mock.patch("urllib.request.urlopen", side_effect=_capture): |
| 147 | transcribe._post_audio("openai", str(audio), "key", timeout=5.0) |
| 148 | assert captured.get("user-agent") == http.USER_AGENT |
| 149 |