import io
import json
import struct
import sys
import wave

import httpx
import pytest
from fastapi.testclient import TestClient

from knaster.audio import rtf, wav_duration
from knaster.bench import bench_tts
from knaster.canary import CanaryProber
from knaster.llm import OpenAIChatClient
from knaster.tts_server import CommandBackend, create_tts_app


def make_wav(seconds: float, rate: int = 22050) -> bytes:
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(rate)
        w.writeframes(struct.pack("<h", 0) * int(seconds * rate))
    return buf.getvalue()


FAKE_PIPER = """
import sys, wave, struct
args = sys.argv[1:]
out = args[args.index("-f") + 1]
model = args[args.index("-m") + 1]
text = sys.stdin.read()
if "FEHLER" in text:
    print("kaputt", file=sys.stderr); sys.exit(3)
with wave.open(out, "wb") as w:
    w.setnchannels(1); w.setsampwidth(2); w.setframerate(16000)
    w.writeframes(struct.pack("<h", 0) * (160 * len(text)))   # 10 ms pro Zeichen
"""


@pytest.fixture
def backend(tmp_path):
    script = tmp_path / "fake_piper.py"
    script.write_text(FAKE_PIPER)
    return CommandBackend({"thorsten": "/voices/t.onnx", "kerstin": "/voices/k.onnx"},
                          cmd=f"{sys.executable} {script} -m {{model}} -f {{out}}")


def test_wav_helpers():
    assert wav_duration(make_wav(1.5)) == pytest.approx(1.5)
    assert wav_duration(b"kein wav") == 0.0
    assert rtf(3.0, 1.5) == 2.0 and rtf(1.0, 0) is None


def test_tts_server(backend):
    client = TestClient(create_tts_app(backend, token="t"))
    assert client.get("/health").json()["default"] == "thorsten"
    H = {"Authorization": "Bearer t"}
    r = client.post("/v1/audio/speech", json={"input": "Hallo Welt"}, headers=H)
    assert r.status_code == 200 and r.headers["content-type"] == "audio/wav"
    assert wav_duration(r.content) == pytest.approx(0.1)
    assert client.post("/v1/audio/speech", json={"input": "x", "voice": "kerstin"}, headers=H).status_code == 200
    assert client.post("/v1/audio/speech", json={"input": "x", "voice": "nope"}, headers=H).status_code == 400
    assert client.post("/v1/audio/speech", json={"input": "FEHLER"}, headers=H).status_code == 502
    assert client.post("/v1/audio/speech", json={"input": "x", "response_format": "mp3"},
                       headers=H).status_code == 400
    assert client.post("/v1/audio/speech", json={"input": "x"}).status_code == 401


def test_backend_requires_voice():
    with pytest.raises(ValueError):
        CommandBackend({})


def tts_transport(seconds: float = 2.0, fail: bool = False):
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/v1/audio/speech"
        body = json.loads(request.content)
        if fail or body["model"] == "kaputt":
            return httpx.Response(503, text="down")
        return httpx.Response(200, content=make_wav(seconds), headers={"content-type": "audio/wav"})

    return httpx.Client(transport=httpx.MockTransport(handler))


def test_bench_tts():
    client = OpenAIChatClient("http://x", client=tts_transport(2.0))
    result = bench_tts("http://x", "tts-de", runs=2, client=client)
    assert len(result["runs"]) == 2 and result["errors"] == []
    assert result["runs"][0]["audio_seconds"] == 2.0 and result["rtf_median"] is not None
    bad = bench_tts("http://x", "kaputt", runs=1, client=client)
    assert bad["runs"] == [] and bad["errors"]


def test_canary_probes_tts_slots(registry):
    from prometheus_client import CollectorRegistry

    from knaster.registry import Registry
    from knaster.store import Store

    from .conftest import REGISTRY

    data = {"slots": {**REGISTRY["slots"], "tts-de": {
        "tier": 1, "device": "cpu", "engine": "piper", "litellm_model": "tts-de", "base_model": "piper",
        "max_context": 5000, "kind": "tts"}}}
    reg = Registry.from_dict(data, Store("sqlite://"))
    chat = OpenAIChatClient("http://x", client=tts_transport(0.8))
    prober = CanaryProber(reg, chat, metrics_registry=CollectorRegistry())
    tts = next(prober.probe(s) for s in reg.slots() if s.name == "tts-de")
    assert tts.up and tts.ok
    silent = OpenAIChatClient("http://x", client=tts_transport(0.05))
    assert not CanaryProber(reg, silent, metrics_registry=CollectorRegistry()).probe(reg.slot("tts-de")).ok
    down = OpenAIChatClient("http://x", client=tts_transport(fail=True))
    assert not CanaryProber(reg, down, metrics_registry=CollectorRegistry()).probe(reg.slot("tts-de")).up