# -*- coding: utf-8 -*- """ Tests für das Vosk STT Plugin. """ import pytest from pathlib import Path from unittest.mock import MagicMock, AsyncMock, patch # Plugin-Module importieren import sys import importlib.util # Plugin-main.py direkt laden plugin_path = Path(__file__).parent.parent / "main.py" spec = importlib.util.spec_from_file_location("vosk_plugin", plugin_path) vosk_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(vosk_module) VoskSTTPlugin = vosk_module.VoskSTTPlugin VoskSTTProvider = vosk_module.VoskSTTProvider VOSK_MODELS = vosk_module.VOSK_MODELS from trixy_core.audio.stt import STTConfig, STTState class TestVoskModelsCatalog: """Tests für den Vosk-Modell-Katalog.""" def test_models_catalog_not_empty(self): """Prüft, dass der Modellkatalog nicht leer ist.""" assert len(VOSK_MODELS) > 0 def test_models_have_required_fields(self): """Prüft, dass alle Modelle die erforderlichen Felder haben.""" required_fields = ["name", "url", "size", "description"] for model_id, model_info in VOSK_MODELS.items(): for field in required_fields: assert field in model_info, f"Modell {model_id} fehlt Feld '{field}'" def test_german_models_available(self): """Prüft, dass deutsche Modelle verfügbar sind.""" assert "de" in VOSK_MODELS assert "de-small" in VOSK_MODELS def test_model_urls_are_valid(self): """Prüft, dass alle Modell-URLs gültige URLs sind.""" for model_id, model_info in VOSK_MODELS.items(): url = model_info["url"] assert url.startswith("https://"), f"Modell {model_id} hat ungültige URL" assert ".zip" in url, f"Modell {model_id} URL sollte ein ZIP-Archiv sein" class TestVoskSTTProvider: """Tests für den VoskSTTProvider.""" @pytest.fixture def provider(self, tmp_path): """Erstellt einen Provider für Tests.""" config = STTConfig(language="de-DE") return VoskSTTProvider( config=config, models_dir=tmp_path / "models", model_name="de-small", auto_download=False, ) def test_provider_name(self, provider): """Prüft den Provider-Namen.""" assert provider.name == "vosk" def test_supported_languages(self, provider): """Prüft die unterstützten Sprachen.""" languages = provider.supported_languages assert "de-DE" in languages assert "en-US" in languages def test_supports_streaming(self, provider): """Prüft Streaming-Unterstützung.""" assert provider.supports_streaming is True def test_supports_word_timings(self, provider): """Prüft Wort-Timings-Unterstützung.""" assert provider.supports_word_timings is True def test_initial_state(self, provider): """Prüft den initialen Zustand.""" assert provider._model_loaded is False assert provider._model is None assert provider._recognizer is None @pytest.mark.asyncio async def test_initialize_without_model_raises(self, provider): """Prüft, dass initialize ohne Modell fehlschlägt.""" with pytest.raises(RuntimeError): await provider.initialize() @pytest.mark.asyncio async def test_transcribe_without_init_raises(self, provider, sample_audio_16khz): """Prüft, dass transcribe ohne Initialisierung fehlschlägt.""" with pytest.raises(RuntimeError, match="Vosk-Modell nicht geladen"): await provider.transcribe(sample_audio_16khz) def test_calculate_confidence_empty(self, provider): """Prüft die Confidence-Berechnung bei leerem Ergebnis.""" result = {"text": ""} assert provider._calculate_confidence(result) == 0.0 def test_calculate_confidence_with_words(self, provider): """Prüft die Confidence-Berechnung mit Wörtern.""" result = { "result": [ {"word": "hallo", "conf": 0.9}, {"word": "welt", "conf": 0.8}, ] } confidence = provider._calculate_confidence(result) assert confidence == pytest.approx(0.85) class TestVoskSTTPlugin: """Tests für das VoskSTTPlugin.""" @pytest.fixture def plugin(self, mock_application, tmp_path): """Erstellt ein Plugin für Tests.""" plugin_path = tmp_path / "stt_vosk" plugin_path.mkdir() (plugin_path / "models").mkdir() config = { "name": "Vosk STT", "enabled": True, "model": "de-small", "language": "de-DE", "auto_download": False, } return VoskSTTPlugin(mock_application, plugin_path, config) def test_plugin_initialization(self, plugin): """Prüft die Plugin-Initialisierung.""" assert plugin._provider is None assert plugin.config.get("model") == "de-small" def test_plugin_path(self, plugin): """Prüft den Plugin-Pfad.""" assert plugin.plugin_path.exists() @pytest.mark.asyncio async def test_on_unload_without_provider(self, plugin): """Prüft on_unload ohne initialisierten Provider.""" await plugin.on_unload() assert plugin._provider is None class TestVoskDownload: """Tests für die Download-Funktionalität.""" @pytest.mark.asyncio async def test_download_model_unknown_model_raises(self, tmp_path): """Prüft, dass unbekannte Modelle einen Fehler werfen.""" config = STTConfig(language="de-DE") provider = VoskSTTProvider( config=config, models_dir=tmp_path / "models", model_name="unknown_model", auto_download=True, ) with pytest.raises(RuntimeError, match="Unbekanntes Modell"): await provider.initialize() class TestVoskEventHandling: """Tests für die Event-Verarbeitung.""" @pytest.fixture def plugin_with_provider(self, mock_application, tmp_path): """Erstellt ein Plugin mit Mock-Provider.""" plugin_path = tmp_path / "stt_vosk" plugin_path.mkdir() (plugin_path / "models").mkdir() config = { "name": "Vosk STT", "enabled": True, "model": "de-small", "language": "de-DE", "auto_download": False, } plugin = VoskSTTPlugin(mock_application, plugin_path, config) # Mock-Provider setzen mock_provider = MagicMock() mock_provider.is_ready = True mock_provider.transcribe = AsyncMock(return_value=MagicMock( text="test text", confidence=0.95, language="de-DE", duration_seconds=1.0, processing_time_ms=100, word_timings=[], )) plugin._provider = mock_provider return plugin # Integration Tests (benötigen echte Installation) @pytest.mark.integration @pytest.mark.requires_vosk class TestVoskIntegration: """Integrationstests für Vosk STT (benötigen Installation).""" @pytest.fixture def provider_with_model(self, tmp_path): """Erstellt einen Provider mit echtem Modell.""" config = STTConfig(language="de-DE") return VoskSTTProvider( config=config, models_dir=tmp_path / "models", model_name="de-small", auto_download=True, ) @pytest.mark.asyncio @pytest.mark.slow async def test_full_transcription_flow(self, provider_with_model, sample_audio_16khz): """Testet den vollständigen Transkriptions-Flow.""" await provider_with_model.initialize() result = await provider_with_model.transcribe(sample_audio_16khz) assert result.text is not None assert result.provider == "vosk" assert result.processing_time_ms > 0 await provider_with_model.shutdown()