Trixy7/plugins/stt_vosk/tests/test_vosk_stt.py

# -*- 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()