Trixy7/plugins/stt_whisper/tests/test_whisper_stt.py

# -*- coding: utf-8 -*-
"""
Tests für das Whisper 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("whisper_plugin", plugin_path)
whisper_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(whisper_module)

WhisperSTTPlugin = whisper_module.WhisperSTTPlugin
WhisperSTTProvider = whisper_module.WhisperSTTProvider
WHISPER_MODELS = whisper_module.WHISPER_MODELS

from trixy_core.audio.stt import STTConfig, STTState


class TestWhisperModelsCatalog:
    """Tests für den Whisper-Modell-Katalog."""

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

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

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

    def test_common_models_available(self):
        """Prüft, dass gängige Modelle verfügbar sind."""
        assert "tiny" in WHISPER_MODELS
        assert "base" in WHISPER_MODELS
        assert "small" in WHISPER_MODELS
        assert "medium" in WHISPER_MODELS
        assert "large" in WHISPER_MODELS

    def test_english_only_models_available(self):
        """Prüft, dass englischsprachige Modelle verfügbar sind."""
        assert "tiny.en" in WHISPER_MODELS
        assert "base.en" in WHISPER_MODELS


class TestWhisperSTTProvider:
    """Tests für den WhisperSTTProvider."""

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

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

    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
        assert "fr-FR" in languages
        assert "ja-JP" in languages

    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

    def test_device_selection_cpu(self, tmp_path):
        """Prüft CPU-Auswahl."""
        config = STTConfig(language="de-DE", use_gpu=False)
        provider = WhisperSTTProvider(
            config=config,
            models_dir=tmp_path / "models",
            model_name="base",
        )
        assert provider._device == "cpu"

    def test_device_selection_gpu(self, tmp_path):
        """Prüft GPU-Auswahl."""
        config = STTConfig(language="de-DE", use_gpu=True)
        provider = WhisperSTTProvider(
            config=config,
            models_dir=tmp_path / "models",
            model_name="base",
        )
        assert provider._device == "cuda"

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


class TestWhisperSTTPlugin:
    """Tests für das WhisperSTTPlugin."""

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

        config = {
            "name": "Whisper STT",
            "enabled": True,
            "model": "base",
            "language": "de-DE",
            "use_gpu": False,
        }

        return WhisperSTTPlugin(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") == "base"

    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


# Integration Tests (benötigen echte Installation)
@pytest.mark.integration
class TestWhisperIntegration:
    """Integrationstests für Whisper STT (benötigen Installation)."""

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

    @pytest.fixture
    def provider_with_model(self, tmp_path):
        """Erstellt einen Provider mit echtem Modell."""
        config = STTConfig(language="de-DE", use_gpu=False)
        return WhisperSTTProvider(
            config=config,
            models_dir=tmp_path / "models",
            model_name="tiny",  # Kleinstes Modell für schnelle Tests
        )

    @pytest.mark.asyncio
    @pytest.mark.slow
    async def test_full_transcription_flow(self, provider_with_model, sample_audio_with_tone):
        """Testet den vollständigen Transkriptions-Flow."""
        await provider_with_model.initialize()

        result = await provider_with_model.transcribe(sample_audio_with_tone)

        assert result.text is not None
        assert result.provider == "whisper"
        assert result.processing_time_ms > 0

        await provider_with_model.shutdown()