STT-Assistant/tests/test_config.py

"""Tests for configuration loading."""

import importlib.util
import tempfile
import tomllib
from pathlib import Path

from stt_daemon.config import load_config

REPO_ROOT = Path(__file__).resolve().parents[1]
LAYA_REV = "970dc8c5f63d7b886a68409493f37d569424f933"


def test_load_config_defaults():
    """Test that load_config returns defaults when no file exists."""
    with tempfile.TemporaryDirectory() as tmpdir:
        # Change to temp dir and load (no config.toml exists)
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
            assert cfg.audio.backend == "parec"
            assert cfg.audio.device == "VirtualMicSink.monitor"
            assert cfg.audio.sample_rate == 16000
            assert cfg.vad.aggressiveness == 2
            assert cfg.whisper.model == "small"
            assert cfg.whisper.language is None
            assert cfg.output.stdout is True
            assert cfg.output.jsonl is False
        finally:
            os.chdir(old_cwd)


def test_load_config_from_file():
    """Test loading config from a TOML file."""
    config_toml = """
[audio]
backend = "sounddevice"
device = "hw:0"
sample_rate = 44100

[vad]
aggressiveness = 3
pre_roll_ms = 500

[whisper]
model = "base"
device = "cuda"
language = "es"

[output]
stdout = false
jsonl = true
jsonl_dir = "logs"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

        assert cfg.audio.backend == "sounddevice"
        assert cfg.audio.device == "hw:0"
        assert cfg.audio.sample_rate == 44100
        assert cfg.vad.aggressiveness == 3
        assert cfg.vad.pre_roll_ms == 500
        assert cfg.whisper.model == "base"
        assert cfg.whisper.device == "cuda"
        assert cfg.whisper.language == "es"
        assert cfg.output.stdout is False
        assert cfg.output.jsonl is True
        assert cfg.output.jsonl_dir == "logs"


def test_language_normalization_auto():
    """Test that language 'auto' is normalized to None."""
    config_toml = """
[whisper]
language = "auto"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))
        assert cfg.whisper.language is None


def test_language_normalization_empty_string():
    """Test that language '' is normalized to None."""
    config_toml = """
[whisper]
language = ""
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))
        assert cfg.whisper.language is None


def test_language_normalization_none():
    """Test that language 'none' is normalized to None."""
    config_toml = """
[whisper]
language = "none"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))
        assert cfg.whisper.language is None


def test_language_explicit_en():
    """Test that explicit language codes are preserved."""
    config_toml = """
[whisper]
language = "en"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))
        assert cfg.whisper.language == "en"


def test_partial_config():
    """Test that partial configs merge with defaults."""
    config_toml = """
[audio]
device = "custom_device"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

        # Overridden
        assert cfg.audio.device == "custom_device"

        # Defaults
        assert cfg.audio.backend == "parec"
        assert cfg.audio.sample_rate == 16000
        assert cfg.vad.aggressiveness == 2


def test_chunking_and_confidence_fields():
    """New chunking/gate fields load from TOML."""
    config_toml = """
[vad]
chunk_pause_ms = 500
soft_max_s = 8.0
rms_floor = 0.01
trailing_trim_ms = 150

[whisper]
no_speech_prob_max = 0.7
avg_logprob_min = -0.8
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.vad.chunk_pause_ms == 500
    assert cfg.vad.soft_max_s == 8.0
    assert cfg.vad.rms_floor == 0.01
    assert cfg.vad.trailing_trim_ms == 150
    assert cfg.whisper.no_speech_prob_max == 0.7
    assert cfg.whisper.avg_logprob_min == -0.8


def test_chunking_defaults():
    """New fields have sane defaults."""
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.vad.chunk_pause_ms == 350
    assert cfg.vad.soft_max_s == 10.0
    assert cfg.vad.rms_floor == 0.008
    assert cfg.vad.trailing_trim_ms == 200
    assert cfg.whisper.no_speech_prob_max == 0.6
    assert cfg.whisper.avg_logprob_min == -1.0


def test_output_defaults_disable_audio_saving():
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.output.save_audio is False
    assert cfg.output.audio_dir == "audio"


def test_output_save_audio_fields():
    config_toml = """
[output]
save_audio = true
audio_dir = "clips"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.output.save_audio is True
    assert cfg.output.audio_dir == "clips"


def test_daemon_defaults():
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.daemon.queue_size == 50
    assert cfg.daemon.backlog_warn_at == 5


def test_daemon_config_fields():
    config_toml = """
[daemon]
queue_size = 20
backlog_warn_at = 3
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.daemon.queue_size == 20
    assert cfg.daemon.backlog_warn_at == 3


def test_unknown_sections_warn_and_are_ignored(caplog):
    """A typo'd section like [assitant] must warn, not silently use defaults."""
    config_toml = """
[assitant]
require_viewer = false
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        with caplog.at_level("WARNING", logger="stt_daemon.config"):
            cfg = load_config(str(config_path))

    assert "assitant" in caplog.text
    assert cfg.assistant.require_viewer is True


def test_cpu_threads_default_is_auto():
    """cpu_threads defaults to 0 (CTranslate2's automatic choice)."""
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.whisper.cpu_threads == 0


def test_cpu_threads_config_field():
    config_toml = """
[whisper]
cpu_threads = 16
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.whisper.cpu_threads == 16


def test_language_strategy_defaults_to_static():
    """language_strategy defaults to 'static'."""
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.whisper.language_strategy == "static"


def test_language_strategy_sticky_is_parsed():
    """language_strategy 'sticky' is parsed correctly."""
    config_toml = """
[whisper]
language_strategy = "sticky"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.whisper.language_strategy == "sticky"


def test_unknown_language_strategy_falls_back_to_static(caplog):
    """An unknown language_strategy value falls back to 'static' with a warning."""
    config_toml = """
[whisper]
language_strategy = "bogus"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        with caplog.at_level("WARNING", logger="stt_daemon.config"):
            cfg = load_config(str(config_path))

    assert cfg.whisper.language_strategy == "static"
    assert "Unknown language_strategy" in caplog.text
    assert "bogus" in caplog.text


def test_sticky_redetect_logprob_defaults_to_minus_0_8():
    """sticky_redetect_logprob defaults to -0.8."""
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.whisper.sticky_redetect_logprob == -0.8


def test_sticky_redetect_logprob_is_parsed():
    """sticky_redetect_logprob is parsed from TOML."""
    config_toml = """
[whisper]
sticky_redetect_logprob = -0.5
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.whisper.sticky_redetect_logprob == -0.5


def test_llm_context_and_compaction_defaults():
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.llm.max_context_tokens == 131072
    assert cfg.llm.compaction_safety_margin == 256
    assert cfg.llm.compaction_enabled is True
    assert cfg.llm.compaction_trigger_ratio == 0.8
    assert cfg.llm.compaction_keep_turns == 6
    assert cfg.llm.compaction_max_tokens == 512
    assert cfg.llm.compaction_timeout_s == 30.0
    assert cfg.llm.compaction_input_tokens == 8192
    assert cfg.llm.inference_queue_size == 8
    assert cfg.llm.prompt_budget == 131072 - 1024 - 256
    assert not hasattr(cfg.llm, "max_prompt_tokens")


def test_llm_context_and_compaction_fields_parsed():
    config_toml = """
[llm]
max_context_tokens = 8192
max_tokens = 2048
compaction_safety_margin = 128
compaction_enabled = false
compaction_trigger_ratio = 0.5
compaction_keep_turns = 2
compaction_max_tokens = 256
compaction_timeout_s = 12.5
compaction_input_tokens = 4096
inference_queue_size = 3
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.llm.max_context_tokens == 8192
    assert cfg.llm.compaction_safety_margin == 128
    assert cfg.llm.compaction_enabled is False
    assert cfg.llm.compaction_trigger_ratio == 0.5
    assert cfg.llm.compaction_keep_turns == 2
    assert cfg.llm.compaction_max_tokens == 256
    assert cfg.llm.compaction_timeout_s == 12.5
    assert cfg.llm.compaction_input_tokens == 4096
    assert cfg.llm.inference_queue_size == 3
    assert cfg.llm.prompt_budget == 8192 - 2048 - 128


def test_max_prompt_tokens_is_deprecated_and_ignored(caplog):
    config_toml = """
[llm]
max_prompt_tokens = 50
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        with caplog.at_level("WARNING", logger="stt_daemon.config"):
            cfg = load_config(str(config_path))

    assert "max_prompt_tokens" in caplog.text
    assert "deprecated" in caplog.text
    assert cfg.llm.prompt_budget == 131072 - 1024 - 256


def test_backend_defaults_to_faster_whisper():
    """backend defaults to 'faster-whisper'."""
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.whisper.backend == "faster-whisper"


def test_backend_pywhispercpp_is_parsed():
    """backend 'pywhispercpp' is parsed from TOML."""
    config_toml = """
[whisper]
backend = "pywhispercpp"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.whisper.backend == "pywhispercpp"


def test_backend_is_normalized():
    """Backend names are normalized: stripped, lowercased, underscores to dashes."""
    config_toml = """
[whisper]
backend = "  Faster_Whisper  "
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.whisper.backend == "faster-whisper"


def test_unknown_backend_falls_back_to_faster_whisper(caplog):
    """An unknown backend value falls back to 'faster-whisper' with a warning."""
    config_toml = """
[whisper]
backend = "bogus"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        with caplog.at_level("WARNING", logger="stt_daemon.config"):
            cfg = load_config(str(config_path))

    assert cfg.whisper.backend == "faster-whisper"
    assert "Unknown whisper backend" in caplog.text
    assert "bogus" in caplog.text


def test_triage_defaults():
    """The [triage] section defaults match the plan's surviving-key list."""
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.triage.enabled is False
    assert cfg.triage.questions_file == "questions.json"
    assert cfg.triage.window == 500
    assert cfg.triage.state_char_budget == 3000
    assert cfg.triage.predict_max_tokens == 1024
    assert cfg.triage.confidence_field == "answer_confidence"
    assert cfg.triage.escalate_threshold == 0.6
    assert cfg.triage.choice_min_confidence == 0.85
    assert cfg.triage.extract_enabled is True
    assert cfg.triage.extract_max_tokens == 512
    assert cfg.triage.extract_timeout_s == 30.0
    assert cfg.triage.device == ""
    assert cfg.triage.preload == ["multilingual"]
    assert cfg.triage.default_checkpoint == "multilingual"


def test_triage_config_fields_parsed():
    """Every surviving [triage] key is applied from TOML."""
    config_toml = """
[triage]
enabled = true
questions_file = "custom_questions.json"
window = 4
state_char_budget = 800
predict_max_tokens = 768
confidence_field = "confidence"
escalate_threshold = 0.7
choice_min_confidence = 0.9
extract_enabled = false
extract_max_tokens = 256
extract_timeout_s = 12.5
device = "cuda"
preload = ["multilingual", "english"]
default_checkpoint = "english"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.triage.enabled is True
    assert cfg.triage.questions_file == "custom_questions.json"
    assert cfg.triage.window == 4
    assert cfg.triage.state_char_budget == 800
    assert cfg.triage.predict_max_tokens == 768
    assert cfg.triage.confidence_field == "confidence"
    assert cfg.triage.escalate_threshold == 0.7
    assert cfg.triage.choice_min_confidence == 0.9
    assert cfg.triage.extract_enabled is False
    assert cfg.triage.extract_max_tokens == 256
    assert cfg.triage.extract_timeout_s == 12.5
    assert cfg.triage.device == "cuda"
    assert cfg.triage.preload == ["multilingual", "english"]
    assert cfg.triage.default_checkpoint == "english"


def test_unknown_triag_section_warns_and_keeps_defaults(caplog):
    """A typo'd [triag] section must warn and never raise; triage keeps defaults."""
    config_toml = """
[triag]
enabled = true
window = 3
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        with caplog.at_level("WARNING", logger="stt_daemon.config"):
            cfg = load_config(str(config_path))

    assert "triag" in caplog.text
    assert cfg.triage.enabled is False
    assert cfg.triage.window == 500


def test_triage_config_has_no_struck_keys():
    """The v5 keys are deliberately gone from TriageConfig (superseded/removed)."""
    from stt_daemon.config import TriageConfig

    struck = {"socket", "model", "predict_timeout_s", "checkpoint", "subfolder", "max_len"}
    for name in struck:
        assert not hasattr(TriageConfig, name), f"TriageConfig must not expose {name!r}"


def test_llm_triage_system_prompt_is_parsed():
    """[llm].triage_system_prompt overrides the JSON-only extractor default."""
    config_toml = """
[llm]
triage_system_prompt = "ONLY JSON, nothing else."
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.llm.triage_system_prompt == "ONLY JSON, nothing else."


def test_no_stt_laya_script():
    """Triage is in-process only: no stt-laya console script may ever be added."""
    with (REPO_ROOT / "pyproject.toml").open("rb") as fh:
        data = tomllib.load(fh)

    scripts = data["project"]["scripts"]
    assert set(scripts) == {"stt-daemon", "stt-assistant", "stt-ctl", "stt-tui"}
    assert not any("laya" in name for name in scripts)


def test_layasvc_module_not_importable():
    """The retired external laya service package must not exist (in-process lane)."""
    assert importlib.util.find_spec("stt_daemon.layasvc") is None


def test_triage_extra_pins_laya_revision():
    """The triage extra pins laya to the exact verified upstream revision."""
    with (REPO_ROOT / "pyproject.toml").open("rb") as fh:
        data = tomllib.load(fh)

    assert data["project"]["optional-dependencies"]["triage"] == ["laya"]

    source = data["tool"]["uv"]["sources"]["laya"]
    assert source["git"] == "https://github.com/NandhaKishorM/laya.git"
    assert source["rev"] == LAYA_REV


def test_questions_json_gitignored():
    """The user's private questions.json is never committed (like config.toml)."""
    lines = (REPO_ROOT / ".gitignore").read_text().splitlines()
    assert "questions.json" in lines


def test_speaker_defaults():
    """The [speaker] section defaults match the plan's 14-key list."""
    with tempfile.TemporaryDirectory() as tmpdir:
        import os
        old_cwd = os.getcwd()
        try:
            os.chdir(tmpdir)
            cfg = load_config()
        finally:
            os.chdir(old_cwd)

    assert cfg.speaker.enabled is False
    assert cfg.speaker.model_path is None
    assert cfg.speaker.provider == "cpu"
    assert cfg.speaker.threshold_id == 0.60
    assert cfg.speaker.threshold_new == 0.45
    assert cfg.speaker.min_enroll_s == 2.0
    assert cfg.speaker.rms_floor == 0.0
    assert cfg.speaker.margin == 0.04
    assert cfg.speaker.mint_cooldown_s == 60.0
    assert cfg.speaker.confirm_s == 12.0
    assert cfg.speaker.max_prototypes == 8
    assert cfg.speaker.provisional_ttl_h == 24.0
    assert cfg.speaker.store_dir == "speakers"
    assert cfg.speaker.join_timeout_s == 2.0


def test_speaker_config_fields_parsed():
    """Every [speaker] key is applied from TOML."""
    config_toml = """
[speaker]
enabled = true
model_path = "models/nemo_en_titanet_small.onnx"
provider = "cuda"
threshold_id = 0.7
threshold_new = 0.5
min_enroll_s = 3.0
rms_floor = 0.01
margin = 0.05
mint_cooldown_s = 30.0
confirm_s = 8.0
max_prototypes = 4
provisional_ttl_h = 12.0
store_dir = "voiceprints"
join_timeout_s = 1.5
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.speaker.enabled is True
    assert cfg.speaker.model_path == "models/nemo_en_titanet_small.onnx"
    assert cfg.speaker.provider == "cuda"
    assert cfg.speaker.threshold_id == 0.7
    assert cfg.speaker.threshold_new == 0.5
    assert cfg.speaker.min_enroll_s == 3.0
    assert cfg.speaker.rms_floor == 0.01
    assert cfg.speaker.margin == 0.05
    assert cfg.speaker.mint_cooldown_s == 30.0
    assert cfg.speaker.confirm_s == 8.0
    assert cfg.speaker.max_prototypes == 4
    assert cfg.speaker.provisional_ttl_h == 12.0
    assert cfg.speaker.store_dir == "voiceprints"
    assert cfg.speaker.join_timeout_s == 1.5


def test_unknown_speaker_key_is_ignored():
    """An unknown key inside [speaker] is silently ignored (per-field .get)."""
    config_toml = """
[speaker]
enabled = true
bogus_key = 42
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        cfg = load_config(str(config_path))

    assert cfg.speaker.enabled is True
    assert not hasattr(cfg.speaker, "bogus_key")
    assert cfg.speaker.provider == "cpu"


def test_unknown_speaker_provider_falls_back_to_cpu(caplog):
    """An unknown speaker provider falls back to 'cpu' with a warning."""
    config_toml = """
[speaker]
provider = "bogus"
"""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.toml"
        config_path.write_text(config_toml)

        with caplog.at_level("WARNING", logger="stt_daemon.config"):
            cfg = load_config(str(config_path))

    assert cfg.speaker.provider == "cpu"
    assert "Unknown speaker provider" in caplog.text
    assert "bogus" in caplog.text