Trixy7/plugins/stt_deepspeech/tests/test_deepspeech_stt.py

# -*- coding: utf-8 -*-
"""
Tests für das DeepSpeech STT Plugin.
"""

import pytest
from pathlib import Path
from unittest.mock import MagicMock, AsyncMock, patch
import importlib.util

# Plugin-main.py direkt laden
plugin_path = Path(__file__).parent.parent / "main.py"
spec = importlib.util.spec_from_file_location("deepspeech_plugin", plugin_path)
deepspeech_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(deepspeech_module)

DeepSpeechSTTPlugin = deepspeech_module.DeepSpeechSTTPlugin
DeepSpeechSTTProvider = deepspeech_module.DeepSpeechSTTProvider
DEEPSPEECH_MODELS = deepspeech_module.DEEPSPEECH_MODELS

from trixy_core.audio.stt import STTConfig, STTState


class TestDeepSpeechModelsCatalog:
    """Tests für den DeepSpeech-Modell-Katalog."""

    def test_models_catalog_not_empty(self):
        """Prüft, dass der Modellkatalog nicht leer ist."""
        assert len(DEEPSPEECH_MODELS) > 0

    def test_models_have_required_fields(self):
        """Prüft, dass alle Modelle die erforderlichen Felder haben."""
        required_fields = ["name", "model_url", "scorer_url", "model_file", "scorer_file", "size", "description"]

        for model_id, model_info in DEEPSPEECH_MODELS.items():
            for field in required_fields:
                assert field in model_info, f"Modell {model_id} fehlt Feld '{field}'"

    def test_german_model_available(self):
        """Prüft, dass das deutsche Modell verfügbar ist."""
        assert "de" in DEEPSPEECH_MODELS

    def test_english_model_available(self):
        """Prüft, dass das englische Modell verfügbar ist."""
        assert "en" in DEEPSPEECH_MODELS

    def test_model_urls_are_valid(self):
        """Prüft, dass alle Modell-URLs gültige URLs sind."""
        for model_id, model_info in DEEPSPEECH_MODELS.items():
            model_url = model_info["model_url"]
            scorer_url = model_info["scorer_url"]
            assert model_url.startswith("https://"), f"Modell {model_id} hat ungültige Model-URL"
            assert scorer_url.startswith("https://"), f"Modell {model_id} hat ungültige Scorer-URL"


class TestDeepSpeechSTTProvider:
    """Tests für den DeepSpeechSTTProvider."""

    @pytest.fixture
    def provider(self, tmp_path):
        """Erstellt einen Provider für Tests."""
        config = STTConfig(language="de-DE")
        return DeepSpeechSTTProvider(
            config=config,
            models_dir=tmp_path / "models",
            model_name="de",
            auto_download=False,
        )

    def test_provider_name(self, provider):
        """Prüft den Provider-Namen."""
        assert provider.name == "deepspeech"

    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

    @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="DeepSpeech-Modell nicht geladen"):
            await provider.transcribe(sample_audio_16khz)


class TestDeepSpeechSTTPlugin:
    """Tests für das DeepSpeechSTTPlugin."""

    @pytest.fixture
    def plugin(self, mock_application, tmp_path):
        """Erstellt ein Plugin für Tests."""
        plugin_path = tmp_path / "stt_deepspeech"
        plugin_path.mkdir()
        (plugin_path / "models").mkdir()

        config = {
            "name": "DeepSpeech STT",
            "enabled": True,
            "model": "de",
            "language": "de-DE",
            "auto_download": False,
        }

        return DeepSpeechSTTPlugin(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"

    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 TestDeepSpeechDownload:
    """Tests für die Download-Funktionalität."""

    @pytest.fixture
    def mock_event_manager(self):
        """Erstellt einen Mock-EventManager."""
        em = MagicMock()
        em.emit = AsyncMock()
        return em

    @pytest.mark.asyncio
    async def test_download_unknown_model_raises(self, tmp_path):
        """Prüft, dass unbekannte Modelle einen Fehler werfen."""
        config = STTConfig(language="de-DE")
        provider = DeepSpeechSTTProvider(
            config=config,
            models_dir=tmp_path / "models",
            model_name="unknown_model",
            auto_download=True,
        )

        # Entweder "Unbekanntes Modell" (wenn deepspeech installiert)
        # oder "nicht installiert" (wenn deepspeech fehlt)
        with pytest.raises(RuntimeError, match="(Unbekanntes Modell|nicht installiert)"):
            await provider.initialize()

    def test_model_paths_configured(self, tmp_path):
        """Prüft, dass Modellpfade korrekt konfiguriert werden."""
        config = STTConfig(language="de-DE")
        provider = DeepSpeechSTTProvider(
            config=config,
            models_dir=tmp_path / "models",
            model_name="de",
            auto_download=False,
        )

        # Nach dem Setzen der model_name sollten die Pfade noch None sein
        # bis initialize() aufgerufen wird
        assert provider._model_path is None
        assert provider._scorer_path is None


class TestDeepSpeechEventHandling:
    """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_deepspeech"
        plugin_path.mkdir()
        (plugin_path / "models").mkdir()

        config = {
            "name": "DeepSpeech STT",
            "enabled": True,
            "model": "de",
            "language": "de-DE",
            "auto_download": False,
        }

        plugin = DeepSpeechSTTPlugin(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
class TestDeepSpeechIntegration:
    """Integrationstests für DeepSpeech STT (benötigen Installation)."""

    @pytest.fixture(autouse=True)
    def skip_if_not_installed(self):
        """Überspringt Tests wenn deepspeech nicht installiert ist."""
        pytest.importorskip("deepspeech", reason="deepspeech nicht installiert")

    @pytest.fixture
    def provider_with_model(self, tmp_path):
        """Erstellt einen Provider mit echtem Modell."""
        config = STTConfig(language="de-DE")
        return DeepSpeechSTTProvider(
            config=config,
            models_dir=tmp_path / "models",
            model_name="de",
            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 == "deepspeech"
        assert result.processing_time_ms > 0

        await provider_with_model.shutdown()