| 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) |
| |
| 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"]) |
|
|
