98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
"""Environment YAML selects VAD and supplies the official iterator options."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from voice_assistant.config.loader import load_config, load_provider_catalog
|
|
from voice_assistant.providers import create_vad_adapter
|
|
from voice_assistant.vad import silero
|
|
from voice_assistant.vad.silero import SileroVADAdapter
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
def configuration_directory():
|
|
return Path(__file__).resolve().parents[1] / "config"
|
|
|
|
|
|
def test_development_yaml_declares_vad_parameters():
|
|
config = load_config(environment="development", config_dir=configuration_directory())
|
|
assert config.vad is not None
|
|
assert config.vad.provider == "silero"
|
|
assert config.vad.options == {
|
|
"threshold": 0.5,
|
|
"min_silence_duration_ms": 300,
|
|
"speech_pad_ms": 30,
|
|
"worker_timeout_seconds": 10.0,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_active_environment_options_reach_adapter_and_iterator(tmp_path, monkeypatch):
|
|
options = {
|
|
"threshold": 0.65,
|
|
"min_silence_duration_ms": 500,
|
|
"speech_pad_ms": 50,
|
|
"worker_timeout_seconds": 2.5,
|
|
}
|
|
(tmp_path / "staging.yaml").write_text(yaml.safe_dump({
|
|
"stt": {"provider": "nvidia"},
|
|
"tts": {"provider": "voxcpm"},
|
|
"agent": {"provider": "mock"},
|
|
"vad": {"provider": "silero", "options": options},
|
|
}))
|
|
monkeypatch.setenv("APP_ENV", "staging")
|
|
received_options = []
|
|
iterator_options = []
|
|
|
|
class Adapter(SileroVADAdapter):
|
|
def __init__(self, **kwargs):
|
|
received_options.append(kwargs)
|
|
super().__init__(**kwargs)
|
|
|
|
class Iterator:
|
|
def __call__(self, samples, *, return_seconds):
|
|
return None
|
|
|
|
def reset_states(self):
|
|
pass
|
|
|
|
def load_iterator(*args):
|
|
iterator_options.append(args)
|
|
return Iterator()
|
|
|
|
monkeypatch.setattr(silero, "SileroVADAdapter", Adapter)
|
|
monkeypatch.setattr(silero, "_load_iterator", load_iterator)
|
|
config = load_config(config_dir=tmp_path)
|
|
catalog = load_provider_catalog(config_dir=configuration_directory())
|
|
detector = create_vad_adapter(config, catalog)
|
|
try:
|
|
assert await detector.process(bytes(1024)) == ()
|
|
assert received_options == [options]
|
|
assert iterator_options == [(0.65, 500, 50)]
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
def test_missing_vad_preserves_config_loading_but_rejects_explicit_creation():
|
|
config = load_config(environment="development", config_dir=configuration_directory())
|
|
legacy = config.model_copy(update={"vad": None})
|
|
catalog = load_provider_catalog(config_dir=configuration_directory())
|
|
with pytest.raises(ValueError, match="not configured"):
|
|
create_vad_adapter(legacy, catalog)
|
|
|
|
|
|
def test_invalid_yaml_option_is_rejected_by_adapter_constructor():
|
|
config = load_config(environment="development", config_dir=configuration_directory())
|
|
bad_config = config.model_copy(update={
|
|
"vad": config.vad.model_copy(update={"options": {"threshold": 2.0}}),
|
|
})
|
|
catalog = load_provider_catalog(config_dir=configuration_directory())
|
|
with pytest.raises(ValueError, match="threshold"):
|
|
create_vad_adapter(bad_config, catalog)
|