www_site/services/api/tests/phase3_reliability.py

"""Isolated Phase 3 reliability checks using a local fake upstream.

Run on the test server with DATABASE_URL and the API virtualenv. This never calls
the real TTS provider and uses test-only accounts in the shared test database.
"""

from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from uuid import uuid4

import httpx
import psycopg


API_PORT = 8100
FAKE_PORT = 9100
API_ROOT = f"http://127.0.0.1:{API_PORT}/api/v1"
FAKE_ROOT = f"http://127.0.0.1:{FAKE_PORT}"


def wav_bytes() -> bytes:
    return b"RIFF" + (36).to_bytes(4, "little") + b"WAVEfmt " + bytes(36)


class FakeTts(BaseHTTPRequestHandler):
    mode = "success"
    calls = 0

    def do_POST(self):  # noqa: N802
        type(self).calls += 1
        if self.mode == "timeout":
            time.sleep(2)
        if self.mode == "non_audio":
            payload, content_type = b"not audio", "text/plain"
        elif self.mode == "corrupt":
            payload, content_type = b"bad wav", "audio/wav"
        elif self.mode == "oversized":
            payload, content_type = b"", "audio/wav"
            self.send_response(200)
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", str(50 * 1024 * 1024 + 1))
            self.end_headers()
            return
        else:
            payload, content_type = wav_bytes(), "audio/wav"
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, *_args):
        return


def start_fake() -> ThreadingHTTPServer:
    server = ThreadingHTTPServer(("127.0.0.1", FAKE_PORT), FakeTts)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


def start_api(storage_dir: Path) -> subprocess.Popen:
    env = os.environ.copy()
    env.update({
        "TTS_UPSTREAM_URL": FAKE_ROOT,
        "TTS_API_KEY": "test-only-fake-key",
        "TTS_TIMEOUT_SECONDS": "1",
        "AUDIO_STORAGE_DIR": str(storage_dir),
        "SESSION_COOKIE_SECURE": "false",
    })
    root = Path(__file__).resolve().parents[1]
    process = subprocess.Popen([sys.executable, "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", str(API_PORT), "--workers", "1"], cwd=root, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    for _ in range(50):
        try:
            if httpx.get(f"http://127.0.0.1:{API_PORT}/healthz", timeout=1).status_code == 200:
                return process
        except httpx.HTTPError:
            time.sleep(0.2)
    process.kill()
    raise RuntimeError("isolated API did not start")


def stop_api(process: subprocess.Popen, force: bool = False):
    if process.poll() is None:
        process.send_signal(signal.SIGKILL if force else signal.SIGTERM)
        process.wait(timeout=10)


def client() -> httpx.Client:
    session = httpx.Client(base_url=API_ROOT, timeout=10, follow_redirects=True)
    csrf = session.get("/auth/csrf").json()["csrf_token"]
    email = f"phase3-fake-{uuid4().hex[:12]}@example.com"
    password = f"Test-{uuid4().hex}-Aa1!"
    body = {"email": email, "password": password}
    assert session.post("/auth/register", json=body, headers={"X-CSRF-Token": csrf}).status_code == 201
    assert session.post("/auth/login", json=body, headers={"X-CSRF-Token": csrf}).status_code == 200
    session.headers.update({"X-CSRF-Token": csrf})
    return session


def create_task(session: httpx.Client, text: str) -> tuple[int, dict]:
    response = session.post("/tts/tasks", headers={"Idempotency-Key": f"test-{uuid4().hex}"}, json={"text": text, "voice_id": "default", "parameters": {"format": "wav", "speed": 1}})
    return response.status_code, response.json()


def wait_task(session: httpx.Client, task_id: str, timeout: float = 8) -> dict:
    end = time.time() + timeout
    while time.time() < end:
        task = session.get(f"/tts/tasks/{task_id}").json()
        if task["status"] in {"succeeded", "failed"}:
            return task
        time.sleep(0.2)
    raise AssertionError("task did not terminate")


def main():
    storage = Path("/tmp/kaotings-phase3-reliability")
    storage.mkdir(parents=True, exist_ok=True)
    fake = start_fake()
    api = start_api(storage)
    results = {}
    try:
        session = client()
        before = session.get("/account/usage").json()
        code, created = create_task(session, "测试")
        if code != 202:
            raise AssertionError(f"success setup task failed: {code} {created}")
        success = wait_task(session, created["id"])
        results["success"] = {"create": code, "status": success["status"], "audio": session.get(f"/tts/tasks/{created['id']}/audio").status_code, "download": session.get(f"/tts/tasks/{created['id']}/download").status_code}

        for mode in ["non_audio", "corrupt", "oversized", "timeout"]:
            FakeTts.mode = mode
            _, task = create_task(session, mode)
            results[mode] = wait_task(session, task["id"])["error_code"]
        FakeTts.mode = "success"

        storage_file = Path("/tmp/kaotings-phase3-storage-file")
        storage_file.write_text("not a directory")
        stop_api(api)
        api = start_api(storage_file)
        _, task = create_task(session, "存储失败")
        results["storage_failure"] = wait_task(session, task["id"])["status"]
        storage_file.unlink(missing_ok=True)
        results["calls"] = FakeTts.calls
        results["quota"] = session.get("/account/usage").json()
        print(json.dumps(results, ensure_ascii=False))
    finally:
        stop_api(api)
        fake.shutdown()


if __name__ == "__main__":
    main()