"""Regression tests for the javis-slave audit fixes (docs/AUDIT_SLAVE.md)."""
from __future__ import annotations

import asyncio
import json
import sys
import time
from pathlib import Path

import httpx
import numpy as np
import pytest

SLAVE_SRC = Path(__file__).resolve().parents[1] / "javis-slave" / "src"
if str(SLAVE_SRC) not in sys.path:
    sys.path.insert(0, str(SLAVE_SRC))

from javis_slave import config as slave_config  # noqa: E402
from javis_slave.api_client import MasterClient, MasterError  # noqa: E402
from javis_slave.config import SlaveConfig, ensure_control_token  # noqa: E402

TOKEN = "t" * 32


@pytest.fixture
def cfg_dir(tmp_path, monkeypatch):
    monkeypatch.setenv("JAVIS_SLAVE_CONFIG_DIR", str(tmp_path))
    return tmp_path


# ── SLV-07: config loading ───────────────────────────────────────────────────

def test_corrupt_config_is_kept_aside(cfg_dir):
    (cfg_dir / "slave_config.json").write_text("{not json", encoding="utf-8")
    cfg = slave_config.load_config()
    assert not cfg.paired
    assert (cfg_dir / "slave_config.json.corrupt").read_text(encoding="utf-8") == "{not json"
    assert not (cfg_dir / "slave_config.json").exists()


def test_config_type_mismatch_keeps_default(cfg_dir):
    (cfg_dir / "slave_config.json").write_text(
        json.dumps({"master_port": "abc", "input_gain": 2, "api_key": "k", "master_host": "h"}),
        encoding="utf-8")
    cfg = slave_config.load_config()
    assert cfg.master_port == 8765
    assert cfg.input_gain == 2.0 and isinstance(cfg.input_gain, float)
    assert cfg.paired


def test_control_token_roundtrip(cfg_dir):
    cfg = SlaveConfig()
    assert ensure_control_token(cfg) is True
    assert len(cfg.control_token) >= 24
    assert ensure_control_token(cfg) is False
    slave_config.save_config(cfg)
    assert slave_config.load_config().control_token == cfg.control_token


# ── SLV-01: control API authentication ───────────────────────────────────────

def _app(token: str = TOKEN):
    from javis_slave.http_server import build_http_app
    cfg = SlaveConfig(control_token=token)
    return build_http_app(cfg, lambda: None, lambda: "idle", lambda: None, lambda: (0.0, 0.0))


def test_control_api_requires_token_from_lan():
    from fastapi.testclient import TestClient
    remote = TestClient(_app(), client=("192.168.1.50", 50000))
    assert remote.get("/mcp/v1/health").status_code == 200
    assert remote.get("/mcp/v1/status").status_code == 401
    # Rejected by the middleware before the handler runs: nothing is executed.
    assert remote.post("/mcp/v1/power/cancel").status_code == 401
    assert remote.get("/mcp/v1/status", headers={"Authorization": "Bearer wrong"}).status_code == 401
    # The loopback header is not a credential for remote callers.
    assert remote.get("/mcp/v1/status", headers={"X-Javis-Control": "1"}).status_code == 401
    ok = remote.get("/mcp/v1/status", headers={"Authorization": f"Bearer {TOKEN}"})
    assert ok.status_code == 200 and ok.json()["audio_state"] == "idle"
    assert remote.get("/docs").status_code in (401, 404)


def test_control_api_loopback_needs_custom_header():
    from fastapi.testclient import TestClient
    for host in ("127.0.0.1", "::1", "::ffff:127.0.0.1"):
        local = TestClient(_app(), client=(host, 50000))
        # Plain loopback request (e.g. a browser form post / CSRF) is refused.
        assert local.get("/mcp/v1/status").status_code == 401, host
        assert local.get("/mcp/v1/status", headers={"X-Javis-Control": "1"}).status_code == 200, host


def test_empty_token_never_authorizes():
    from fastapi.testclient import TestClient
    remote = TestClient(_app(token=""), client=("10.0.0.2", 50000))
    assert remote.get("/mcp/v1/status", headers={"Authorization": "Bearer "}).status_code == 401


# ── SLV-03: no blind retry of a non-idempotent turn after a read timeout ────

def _client_with(handler) -> MasterClient:
    cfg = SlaveConfig(master_host="master.invalid", api_key="k")
    client = MasterClient(cfg)
    client._backoff_base = 0.0
    client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    return client


def test_send_turn_does_not_retry_after_read_timeout():
    calls = []

    def handler(request):
        calls.append(request.url.path)
        raise httpx.ReadTimeout("slow", request=request)

    async def run():
        client = _client_with(handler)
        try:
            with pytest.raises(MasterError, match="Zeitüberschreitung"):
                await client.send_turn(b"RIFF")
        finally:
            await client.aclose()

    asyncio.run(run())
    assert calls == ["/api/v1/turn"]


def test_send_turn_retries_connect_error_once():
    calls = []

    def handler(request):
        calls.append(1)
        raise httpx.ConnectError("refused", request=request)

    async def run():
        client = _client_with(handler)
        try:
            with pytest.raises(MasterError):
                await client.send_turn(b"RIFF")
        finally:
            await client.aclose()

    asyncio.run(run())
    assert len(calls) == 2


def test_send_turn_reads_master_followup_flag():
    def handler(request):
        return httpx.Response(
            200,
            json={
                "user_text": "Ja",
                "reply_text": "Noch etwas?",
                "expects_followup": True,
            },
        )

    async def run():
        client = _client_with(handler)
        try:
            return await client.send_turn(b"RIFF")
        finally:
            await client.aclose()

    result = asyncio.run(run())
    assert result.expects_followup is True


def test_slave_followup_skips_wake_word_and_starts_recording():
    from javis_slave.audio_engine import AudioEngine

    async def run():
        eng = AudioEngine(SlaveConfig(), client=None)
        eng._followup_pending = True
        calls = []

        async def config_poll():
            await eng._stop.wait()

        async def wait_for_wake():
            calls.append("wake")
            return True

        async def handle_turn(*, is_followup=False):
            calls.append(("turn", is_followup))
            eng._stop.set()

        async def play_chime(*args):
            calls.append("chime")

        eng._config_poll_loop = config_poll
        eng._wait_for_wake = wait_for_wake
        eng._handle_turn = handle_turn
        eng._play_chime = play_chime
        await eng.run()
        await asyncio.sleep(0)
        return calls

    calls = asyncio.run(run())
    assert "wake" not in calls
    assert ("turn", True) in calls


# ── SLV-04 / SLV-06: audio engine ────────────────────────────────────────────

def test_engine_start_failure_releases_mic(monkeypatch):
    from javis_slave import audio_engine

    class FakeMic:
        instances: list = []

        def __init__(self, **kw):
            self.stopped = False
            FakeMic.instances.append(self)

        async def start(self):
            pass

        async def stop(self):
            self.stopped = True

    def broken_vad(**kw):
        raise RuntimeError("vad model missing")

    monkeypatch.setattr(audio_engine, "MicStream", FakeMic)
    monkeypatch.setattr(audio_engine, "make_vad", broken_vad)

    async def run():
        eng = audio_engine.AudioEngine(SlaveConfig(), client=None)
        with pytest.raises(RuntimeError, match="vad model missing"):
            await eng.start()
        return eng

    eng = asyncio.run(run())
    assert FakeMic.instances and FakeMic.instances[0].stopped
    assert eng.mic is None


@pytest.mark.parametrize("name", ["", ".", "..", "../evil", "sub/model", "..\\evil", "C:\\x\\model"])
def test_wake_model_name_cannot_address_a_path(name):
    from javis_slave.audio_engine import AudioEngine

    class NoDownload:
        async def download_wake_model(self, dest):  # pragma: no cover - must not be reached
            raise AssertionError(f"download attempted for {dest}")

    async def run():
        eng = AudioEngine(SlaveConfig(), client=NoDownload())
        with pytest.raises(RuntimeError, match="invalid wake word model name"):
            await eng._load_wake(name, 0.5)

    asyncio.run(run())


# ── SLV-05: playback writer stops before cancellation completes ─────────────

def test_write_pcm_blocks_cancel_stops_writer_thread():
    from javis_slave.audio_io.stream import write_pcm_blocks

    class SlowStream:
        writes = 0

        def write(self, block):
            time.sleep(0.01)
            SlowStream.writes += 1

    async def run():
        task = asyncio.ensure_future(
            write_pcm_blocks(SlowStream(), np.zeros(16000 * 5, dtype=np.float32), 160))
        await asyncio.sleep(0.05)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        done = SlowStream.writes
        await asyncio.sleep(0.05)
        return done

    done = asyncio.run(run())
    assert 0 < done < 500
    assert SlowStream.writes == done  # no write after the stream may be closed


def test_write_pcm_blocks_cancel_event():
    from javis_slave.audio_io.stream import write_pcm_blocks

    class Stream:
        def write(self, block):
            pass

    async def run():
        ev = asyncio.Event()
        ev.set()
        return await write_pcm_blocks(Stream(), np.zeros(1600, dtype=np.float32), 160, ev)

    assert asyncio.run(run()) is True


# ── SLV-02 / SLV-09: Windows system control ─────────────────────────────────

def test_volume_controller_uses_current_pycaw_api(monkeypatch):
    import types

    from javis_slave import system_control

    class Endpoint:
        level = 0.25

        def GetMasterVolumeLevelScalar(self):
            return Endpoint.level

        def SetMasterVolumeLevelScalar(self, level, ctx):
            Endpoint.level = level

        def GetMute(self):
            return 1

    class Device:  # pycaw >= 2024 AudioDevice: has .EndpointVolume, no .Activate
        EndpointVolume = Endpoint()

    fake = types.ModuleType("pycaw.pycaw")
    fake.AudioUtilities = types.SimpleNamespace(GetSpeakers=lambda: Device())
    fake.IAudioEndpointVolume = object
    monkeypatch.setitem(sys.modules, "pycaw", types.ModuleType("pycaw"))
    monkeypatch.setitem(sys.modules, "pycaw.pycaw", fake)
    monkeypatch.setattr(system_control.platform, "system", lambda: "Windows")

    vol = system_control.VolumeController()
    assert vol.get_volume() == 0.25
    assert vol.set_volume(1.7) and Endpoint.level == 1.0
    assert vol.is_muted() is True


@pytest.mark.skipif(sys.platform != "win32", reason="Windows exit-code handling")
def test_windows_power_runner_checks_exit_codes():
    from javis_slave.system_control import _run_win
    assert _run_win(["cmd", "/c", "exit 0"]) is True
    assert _run_win(["cmd", "/c", "exit 5"]) is False
    assert _run_win(["cmd", "/c", "exit 1116"], ok_codes=(0, 1116)) is True