import json from pathlib import Path import subprocess import sys import time import unittest from fastapi.testclient import TestClient import app from tests.fakes import AUDIO, BrokenWorker, FakeWorker, ProgressiveWorker class ApiTests(unittest.TestCase): def setUp(self): self.application = app.create_app(service_options={"worker_factory": FakeWorker}) self.context = TestClient(self.application) self.client = self.context.__enter__() for _ in range(100): if self.client.get("/tts/health").status_code == 200: break time.sleep(0.002) def tearDown(self): self.context.__exit__(None, None, None) def ready_job(self): response = self.client.post("/tts/jobs", json={"text": "这是正文。"}) self.assertEqual(response.status_code, 202) id = response.json()["id"] for _ in range(100): row = self.client.get("/tts/jobs/" + id).json() if row["state"] == "ready": return row time.sleep(0.002) self.fail("Task did not become ready") def test_original_quote_contract_and_arbitrary_origin_are_preserved(self): sample = app.DATA[0] id = str(sample["id"]) expected = app.format_quote(sample) response = self.client.get("/", params={"id": id, "c": "invalid"}, headers={"Origin": "https://unrelated.test"}) self.assertEqual(response.status_code, 200) self.assertEqual(response.text, expected) self.assertTrue(response.headers["content-type"].startswith("text/plain")) self.assertEqual(response.headers["access-control-allow-origin"], "*") self.assertEqual(self.client.get("/?id=not-an-id").status_code, 404) self.assertEqual(self.client.get("/?c=invalid").status_code, 404) for path in ("/?c=mx&c=l", "/?c=mx,l"): self.assertIn(self.client.get(path).text, {app.format_quote(item) for item in app.DATA if set(item["cat"]) & {"mx", "l"}}) def test_tts_cross_origin_preflight_and_origin_rejection(self): for origin in ("https://vomebook.github.io", "https://vomebook-search.hf.space", "https://vomebook-bha-search.hf.space", "https://voiceofml-search.hf.space"): response = self.client.options("/tts/jobs", headers={"Origin": origin, "Access-Control-Request-Method": "POST", "Access-Control-Request-Headers": "Content-Type"}) self.assertEqual(response.status_code, 200) self.assertEqual(response.headers["access-control-allow-origin"], origin) response = self.client.post("/tts/jobs", json={"text": "正文"}, headers={"Origin": "https://unrelated.test"}) self.assertEqual(response.status_code, 403) self.assertEqual(self.application.state.tts.health()["queued"], 0) def test_invalid_requests_and_large_bodies_fail_without_text_echo(self): for data in ({"text": "文" * 801}, {"text": "正文", "voice": "../model"}, {"text": "正文", "url": "https://example.test"}, {"text": "正文", "speed": 4}): response = self.client.post("/tts/jobs", json=data) self.assertEqual(response.status_code, 422) self.assertNotIn("正文", response.text.replace("正文长度", "")) self.assertEqual(self.client.post("/tts/jobs", content="x" * 20000, headers={"Content-Type": "application/json"}).status_code, 413) self.assertEqual(self.client.post("/tts/jobs", content="plain text").status_code, 415) def test_job_audio_head_range_and_cancel(self): row = self.ready_job() path = row["result"]["audio_url"] response = self.client.get(path) self.assertEqual(response.content, AUDIO) self.assertEqual(response.headers["content-type"], "audio/mpeg") self.assertEqual(int(response.headers["content-length"]), len(AUDIO)) self.assertNotIn("content-encoding", response.headers) self.assertEqual(self.client.head(path).content, b"") ranged = self.client.get(path, headers={"Range": "bytes=2-7"}) self.assertEqual(ranged.status_code, 206) self.assertEqual(ranged.content, AUDIO[2:8]) self.assertEqual(ranged.headers["content-range"], f"bytes 2-7/{len(AUDIO)}") self.assertEqual(next(iter(self.application.state.tts.cache.values())).leases, 0) self.client.delete("/tts/jobs/" + row["id"]) self.assertEqual(self.client.get(path).status_code, 410) def test_unknown_jobs_and_unfinished_audio_are_not_successful_files(self): self.assertEqual(self.client.get("/tts/jobs/invalid").status_code, 404) self.assertEqual(self.client.get("/tts/audio/" + "a" * 32 + ".mp3").status_code, 404) def test_model_failure_does_not_break_quote_startup(self): with TestClient(app.create_app(service_options={"worker_factory": BrokenWorker})) as client: self.assertEqual(client.get("/").status_code, 200) self.assertEqual(client.get("/tts/health").status_code, 503) self.assertEqual(client.post("/tts/jobs", json={"text": "正文"}).status_code, 503) def test_static_audition_and_capabilities(self): self.assertEqual(self.client.get("/tts/").status_code, 200) self.assertEqual(self.client.get("/tts/app.js").status_code, 200) self.assertEqual(self.client.get("/tts/embed.js").status_code, 200) embedded = self.client.get("/tts/?embed=ccrd-v1") self.assertEqual(embedded.headers["content-security-policy"], "frame-ancestors https://vomebook-search.hf.space") voices = self.client.get("/tts/voices").json() self.assertEqual(len(voices["voices"]), 4) self.assertTrue(voices["pitch"]) self.assertEqual((voices["pitch_semitones"]["min"], voices["pitch_semitones"]["max"]), (-3, 3)) self.assertEqual(voices["alignment"], "segment-approximate") schema = self.client.get("/tts/openapi.json").json() operations = [operation["operationId"] for path in schema["paths"].values() for operation in path.values()] self.assertEqual(len(operations), len(set(operations))) def test_api_import_does_not_import_inference_libraries(self): result = subprocess.check_output([sys.executable, "-B", "-c", "import app,sys,json;print(json.dumps([name for name in ('numpy','onnxruntime','misaki','lameenc','transformers','cn2an','pedalboard') if name in sys.modules]))"], cwd=Path(__file__).resolve().parents[1], text=True) self.assertEqual(json.loads(result), []) def test_running_part_head_range_and_cancel_without_complete_file(self): with TestClient(app.create_app(service_options={"worker_factory": ProgressiveWorker})) as client: for _ in range(100): if client.get("/tts/health").status_code == 200: break time.sleep(.002) body = {"text": "先播第一段再播放第二段", "progressive": True, "request_id": "p" * 32} created = client.post("/tts/jobs", json=body).json() self.assertEqual(created["source_segments"], [{"start": 0, "end": len(body["text"])}]) self.assertNotIn(body["text"], str(created)) self.assertEqual(client.post("/tts/jobs", json=body).json()["id"], created["id"]) path = "/tts/jobs/" + created["id"] for _ in range(100): row = client.get(path).json() if row.get("progress", {}).get("parts"): break time.sleep(.002) self.assertEqual(row["state"], "running") self.assertEqual(client.get("/tts/audio/" + created["id"] + ".mp3").status_code, 410) clip = row["progress"]["parts"][0]["audio_url"] self.assertEqual(client.get(clip).content, AUDIO) self.assertEqual(client.head(clip).content, b"") ranged = client.get(clip, headers={"Range": "bytes=2-7", "Origin": "https://vomebook.github.io"}) self.assertEqual(ranged.status_code, 206) self.assertEqual(ranged.content, AUDIO[2:8]) self.assertEqual(ranged.headers["access-control-allow-origin"], "https://vomebook.github.io") self.assertEqual(client.get(clip.replace("/0.mp3", "/63.mp3")).status_code, 410) client.delete(path) self.assertEqual(client.get(clip).status_code, 410) def test_context_and_request_retry_metadata_are_bounded(self): for extra in ({"context_before": "甲" * 33}, {"context_after": "\0"}, {"progressive": 1}, {"request_id": "bad"}, *({"pitch_semitones": value} for value in (-4, 4, .5, True, "2", None))): self.assertEqual(self.client.post("/tts/jobs", json={"text": "长文", **extra}).status_code, 422) def test_unknown_pause_policy_is_opt_in_bounded_and_retry_identity_owned(self): self.assertEqual(self.client.post("/tts/jobs", json={"text": "正文", "unsupported_policy": "anything"}).status_code, 422) self.assertEqual(self.client.post("/tts/jobs", json={"text": "😀※"}).status_code, 422) body = {"text": "😀※\f正文", "unsupported_policy": "pause", "progressive": True, "request_id": "u" * 48} created = self.client.post("/tts/jobs", json=body) self.assertEqual(created.status_code, 202) id = created.json()["id"] self.assertEqual(self.client.post("/tts/jobs", json=body).json()["id"], id) self.assertEqual(self.client.post("/tts/jobs", json={**body, "text": "正文", "unsupported_policy": "error"}).status_code, 409) for _ in range(100): if self.application.state.tts.worker.submissions: break time.sleep(.002) self.assertEqual(self.application.state.tts.worker.submissions[0]["unsupported_policy"], "pause") self.assertEqual(self.client.post("/tts/jobs", json={"text": "文" * 801, "unsupported_policy": "pause"}).status_code, 422) def test_custom_word_segments_are_exact_and_wait_first_returns_a_published_part(self): with TestClient(app.create_app(service_options={"worker_factory": ProgressiveWorker})) as client: for _ in range(100): if client.get("/tts/health").status_code == 200: break time.sleep(.002) body = {"text": "校长读长文，𠮷后文继续。", "progressive": True, "segment_ends": [6, 13], "unsupported_policy": "pause", "wait_first": True, "request_id": "w" * 48} self.assertEqual(len(body["text"].encode("utf-16-le")) // 2, 13) response = client.post("/tts/jobs", json=body) self.assertEqual(response.status_code, 202) row = response.json() self.assertTrue(row.get("progress", {}).get("parts")) self.assertEqual(row["source_segments"], [{"start": 0, "end": 6}, {"start": 6, "end": 13}]) self.assertEqual(row["progress"]["parts"][0]["end"], 6) # Long-poll returns once a further part is published or the job ends. polled = client.get(f"/tts/jobs/{row['id']}?after=1").json() self.assertTrue(len(polled.get("progress", {}).get("parts", [])) > 1 or polled["state"] not in {"queued", "running"}) self.assertEqual(client.get(f"/tts/jobs/{row['id']}?after=99").status_code, 200) self.assertEqual(client.get(f"/tts/jobs/{row['id']}?after=x").status_code, 422) self.assertEqual(client.post("/tts/jobs", json={**body, "segment_ends": [13]}).status_code, 409) for ends in ([1, 1, 13], [6, 7, 13], [6], [13, 6], [True, 13], [13] * 65): self.assertEqual(client.post("/tts/jobs", json={**body, "request_id": None, "segment_ends": ends}).status_code, 422, ends) self.assertEqual(client.get("/?id=1").status_code, 200) client.delete("/tts/jobs/" + row["id"]) def test_cache_only_history_restores_ready_audio_but_misses_never_queue_inference(self): with TestClient(app.create_app(service_options={"worker_factory": ProgressiveWorker})) as client: for _ in range(100): if client.get("/tts/health").status_code == 200: break time.sleep(.002) body = {"text": "已经完成的原文句段。", "progressive": True, "cache_only": True} self.assertEqual(client.post("/tts/jobs", json=body).status_code, 410) self.assertEqual(client.get("/tts/health").json()["queued"], 0) created = client.post("/tts/jobs", json={**body, "cache_only": False, "request_id": "c" * 48}).json() self.assertEqual(client.post("/tts/jobs", json={**body, "request_id": "c" * 48}).status_code, 410) for _ in range(150): data = client.get("/tts/jobs/" + created["id"]).json() if data["state"] == "ready": break time.sleep(.003) self.assertEqual(data["state"], "ready") client.delete("/tts/jobs/" + created["id"]) cached = client.post("/tts/jobs", json=body).json() self.assertTrue(cached["cache_hit"]) self.assertEqual(cached["state"], "ready") self.assertEqual(cached["source_segments"], data["source_segments"]) self.assertEqual(client.get(cached["progress"]["parts"][0]["audio_url"]).content, AUDIO) for bad in (1, "yes", None): self.assertEqual(client.post("/tts/jobs", json={**body, "cache_only": bad}).status_code, 422) client.delete("/tts/jobs/" + cached["id"])