my_voice_assistant_v3_jamulix/scripts/smoke_e2e.py

#!/usr/bin/env python3
"""End-to-End-Rauchtest gegen die echten Provider (OpenRouter).

ACHTUNG: macht echte Netz-Aufrufe und verursacht (geringe) Kosten. Bewusst NICHT
Teil der pytest-Suite. Manuell ausfuehren:

    OPENROUTER_API_KEY=sk-or-... python scripts/smoke_e2e.py
    # oder, wenn der Key in ~/.bashrc exportiert ist:
    python scripts/smoke_e2e.py

Prueft drei Dinge isoliert gegen OpenRouter:
  1. LLM   (Chat-Antwort; TTS auf lokalen Stub, um den LLM zu isolieren)
  2. TTS   (Text -> Audio)
  3. STT   (Round-Trip: TTS-Audio als WAV zurueck durch die Transkription)

Exit-Code 0 = alles ok, 1 = mindestens ein Test fehlgeschlagen, 2 = kein API-Key.
"""

import io
import os
import sys
import tempfile
import wave

# Defaults setzen, BEVOR die App-Config geladen wird.
os.environ.setdefault("AUTH_ENABLED", "false")
os.environ.setdefault("DB_PATH", os.path.join(tempfile.mkdtemp(), "smoke.db"))

from fastapi.testclient import TestClient  # noqa: E402

from app.config import settings  # noqa: E402
from app.main import app  # noqa: E402

OR = "openrouter"


def _check_key() -> None:
    if not settings.openrouter_api_key.strip():
        print("FEHLER: OPENROUTER_API_KEY ist nicht gesetzt (Umgebung).")
        sys.exit(2)


def _pcm_to_wav(pcm: bytes, sample_rate: int = 24000) -> bytes:
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(sample_rate)
        w.writeframes(pcm)
    return buf.getvalue()


def main() -> int:
    _check_key()
    print(f"Modelle: LLM={settings.openrouter_llm_model}, "
          f"TTS={settings.openrouter_tts_model} ({settings.openrouter_tts_voice}), "
          f"STT={settings.openrouter_stt_model}\n")

    client = TestClient(app)
    failures = 0

    # 1) LLM (TTS via piper-Stub isoliert)
    try:
        r = client.post(
            "/api/chat?debug=true",
            json={"text": "Sag bitte in einem kurzen Satz Hallo.",
                  "llm_provider": OR, "tts_provider": "piper"},
        )
        answer = r.json().get("trace", {}).get("semantic_response") if r.status_code == 200 else None
        if r.status_code == 200 and answer:
            print(f"[OK]   LLM     -> {answer!r}")
        else:
            failures += 1
            print(f"[FAIL] LLM     -> HTTP {r.status_code}: {r.text[:200]}")
    except Exception as exc:  # noqa: BLE001
        failures += 1
        print(f"[FAIL] LLM     -> {exc}")

    # 2) TTS
    pcm = b""
    try:
        r = client.post("/api/speak", json={"text": "Guten Tag, schoen dass Sie da sind.",
                                            "tts_provider": OR})
        if r.status_code == 200 and r.content:
            pcm = r.content
            print(f"[OK]   TTS     -> {len(pcm)} Bytes Audio")
        else:
            failures += 1
            print(f"[FAIL] TTS     -> HTTP {r.status_code}: {r.text[:200]}")
    except Exception as exc:  # noqa: BLE001
        failures += 1
        print(f"[FAIL] TTS     -> {exc}")

    # 3) STT (Round-Trip ueber das TTS-Audio)
    if pcm:
        try:
            wav = _pcm_to_wav(pcm)
            r = client.post(
                "/api/transcribe",
                files={"file": ("roundtrip.wav", wav, "audio/wav")},
                data={"language": "de", "stt_provider": OR},
            )
            text = r.json().get("trace", {}).get("raw_transcript") if r.status_code == 200 else None
            if r.status_code == 200 and text:
                print(f"[OK]   STT     -> {text!r}")
            else:
                failures += 1
                print(f"[FAIL] STT     -> HTTP {r.status_code}: {r.text[:200]}")
        except Exception as exc:  # noqa: BLE001
            failures += 1
            print(f"[FAIL] STT     -> {exc}")
    else:
        print("[SKIP] STT     -> kein TTS-Audio fuer den Round-Trip")

    print()
    if failures:
        print(f"ERGEBNIS: {failures} Test(s) fehlgeschlagen.")
        return 1
    print("ERGEBNIS: alle Live-Tests bestanden.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())