"""Tests for PipelineSession._build_turn_detection (Omni turn_detection object).

Like the other server-level tests this imports qwen_pipeline_server, which
requires DASHSCOPE_API_KEY in the environment (or .env) at import time.
"""

from types import SimpleNamespace

import pytest

from qwen_pipeline_server import PipelineSession


def _stub(**overrides):
    base = dict(
        vad_type='server_vad', vad_threshold=0.2, silence_duration_ms=800,
        prefix_padding_ms=None, idle_timeout_ms=None,
        omni_model='qwen3.5-omni-plus-realtime', _omni_requested_td=None,
    )
    base.update(overrides)
    return SimpleNamespace(**base)


def build(**overrides):
    s = _stub(**overrides)
    td = PipelineSession._build_turn_detection(s)
    return td, s


class TestBuildTurnDetection:

    def test_default_server_vad_matches_previous_sdk_wire_shape(self):
        td, s = build()
        assert td == {'type': 'server_vad', 'threshold': 0.2,
                      'silence_duration_ms': 800, 'prefix_padding_ms': 300}
        assert s._omni_requested_td == td
        assert s._omni_requested_td is not td  # snapshot, not alias

    def test_semantic_vad_sends_no_prefix_or_idle(self):
        td, _ = build(vad_type='semantic_vad', prefix_padding_ms=500,
                      idle_timeout_ms=10000)
        assert td == {'type': 'semantic_vad', 'threshold': 0.2,
                      'silence_duration_ms': 800}

    def test_server_vad_extras(self):
        td, _ = build(prefix_padding_ms=500, idle_timeout_ms=10000)
        assert td['prefix_padding_ms'] == 500
        assert td['idle_timeout_ms'] == 10000

    def test_clamping(self):
        td, _ = build(vad_threshold=1.7, silence_duration_ms=100,
                      prefix_padding_ms=9000, idle_timeout_ms=1000)
        assert td['threshold'] == 1.0
        assert td['silence_duration_ms'] == 200
        assert td['prefix_padding_ms'] == 5000
        assert td['idle_timeout_ms'] == 5000
        td, _ = build(vad_threshold=-3)
        assert td['threshold'] == -1.0

    def test_idle_timeout_only_for_qwen35(self):
        td, _ = build(omni_model='qwen3.8-omni-flash-realtime', idle_timeout_ms=10000)
        assert 'idle_timeout_ms' not in td
        td, _ = build(omni_model='qwen3.5-omni-flash-realtime', idle_timeout_ms=10000)
        assert td['idle_timeout_ms'] == 10000

    def test_junk_values_are_dropped_not_fatal(self):
        td, _ = build(vad_threshold='x', silence_duration_ms='', prefix_padding_ms='abc')
        assert td == {'type': 'server_vad', 'prefix_padding_ms': 300}

    def test_string_numbers_accepted(self):
        td, _ = build(vad_threshold='0.4', silence_duration_ms='600',
                      prefix_padding_ms='250')
        assert td['threshold'] == 0.4
        assert td['silence_duration_ms'] == 600
        assert td['prefix_padding_ms'] == 250


class TestNothinkSuffixOrdering:
    """The '-nothink' pseudo-suffix is stripped from omni_model in __init__
    before any model-family check, so idle_timeout gating sees the real id."""

    def test_suffix_stripped_in_init_and_family_check_runs_later(self):
        import inspect
        init_src = inspect.getsource(PipelineSession.__init__)
        strip_at = init_src.index("endswith('-nothink')")
        assert strip_at > 0
        # Nothing in __init__ before the strip inspects the model family by
        # any spelling (literal family prefixes or the builder itself).
        before = '\n'.join(line.split('#', 1)[0]          # code only, no comments
                           for line in init_src[:strip_at].splitlines())
        for needle in ('qwen3.5', 'qwen3.8', '_build_turn_detection'):
            assert needle not in before, f'{needle!r} used before -nothink strip'
        # The family check lives in _build_turn_detection, which __init__
        # never calls, so by control flow it always sees the stripped id.
        assert '_build_turn_detection' not in init_src
        builder_src = inspect.getsource(PipelineSession._build_turn_detection)
        assert "startswith('qwen3.5-omni')" in builder_src

    def test_builder_sees_stripped_model_id(self):
        # Simulates the state __init__ leaves behind for a '-nothink' profile:
        # omni_model already stripped, so the 3.5 gate must pass.
        td, _ = build(omni_model='qwen3.5-omni-plus-realtime', idle_timeout_ms=8000)
        assert td['idle_timeout_ms'] == 8000
        # And an unstripped id would (correctly) fail the gate: this is what
        # the ordering guarantee protects against.
        td, _ = build(omni_model='qwen3.5-omni-plus-realtime-nothink', idle_timeout_ms=8000)
        assert td['idle_timeout_ms'] == 8000  # startswith still matches