300 lines
12 KiB
Python
300 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import main
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
data_dir = tmp_path / "data"
|
|
monkeypatch.setattr(main, "DATA_DIR", data_dir)
|
|
monkeypatch.setattr(main, "REFERENCES_DIR", data_dir / "references")
|
|
monkeypatch.setattr(main, "GENERATIONS_DIR", data_dir / "generations")
|
|
monkeypatch.setattr(main, "DB_PATH", data_dir / "studio.sqlite3")
|
|
queue = main.JobQueue()
|
|
async def start_without_worker():
|
|
main.init_db()
|
|
monkeypatch.setattr(queue, "start", start_without_worker)
|
|
monkeypatch.setattr(main, "job_queue", queue)
|
|
with TestClient(main.app) as test_client:
|
|
yield test_client
|
|
|
|
|
|
def wav_bytes() -> bytes:
|
|
# Minimal RIFF/WAVE-shaped fixture; endpoint validates the container header.
|
|
return b"RIFF\x24\x00\x00\x00WAVEfmt " + b"\x10\x00\x00\x00\x01\x00\x01\x00\x44\xac\x00\x00\x88\x58\x01\x00\x02\x00\x10\x00data\x00\x00\x00\x00"
|
|
|
|
|
|
def create_reference(client: TestClient) -> dict:
|
|
response = client.post(
|
|
"/api/references",
|
|
files={"file": ("voice.wav", wav_bytes(), "audio/wav")},
|
|
data={"transcript": "Hello, this is a sample.", "name": "Studio voice"},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
def test_reference_upload_and_transcript_required(client: TestClient):
|
|
response = client.post(
|
|
"/api/references",
|
|
files={"file": ("voice.wav", wav_bytes(), "audio/wav")},
|
|
data={"transcript": " ", "name": "Empty"},
|
|
)
|
|
assert response.status_code == 422
|
|
reference = create_reference(client)
|
|
assert reference["transcript"] == "Hello, this is a sample."
|
|
assert client.get(reference["audio_url"]).content == wav_bytes()
|
|
assert client.get("/api/references").json()[0]["id"] == reference["id"]
|
|
|
|
|
|
def test_rejects_non_wav_upload(client: TestClient):
|
|
response = client.post(
|
|
"/api/references",
|
|
files={"file": ("voice.mp3", b"not wav", "audio/mpeg")},
|
|
data={"transcript": "sample"},
|
|
)
|
|
assert response.status_code == 415
|
|
|
|
|
|
def test_generations_are_queued_and_can_be_cancelled(client: TestClient):
|
|
reference = create_reference(client)
|
|
created = client.post("/api/generations", json={"text": "Read this aloud", "reference_id": reference["id"]})
|
|
assert created.status_code == 202
|
|
job = created.json()
|
|
assert job["status"] in {"queued", "running"}
|
|
# Cancellation is valid regardless of whether the single worker has picked it up.
|
|
cancelled = client.post(f"/api/generations/{job['id']}/cancel")
|
|
assert cancelled.status_code == 200
|
|
assert cancelled.json()["status"] == "cancelled"
|
|
assert client.post(f"/api/generations/{job['id']}/cancel").json()["status"] == "cancelled"
|
|
|
|
|
|
def test_running_jobs_cancel_and_completed_jobs_cannot(client: TestClient):
|
|
reference = create_reference(client)
|
|
running = client.post("/api/generations", json={"text": "Active request", "reference_id": reference["id"]}).json()
|
|
completed = client.post("/api/generations", json={"text": "Finished request", "reference_id": reference["id"]}).json()
|
|
with main.connect_db() as db:
|
|
db.execute("UPDATE generations SET status='running' WHERE id=?", (running["id"],))
|
|
db.execute("UPDATE generations SET status='completed', filename='done.mp3' WHERE id=?", (completed["id"],))
|
|
cancelled = client.post(f"/api/generations/{running['id']}/cancel")
|
|
assert cancelled.status_code == 200
|
|
assert cancelled.json()["status"] == "cancelled"
|
|
cannot_cancel = client.post(f"/api/generations/{completed['id']}/cancel")
|
|
assert cannot_cancel.status_code == 409
|
|
|
|
|
|
def test_generation_history_paginates_at_fifty(client: TestClient):
|
|
reference = create_reference(client)
|
|
with main.connect_db() as db:
|
|
for index in range(52):
|
|
stamp = f"2026-10-27T12:{index:02}:00+00:00"
|
|
db.execute(
|
|
"INSERT INTO generations(id,text,reference_id,status,created_at,updated_at) VALUES(?,?,?,'failed',?,?)",
|
|
(f"job-{index:02}", f"Take {index}", reference["id"], stamp, stamp),
|
|
)
|
|
page_one = client.get("/api/generations?page=1&page_size=50").json()
|
|
page_two = client.get("/api/generations?page=2&page_size=50").json()
|
|
assert page_one["total"] == 52
|
|
assert len(page_one["items"]) == 50
|
|
assert len(page_two["items"]) == 2
|
|
assert page_one["items"][0]["text"] == "Take 51"
|
|
assert client.get("/api/generations?page=1&page_size=20").status_code == 422
|
|
|
|
|
|
def test_fish_speech_mock_request_persists_mp3_and_respects_cancel(client: TestClient):
|
|
import asyncio
|
|
import httpx
|
|
import msgpack
|
|
|
|
reference = create_reference(client)
|
|
job = client.post("/api/generations", json={"text": "A test take", "reference_id": reference["id"]}).json()
|
|
outgoing: list[dict] = []
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
outgoing.append(msgpack.unpackb(request.content, raw=False))
|
|
return httpx.Response(200, content=b"fake-mp3")
|
|
|
|
queue = main.JobQueue()
|
|
queue.http = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
asyncio.run(queue.process(job["id"]))
|
|
asyncio.run(queue.http.aclose())
|
|
assert outgoing[0]["text"] == "A test take"
|
|
assert outgoing[0]["format"] == "mp3"
|
|
assert outgoing[0]["references"][0]["text"] == "Hello, this is a sample."
|
|
result = client.get(f"/api/generations/{job['id']}").json()
|
|
assert result["status"] == "completed"
|
|
assert client.get(result["audio_url"]).content == b"fake-mp3"
|
|
|
|
|
|
def test_cancelled_queued_job_never_calls_fish_speech(client: TestClient):
|
|
import asyncio
|
|
import httpx
|
|
|
|
reference = create_reference(client)
|
|
job = client.post("/api/generations", json={"text": "A cancelled take", "reference_id": reference["id"]}).json()
|
|
client.post(f"/api/generations/{job['id']}/cancel")
|
|
calls = 0
|
|
|
|
async def handler(_: httpx.Request) -> httpx.Response:
|
|
nonlocal calls
|
|
calls += 1
|
|
return httpx.Response(200, content=b"unexpected")
|
|
|
|
queue = main.JobQueue()
|
|
queue.http = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
asyncio.run(queue.process(job["id"]))
|
|
asyncio.run(queue.http.aclose())
|
|
assert calls == 0
|
|
assert client.get(f"/api/generations/{job['id']}").json()["status"] == "cancelled"
|
|
|
|
|
|
def test_running_cancellation_discards_upstream_result(client: TestClient):
|
|
import asyncio
|
|
import httpx
|
|
|
|
reference = create_reference(client)
|
|
job = client.post("/api/generations", json={"text": "Cancel while running", "reference_id": reference["id"]}).json()
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def scenario():
|
|
async def handler(_: httpx.Request) -> httpx.Response:
|
|
started.set()
|
|
await release.wait()
|
|
return httpx.Response(200, content=b"discard this mp3")
|
|
|
|
queue = main.JobQueue()
|
|
queue.http = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
work = asyncio.create_task(queue.process(job["id"]))
|
|
await started.wait()
|
|
response = client.post(f"/api/generations/{job['id']}/cancel")
|
|
assert response.json()["status"] == "cancelled"
|
|
release.set()
|
|
await work
|
|
await queue.http.aclose()
|
|
|
|
asyncio.run(scenario())
|
|
result = client.get(f"/api/generations/{job['id']}").json()
|
|
assert result["status"] == "cancelled"
|
|
assert result["audio_url"] is None
|
|
assert not (main.GENERATIONS_DIR / f"{job['id']}.mp3").exists()
|
|
|
|
|
|
def test_fish_speech_errors_are_saved_on_the_job(client: TestClient):
|
|
import asyncio
|
|
import httpx
|
|
|
|
reference = create_reference(client)
|
|
job = client.post("/api/generations", json={"text": "An upstream error", "reference_id": reference["id"]}).json()
|
|
|
|
async def handler(_: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(503, text="model unavailable")
|
|
|
|
async def scenario():
|
|
queue = main.JobQueue()
|
|
queue.http = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
await queue.process(job["id"])
|
|
await queue.http.aclose()
|
|
|
|
asyncio.run(scenario())
|
|
result = client.get(f"/api/generations/{job['id']}").json()
|
|
assert result["status"] == "failed"
|
|
assert "503" in result["error"]
|
|
assert "model unavailable" in result["error"]
|
|
|
|
|
|
def test_fish_speech_connection_failure_is_saved_on_the_job(client: TestClient):
|
|
import asyncio
|
|
import httpx
|
|
|
|
reference = create_reference(client)
|
|
job = client.post("/api/generations", json={"text": "A disconnected request", "reference_id": reference["id"]}).json()
|
|
|
|
async def handler(_: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("server is offline")
|
|
|
|
async def scenario():
|
|
queue = main.JobQueue()
|
|
queue.http = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
await queue.process(job["id"])
|
|
await queue.http.aclose()
|
|
|
|
asyncio.run(scenario())
|
|
result = client.get(f"/api/generations/{job['id']}").json()
|
|
assert result["status"] == "failed"
|
|
assert "Could not reach Fish Speech" in result["error"]
|
|
|
|
|
|
def test_tts_url_normalization_and_validation():
|
|
assert main.normalize_tts_url(" http://192.168.1.42:8080/ ") == "http://192.168.1.42:8080/v1/tts"
|
|
assert main.normalize_tts_url("http://192.168.1.42:8080/v1/tts") == "http://192.168.1.42:8080/v1/tts"
|
|
with pytest.raises(main.HTTPException):
|
|
main.normalize_tts_url("file:///etc/passwd")
|
|
|
|
|
|
def test_tts_url_probe_uses_fish_health_endpoint(monkeypatch: pytest.MonkeyPatch):
|
|
import asyncio
|
|
|
|
requests = []
|
|
async def handler(request):
|
|
requests.append(request)
|
|
return main.httpx.Response(200, json={"status": "ok"})
|
|
|
|
client_class = main.httpx.AsyncClient
|
|
monkeypatch.setattr(
|
|
main.httpx,
|
|
"AsyncClient",
|
|
lambda **kwargs: client_class(transport=main.httpx.MockTransport(handler), **kwargs),
|
|
)
|
|
url, message = asyncio.run(main.probe_tts_url("http://192.168.1.42:8080"))
|
|
assert url == "http://192.168.1.42:8080/v1/tts"
|
|
assert "healthy" in message
|
|
assert requests[0].url.path == "/v1/health"
|
|
|
|
|
|
def test_tts_urls_must_pass_probe_before_saving_and_can_be_activated(client: TestClient, monkeypatch: pytest.MonkeyPatch):
|
|
async def passing_probe(url: str) -> tuple[str, str]:
|
|
return main.normalize_tts_url(url), "Probe passed"
|
|
|
|
monkeypatch.setattr(main, "probe_tts_url", passing_probe)
|
|
test_result = client.post("/api/tts-endpoints/test", json={"url": "http://192.168.1.42:8080"})
|
|
assert test_result.status_code == 200
|
|
assert test_result.json()["url"] == "http://192.168.1.42:8080/v1/tts"
|
|
|
|
saved = client.post("/api/tts-endpoints", json={"url": "http://192.168.1.42:8080"})
|
|
assert saved.status_code == 201
|
|
endpoint = saved.json()
|
|
assert endpoint["active"] is True
|
|
assert main.active_tts_url() == endpoint["url"]
|
|
|
|
reference = create_reference(client)
|
|
generation = client.post("/api/generations", json={"text": "Use the selected host", "reference_id": reference["id"]}).json()
|
|
assert main.get_generation(generation["id"])["tts_url"] == endpoint["url"]
|
|
|
|
second = client.post("/api/tts-endpoints", json={"url": "http://192.168.1.43:8080/v1/tts"}).json()
|
|
activated = client.post(f"/api/tts-endpoints/{endpoint['id']}/activate")
|
|
assert activated.status_code == 200
|
|
assert activated.json()["url"] == endpoint["url"]
|
|
assert main.active_tts_url() == endpoint["url"]
|
|
assert client.get("/api/tts-endpoints").json()["items"]
|
|
assert second["url"] != endpoint["url"]
|
|
|
|
|
|
def test_startup_marks_interrupted_jobs_failed(client: TestClient):
|
|
reference = create_reference(client)
|
|
with main.connect_db() as db:
|
|
db.execute(
|
|
"INSERT INTO generations(id,text,reference_id,status,created_at,updated_at) VALUES('interrupted','hello',?,'running','2026-10-27T12:00:00+00:00','2026-10-27T12:00:00+00:00')",
|
|
(reference["id"],),
|
|
)
|
|
main.init_db()
|
|
row = main.get_generation("interrupted")
|
|
assert row["status"] == "failed"
|
|
assert "restarted" in row["error"].lower()
|
|
|