diff options
| author | historia <historiavg@proton.me> | 2026-08-24 06:04:38 -0400 |
|---|---|---|
| committer | historia <historiavg@proton.me> | 2026-08-24 06:04:38 -0400 |
| commit | 4db8ea7a63107297450819d227497ebbb121ff38 (patch) | |
| tree | 250b646195f98a4b4c42b04589c0ca8951d13fbb /app/tests/test_backends_audiocpp.py | |
| parent | 4f3d16e9d532efd6a6bda283925efeb8c93b51c7 (diff) | |
| download | tts-audiobook-generator-4db8ea7a63107297450819d227497ebbb121ff38.tar.gz | |
fix: don't expect/use local config for remote servers
Diffstat (limited to 'app/tests/test_backends_audiocpp.py')
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 90 |
1 files changed, 90 insertions, 0 deletions
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index 9882ce1..e2b09d0 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -1058,5 +1058,95 @@ class NonInteractiveMainTests(unittest.TestCase): self.assertEqual(ctx.exception.code, 2) +class FetchServerEndpointsTests(unittest.TestCase): + """fetch_server_models / fetch_server_voices: live queries against a + running audiocpp_server (urlopen mocked).""" + + @staticmethod + def _urlopen_responding(bodies, errors=None): + """A urlopen stub returning successive BODIES (bytes) or raising + successive ERRORS; records every requested URL.""" + calls = [] + + def fake_urlopen(url, timeout=10): + calls.append(url) + if errors: + raise errors.pop(0) + body = bodies.pop(0) + context = MagicMock() + context.__enter__.return_value = context + context.__exit__.return_value = False + context.read.return_value = body + return context + + return fake_urlopen, calls + + def test_fetch_models_parses_id_family_task(self): + urlopen, calls = self._urlopen_responding([json.dumps({ + "data": [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}, + {"id": "legacy"}], + }).encode("utf-8")]) + with patch.object(make_server.urllib.request, "urlopen", urlopen): + models = make_server.fetch_server_models("http://127.0.0.1:8080") + # Missing fields mirror the converter's client: empty strings. + self.assertEqual(models, [ + {"id": "qwen", "family": "qwen3_tts", "task": "tts"}, + {"id": "legacy", "family": "", "task": ""}, + ]) + self.assertEqual(calls, ["http://127.0.0.1:8080/v1/models"]) + + def test_fetch_models_trailing_slash_url(self): + urlopen, calls = self._urlopen_responding( + [b'{"data": [{"id": "m"}]}']) + with patch.object(make_server.urllib.request, "urlopen", urlopen): + make_server.fetch_server_models("http://host:8080/") + self.assertEqual(calls, ["http://host:8080/v1/models"]) + + def test_fetch_models_connection_error_returns_none(self): + import urllib.error + urlopen, _ = self._urlopen_responding( + [], errors=[urllib.error.URLError("Connection refused")]) + with patch.object(make_server.urllib.request, "urlopen", urlopen): + self.assertIsNone( + make_server.fetch_server_models("http://127.0.0.1:8080")) + + def test_fetch_models_non_json_body_returns_none(self): + # A port answering TCP but not speaking audiocpp_server JSON. + urlopen, _ = self._urlopen_responding([b"<html>not json</html>"]) + with patch.object(make_server.urllib.request, "urlopen", urlopen): + self.assertIsNone( + make_server.fetch_server_models("http://127.0.0.1:8080")) + + def test_fetch_models_unexpected_document_yields_empty_list(self): + urlopen, _ = self._urlopen_responding([b'{"foo": 1}']) + with patch.object(make_server.urllib.request, "urlopen", urlopen): + self.assertEqual( + make_server.fetch_server_models("http://127.0.0.1:8080"), []) + + def test_fetch_voices_parses_names_and_encodes_model(self): + urlopen, calls = self._urlopen_responding( + [b'{"voices": ["narrator", "obama"]}']) + with patch.object(make_server.urllib.request, "urlopen", urlopen): + voices = make_server.fetch_server_voices( + "http://127.0.0.1:8080", "qwen") + self.assertEqual(voices, ["narrator", "obama"]) + self.assertEqual(calls, + ["http://127.0.0.1:8080/v1/audio/voices?model=qwen"]) + + def test_fetch_voices_error_returns_none(self): + import urllib.error + urlopen, _ = self._urlopen_responding( + [], errors=[urllib.error.URLError("boom")]) + with patch.object(make_server.urllib.request, "urlopen", urlopen): + self.assertIsNone( + make_server.fetch_server_voices("http://h", "qwen")) + + def test_fetch_voices_non_list_shape_returns_none(self): + urlopen, _ = self._urlopen_responding([b'{"voices": 5}']) + with patch.object(make_server.urllib.request, "urlopen", urlopen): + self.assertIsNone( + make_server.fetch_server_voices("http://h", "qwen")) + + if __name__ == "__main__": unittest.main() |