"""LibriVox / Wikimedia Commons voice sources: validation, caps, host checks, mapping, provenance, cache/throttle (HTTP mocked).""" import asyncio import os import sys import tempfile import unittest from pathlib import Path from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parents[1])) os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-vs-test-config-")) from core import open_voice_sources as ovs import routes.sources as src def book(i=1, lang="English", url="https://www.archive.org/download/x/a.mp3"): return {"title": f"Book {i}", "language": lang, "url_librivox": f"https://librivox.org/b{i}/", "authors": [{"first_name": "Jane", "last_name": "Austen"}], "sections": [{"title": "Ch 1", "listen_url": url, "playtime": "600", "readers": [{"display_name": "Ann"}]}]} def commons(**over): info = {"url": "https://upload.wikimedia.org/a.opus", "descriptionurl": "https://commons.wikimedia.org/wiki/File:A.opus", "mediatype": "AUDIO", "duration": 12.4, "extmetadata": {"LicenseShortName": {"value": "CC BY-SA 4.0"}, "Artist": {"value": "Bob & Co"}, "LicenseUrl": {"value": "https://creativecommons.org/licenses/by-sa/4.0"}}} info.update(over) return {"query": {"pages": {"1": {"title": "File:A.opus", "index": 1, "imageinfo": [info]}}}} class Base(unittest.TestCase): def setUp(self): ovs._cache.clear() ovs._last_call.clear() p = mock.patch.object(ovs.time, "sleep") p.start() self.addCleanup(p.stop) class ValidationTests(Base): def test_query_and_lang(self): for bad in ("", " ", "x" * 81): with self.assertRaises(ovs.SourceError): ovs.clean_query(bad) self.assertEqual(ovs.clean_query(" a\x00 \n b "), "a b") with self.assertRaises(ovs.SourceError): ovs.clean_lang("en;rm") with self.assertRaises(ovs.SourceError): ovs.search("evil", "x") def test_host_check(self): self.assertTrue(ovs.host_ok("https://ia80.us.archive.org/x.mp3", ovs.AUDIO_HOSTS)) for bad in ("http://archive.org/x", "https://evilarchive.org/x", "https://archive.org.evil.com/x", "https://u:p@archive.org/x", "https://127.0.0.1/x", "file:///etc/passwd"): self.assertFalse(ovs.host_ok(bad, ovs.AUDIO_HOSTS), bad) class MappingTests(Base): def test_librivox_mapping_filter_cap_and_bad_hosts(self): books = [book(i) for i in range(15)] + [book(99, url="https://evil.example/a.mp3"), book(98, lang="German")] with mock.patch.object(ovs, "_get_json", return_value={"books": books}) as g: items = ovs.search("librivox", "pride", "en") self.assertEqual(len(items), ovs.MAX_RESULTS) self.assertEqual(g.call_args[0][0], ovs.LIBRIVOX_API) it = items[0] self.assertEqual((it["reader"], it["author"], it["language"], it["duration"]), ("Ann", "Jane Austen", "English", 600)) self.assertTrue(it["licence_url"].startswith("https://librivox.org/")) self.assertTrue(all("evil" not in i["audio_url"] and i["language"] == "English" for i in items)) def test_wikimedia_mapping(self): with mock.patch.object(ovs, "_get_json", return_value=commons()): it = ovs.search("wikimedia", "speech")[0] self.assertEqual((it["title"], it["reader"], it["licence"], it["duration"]), ("A", "Bob & Co", "CC BY-SA 4.0", 12)) self.assertIn("Bob & Co / A, CC BY-SA 4.0", it["attribution"]) def test_wikimedia_drops_unsafe_or_unlicensed(self): for over in ({"url": "https://evil.example/a.opus"}, {"mediatype": "VIDEO"}, {"extmetadata": {}}): ovs._cache.clear() with mock.patch.object(ovs, "_get_json", return_value=commons(**over)): self.assertEqual(ovs.search("wikimedia", "speech"), []) class ProvenanceTests(Base): def test_librivox(self): p = ovs.import_provenance_for("librivox", "https://librivox.org/b/", "", "") self.assertEqual(p["basis"], "public_domain") self.assertIn("reader's declared licence", p["note"]) self.assertEqual(p["source"], "https://librivox.org/b/") def test_wikimedia(self): p = ovs.import_provenance_for("wikimedia", "https://commons.wikimedia.org/wiki/File:A.opus", "CC BY 4.0", "Bob / A") self.assertEqual((p["basis"], p["licence"]), ("licensed", "CC BY 4.0")) self.assertIn("Bob / A", p["note"]) self.assertEqual(ovs.import_provenance_for("wikimedia", "https://c.org/", "Public domain", "")["basis"], "public_domain") with self.assertRaises(ValueError): ovs.import_provenance_for("other", "", "", "") class CacheThrottleTests(Base): def test_cache_hit_skips_upstream(self): with mock.patch.object(ovs, "_get_json", return_value={"books": [book()]}) as g: ovs.search("librivox", "Pride"); ovs.search("librivox", "pride") self.assertEqual(g.call_count, 1) def test_throttle_sleeps_and_errors_are_generic(self): resp = mock.Mock(status_code=200) resp.raise_for_status.return_value = None resp.iter_content.return_value = [b'{"books": []}'] with mock.patch.object(ovs.requests, "get", return_value=resp) as g: ovs._get_json(ovs.LIBRIVOX_API, {}); ovs._get_json(ovs.LIBRIVOX_API, {}) self.assertTrue(ovs.time.sleep.called) self.assertFalse(g.call_args.kwargs["allow_redirects"]) self.assertIn("User-Agent", g.call_args.kwargs["headers"]) with mock.patch.object(ovs.requests, "get", side_effect=OSError("secret host")): with self.assertRaises(ovs.SourceError) as cm: ovs._get_json(ovs.COMMONS_API, {}) self.assertNotIn("secret", str(cm.exception)) class RouteTests(Base): def test_search_route_errors_and_allowlist(self): with self.assertRaises(src.HTTPException) as cm: asyncio.run(src.open_voice_source_search("librivox", "")) self.assertEqual(cm.exception.status_code, 400) self.assertIn("archive.org", src._PROXY_AUDIO_DOMAINS) self.assertIn("upload.wikimedia.org", src._PROXY_AUDIO_DOMAINS) def test_quick_import_rejects_foreign_host(self): req = mock.Mock() async def js(): return {"audio_url": "https://example.com/a.mp3", "voice_id": "x", "source_kind": "librivox"} req.json = js with mock.patch.object(src, "_source_import_url", side_effect=lambda u: u): with self.assertRaises(src.HTTPException) as cm: asyncio.run(src.quick_import_voice(req)) self.assertEqual(cm.exception.status_code, 400) if __name__ == "__main__": unittest.main()