#!/usr/bin/env python3
"""Config value validation: fail clearly on bad values, fall back to defaults (#21).

load_config merges DEFAULTS, config.json, and config.local.json and used to guard
only against invalid JSON, so a bad value (a wake_threshold above 1, an unknown
whisper_compute, a zero timeout, an unknown wake_word) surfaced deep in detect_wake,
_get_whisper, or speak rather than at load. validate_config now catches those at load,
prints a clear stderr message naming the key, and falls the key back to its default.

These checks are model-free and deterministic: they call validate_config / load_config
directly and monkeypatch the wake-model library, so they load no wake or whisper model.

Run:  .venv/bin/python test_config_validation.py
Exit code is 0 only if every check passes.
"""

from __future__ import annotations

import contextlib
import io
import json
import sys
from pathlib import Path

import pipeline
from pipeline import DEFAULTS

PASS, FAIL = "PASS", "FAIL"
results: list[tuple[str, str]] = []


def check(name: str, ok: bool, detail: str) -> bool:
    results.append((PASS if ok else FAIL, name))
    print(f"  [{PASS if ok else FAIL}] {name}: {detail}")
    return ok


def validate(cfg: dict) -> tuple[dict, str]:
    """Run validate_config on a copy, returning (validated_cfg, stderr_text)."""
    buf = io.StringIO()
    with contextlib.redirect_stderr(buf):
        out = pipeline.validate_config(dict(cfg))
    return out, buf.getvalue()


# --- unit-interval floats ---------------------------------------------------- #


def test_wake_threshold_out_of_range() -> None:
    for bad in (1.5, -0.1, 2, "high", True):
        cfg, err = validate({"wake_threshold": bad})
        check(
            f"wake_threshold {bad!r} -> default",
            cfg["wake_threshold"] == DEFAULTS["wake_threshold"]
            and "wake_threshold" in err,
            f"got {cfg['wake_threshold']!r}, stderr names key: {'wake_threshold' in err}",
        )


def test_wake_threshold_valid_untouched() -> None:
    for good in (0.0, 0.42, 1.0):
        cfg, err = validate({"wake_threshold": good})
        check(
            f"wake_threshold {good!r} kept",
            cfg["wake_threshold"] == good and err == "",
            f"got {cfg['wake_threshold']!r}, no warning: {err == ''}",
        )


def test_capture_vad_threshold_sibling() -> None:
    cfg, err = validate({"capture_vad_threshold": 9.0})
    check(
        "capture_vad_threshold 9.0 -> default",
        cfg["capture_vad_threshold"] == DEFAULTS["capture_vad_threshold"]
        and "capture_vad_threshold" in err,
        f"got {cfg['capture_vad_threshold']!r}",
    )


def test_stt_max_no_speech_prob_sibling() -> None:
    # Same [0, 1] rule as the two thresholds: out of range falls back, in range is kept.
    cfg, err = validate({"stt_max_no_speech_prob": 1.5})
    check(
        "stt_max_no_speech_prob 1.5 -> default",
        cfg["stt_max_no_speech_prob"] == DEFAULTS["stt_max_no_speech_prob"]
        and "stt_max_no_speech_prob" in err,
        f"got {cfg['stt_max_no_speech_prob']!r}",
    )
    cfg, err = validate({"stt_max_no_speech_prob": 0.6})
    check(
        "stt_max_no_speech_prob 0.6 kept",
        cfg["stt_max_no_speech_prob"] == 0.6 and err == "",
        f"got {cfg['stt_max_no_speech_prob']!r}, no warning: {err == ''}",
    )


# --- positive-int timeouts --------------------------------------------------- #


def test_timeout_bad_values() -> None:
    for bad in (0, -5, 1.5, True, "60"):
        cfg, err = validate({"claude_timeout_s": bad})
        check(
            f"claude_timeout_s {bad!r} -> default",
            cfg["claude_timeout_s"] == DEFAULTS["claude_timeout_s"]
            and "claude_timeout_s" in err,
            f"got {cfg['claude_timeout_s']!r}",
        )


def test_timeout_valid_untouched() -> None:
    cfg, err = validate({"claude_timeout_s": 90})
    check(
        "claude_timeout_s 90 kept",
        cfg["claude_timeout_s"] == 90 and err == "",
        f"got {cfg['claude_timeout_s']!r}, no warning: {err == ''}",
    )


# --- log level --------------------------------------------------------------- #


def test_log_level_validation() -> None:
    for good, expected in (("DEBUG", "DEBUG"), ("warning", "WARNING")):
        cfg, err = validate({"log_level": good})
        check(
            f"log_level {good!r} normalized",
            cfg["log_level"] == expected and err == "",
            f"got {cfg['log_level']!r}, no warning: {err == ''}",
        )
    for bad in ("verbose", 10, True):
        cfg, err = validate({"log_level": bad})
        check(
            f"log_level {bad!r} -> default",
            cfg["log_level"] == DEFAULTS["log_level"] and "log_level" in err,
            f"got {cfg['log_level']!r}",
        )


# --- whisper_compute enum ---------------------------------------------------- #


def test_whisper_compute_unknown() -> None:
    cfg, err = validate({"whisper_compute": "int9"})
    check(
        "whisper_compute 'int9' -> default",
        cfg["whisper_compute"] == DEFAULTS["whisper_compute"]
        and "whisper_compute" in err,
        f"got {cfg['whisper_compute']!r}",
    )


def test_whisper_compute_valid_untouched() -> None:
    # "default"/"auto" are always accepted (they let CTranslate2 choose), so they are the
    # host-independent valid examples: a concrete type like float16 is CPU-dependent.
    for good in ("default", "auto"):
        cfg, err = validate({"whisper_compute": good})
        check(
            f"whisper_compute {good!r} kept",
            cfg["whisper_compute"] == good and err == "",
            f"got {cfg['whisper_compute']!r}, no warning: {err == ''}",
        )


def test_whisper_compute_unsupported_backend_type_falls_back() -> None:
    # _get_whisper forces device="cpu" and CTranslate2 raises on a real type name the CPU
    # backend cannot load (float16/bfloat16/int16 on a Pi). validate_config must reject
    # those, not just typos. Host-independent: skip a type this box happens to support.
    supported = pipeline._supported_whisper_compute_types()
    if not supported:
        check(
            "whisper_compute deferred when support is unknown",
            True,
            "ctranslate2 unavailable; nothing to validate against",
        )
        return
    unsupported = next(
        (
            t
            for t in ("float16", "bfloat16", "int16", "int8_float16", "int8_bfloat16")
            if t not in supported
        ),
        None,
    )
    if unsupported is None:
        check(
            "whisper_compute CPU-unsupported path",
            True,
            "this backend supports all candidate types; skipped",
        )
        return
    cfg, err = validate({"whisper_compute": unsupported})
    check(
        f"CPU-unsupported whisper_compute {unsupported!r} -> default",
        cfg["whisper_compute"] == DEFAULTS["whisper_compute"]
        and "whisper_compute" in err,
        f"got {cfg['whisper_compute']!r}, stderr names key: {'whisper_compute' in err}",
    )


def test_whisper_compute_non_string_falls_back() -> None:
    # A JSON array/object is unhashable, so a naive `x not in set` would raise TypeError
    # instead of falling back; the string guard has to run first.
    for bad in ([], {}, 5):
        cfg, err = validate({"whisper_compute": bad})
        check(
            f"whisper_compute {bad!r} -> default (non-string)",
            cfg["whisper_compute"] == DEFAULTS["whisper_compute"]
            and "whisper_compute" in err,
            f"got {cfg['whisper_compute']!r}",
        )


def test_whisper_cpu_threads_validation() -> None:
    for good in (0, 1, 4):
        cfg, err = validate({"whisper_cpu_threads": good})
        check(
            f"whisper_cpu_threads {good!r} kept",
            cfg["whisper_cpu_threads"] == good and err == "",
            f"got {cfg['whisper_cpu_threads']!r}, no warning: {err == ''}",
        )
    for bad in (-1, 1.5, True, "4"):
        cfg, err = validate({"whisper_cpu_threads": bad})
        check(
            f"whisper_cpu_threads {bad!r} -> default",
            cfg["whisper_cpu_threads"] == DEFAULTS["whisper_cpu_threads"]
            and "whisper_cpu_threads" in err,
            f"got {cfg['whisper_cpu_threads']!r}",
        )


def test_whisper_vad_filter_validation() -> None:
    for good in (True, False):
        cfg, err = validate({"whisper_vad_filter": good})
        check(
            f"whisper_vad_filter {good!r} kept",
            cfg["whisper_vad_filter"] is good and err == "",
            f"got {cfg['whisper_vad_filter']!r}, no warning: {err == ''}",
        )
    for bad in (0, 1, "true", None):
        cfg, err = validate({"whisper_vad_filter": bad})
        check(
            f"whisper_vad_filter {bad!r} -> default",
            cfg["whisper_vad_filter"] == DEFAULTS["whisper_vad_filter"]
            and "whisper_vad_filter" in err,
            f"got {cfg['whisper_vad_filter']!r}",
        )


# --- live_output_pcm (spliced into the aplay argv) --------------------------- #


def test_live_output_pcm_non_string_falls_back() -> None:
    # A non-string reaches subprocess.run as an argv element and raises TypeError,
    # silencing every reply; reject it at load so it falls back to the ALSA default.
    for bad in (True, 5, [], {"a": 1}):
        cfg, err = validate({"live_output_pcm": bad})
        check(
            f"live_output_pcm {bad!r} -> default (non-string)",
            cfg["live_output_pcm"] == DEFAULTS["live_output_pcm"]
            and "live_output_pcm" in err,
            f"got {cfg['live_output_pcm']!r}",
        )


def test_live_output_pcm_string_kept() -> None:
    for good in ("", "plughw:CARD=PowerConf,DEV=0"):
        cfg, err = validate({"live_output_pcm": good})
        check(
            f"live_output_pcm {good!r} kept",
            cfg["live_output_pcm"] == good and err == "",
            f"got {cfg['live_output_pcm']!r}, no warning: {err == ''}",
        )


# --- wake_word resolution ---------------------------------------------------- #


def test_wake_word_unknown_falls_back(monkeypatch_lib) -> None:
    monkeypatch_lib({"hey_jarvis": "/x", "alexa": "/y"})
    cfg, err = validate({"wake_word": "not_a_real_model"})
    check(
        "unknown wake_word -> default (lib non-empty)",
        cfg["wake_word"] == DEFAULTS["wake_word"] and "wake_word" in err,
        f"got {cfg['wake_word']!r}, stderr names key: {'wake_word' in err}",
    )


def test_wake_word_known_kept(monkeypatch_lib) -> None:
    monkeypatch_lib({"hey_jarvis": "/x", "alexa": "/y"})
    cfg, err = validate({"wake_word": "alexa"})
    check(
        "known wake_word kept",
        cfg["wake_word"] == "alexa" and err == "",
        f"got {cfg['wake_word']!r}, no warning: {err == ''}",
    )


def test_wake_word_empty_lib_deferred(monkeypatch_lib) -> None:
    # A fresh clone with no installed models cannot validate the word; load must not
    # rewrite it or warn, leaving detect_wake to surface the real error at use.
    monkeypatch_lib({})
    cfg, err = validate({"wake_word": "whatever_custom"})
    check(
        "empty lib defers wake_word (no rewrite, no warning)",
        cfg["wake_word"] == "whatever_custom" and err == "",
        f"got {cfg['wake_word']!r}, no warning: {err == ''}",
    )


def test_wake_word_non_string_falls_back(monkeypatch_lib) -> None:
    # A non-string can never name a model, so it falls back regardless of the library, and
    # the unhashable array/object case must not reach the membership lookup and raise.
    monkeypatch_lib({"hey_jarvis": "/x", "alexa": "/y"})
    for bad in ([], {}, 7):
        cfg, err = validate({"wake_word": bad})
        check(
            f"wake_word {bad!r} -> default (non-string)",
            cfg["wake_word"] == DEFAULTS["wake_word"] and "wake_word" in err,
            f"got {cfg['wake_word']!r}",
        )


# --- integration: load_config wires validation ------------------------------- #


def test_load_config_integration(tmp_dir) -> None:
    bad = tmp_dir / "config.json"
    bad.write_text(json.dumps({"wake_threshold": 5, "claude_timeout_s": 0}))
    orig_cfg, orig_local = pipeline.CONFIG_PATH, pipeline.LOCAL_CONFIG_PATH
    pipeline.CONFIG_PATH = bad
    pipeline.LOCAL_CONFIG_PATH = tmp_dir / "config.local.json"  # absent
    try:
        buf = io.StringIO()
        with contextlib.redirect_stderr(buf):
            cfg = pipeline.load_config()
        err = buf.getvalue()
    finally:
        pipeline.CONFIG_PATH, pipeline.LOCAL_CONFIG_PATH = orig_cfg, orig_local
    check(
        "load_config falls both bad keys back to defaults",
        cfg["wake_threshold"] == DEFAULTS["wake_threshold"]
        and cfg["claude_timeout_s"] == DEFAULTS["claude_timeout_s"]
        and "wake_threshold" in err
        and "claude_timeout_s" in err,
        f"wake_threshold={cfg['wake_threshold']!r}, claude_timeout_s={cfg['claude_timeout_s']!r}",
    )


def main() -> int:
    saved = pipeline.available_wake_models

    def monkeypatch_lib(lib: dict) -> None:
        pipeline.available_wake_models = lambda: lib

    import tempfile

    try:
        test_wake_threshold_out_of_range()
        test_wake_threshold_valid_untouched()
        test_capture_vad_threshold_sibling()
        test_stt_max_no_speech_prob_sibling()
        test_timeout_bad_values()
        test_timeout_valid_untouched()
        test_log_level_validation()
        test_whisper_compute_unknown()
        test_whisper_compute_valid_untouched()
        test_whisper_compute_unsupported_backend_type_falls_back()
        test_whisper_compute_non_string_falls_back()
        test_whisper_cpu_threads_validation()
        test_whisper_vad_filter_validation()
        test_live_output_pcm_non_string_falls_back()
        test_live_output_pcm_string_kept()
        test_wake_word_unknown_falls_back(monkeypatch_lib)
        test_wake_word_known_kept(monkeypatch_lib)
        test_wake_word_empty_lib_deferred(monkeypatch_lib)
        test_wake_word_non_string_falls_back(monkeypatch_lib)
        with tempfile.TemporaryDirectory() as d:
            test_load_config_integration(Path(d))
    finally:
        pipeline.available_wake_models = saved

    failed = [n for r, n in results if r == FAIL]
    print(f"\n{len(results) - len(failed)}/{len(results)} checks passed")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())