Trixy7/plugins/voice_morph/tests/test_voice_morph.py

# -*- coding: utf-8 -*-
"""
Tests fuer das VoiceMorph Plugin.
"""

import asyncio
import numpy as np
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from trixy_core.nlp.handler import IntentReceivedData, IntentResult
from trixy_core.audio_morpher.presets import VoiceMorphPreset
from plugins.voice_morph.main import VoiceMorphPlugin


# ============================================================================
# Fixtures
# ============================================================================

def _make_plugin(tmp_path, config_overrides=None) -> tuple[VoiceMorphPlugin, MagicMock]:
    """Erstellt ein VoiceMorphPlugin mit Mock-Application."""
    mock_app = MagicMock()
    mock_app.events = MagicMock()
    mock_app.events.emit = AsyncMock()
    mock_app.events.trigger = AsyncMock()
    mock_app.events.register = MagicMock()
    mock_app.events.unregister = MagicMock()
    mock_app.debug = True
    mock_app.services = MagicMock()
    mock_app.services.get_service = MagicMock(return_value=None)

    config = {
        "enabled": True,
        "active_preset": "test_preset",
        "intensity": 1.0,
        "presets": {
            "test_preset": {
                "label": "Test Preset",
                "pitch": {"semitones": 4.0},
                "speed": {"factor": 0.95},
            },
            "robot": {
                "label": "Robot Stimme",
                "pitch": {"semitones": -2.0},
                "distortion": {"gain": 1.5},
            },
        },
    }
    if config_overrides:
        config.update(config_overrides)

    plugin_path = tmp_path / "voice_morph"
    plugin_path.mkdir(exist_ok=True)

    plugin = VoiceMorphPlugin(
        application=mock_app,
        plugin_path=plugin_path,
        config=config,
    )
    return plugin, mock_app


def _make_data(**kwargs) -> IntentReceivedData:
    """Erstellt IntentReceivedData mit Defaults."""
    defaults = {
        "intent": "voice_morph_status",
        "satellite_id": "sat-1",
        "room_id": "wohnzimmer",
        "is_authenticated": True,
    }
    defaults.update(kwargs)
    return IntentReceivedData(**defaults)


def _make_sine_audio(freq: float = 440.0, duration: float = 0.1,
                     sample_rate: int = 22050) -> bytes:
    """Generiert Sinus-Audio als int16 hex-encoded bytes."""
    t = np.linspace(0, duration, int(sample_rate * duration), endpoint=False)
    audio = (np.sin(2 * np.pi * freq * t) * 16000).astype(np.int16)
    return audio.tobytes().hex()


# ============================================================================
# Tests: Plugin Lifecycle
# ============================================================================

class TestLifecycle:
    """Tests fuer on_load und on_unload."""

    @pytest.mark.asyncio
    async def test_on_load_registers_handler(self, tmp_path):
        """on_load registriert tts_completed Event-Handler."""
        plugin, mock_app = _make_plugin(tmp_path)
        await plugin.on_load()

        mock_app.events.register.assert_called_once()
        call_args = mock_app.events.register.call_args
        assert call_args[0][0] == "tts_completed"

    @pytest.mark.asyncio
    async def test_on_load_sets_active_preset(self, tmp_path):
        """on_load setzt das konfigurierte Preset."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        assert plugin._active_preset is not None
        assert plugin._active_preset.preset_id == "test_preset"
        assert plugin._active_preset.label == "Test Preset"

    @pytest.mark.asyncio
    async def test_on_load_unknown_preset(self, tmp_path):
        """on_load mit unbekanntem Preset bleibt deaktiviert."""
        plugin, _ = _make_plugin(tmp_path, {"active_preset": "nonexistent"})
        await plugin.on_load()

        assert plugin._active_preset is None

    @pytest.mark.asyncio
    async def test_on_load_no_preset(self, tmp_path):
        """on_load ohne active_preset bleibt deaktiviert."""
        plugin, _ = _make_plugin(tmp_path, {"active_preset": ""})
        await plugin.on_load()

        assert plugin._active_preset is None

    @pytest.mark.asyncio
    async def test_on_load_custom_intensity(self, tmp_path):
        """on_load laedt Intensitaet aus Config."""
        plugin, _ = _make_plugin(tmp_path, {"intensity": 0.75})
        await plugin.on_load()

        assert plugin._intensity == 0.75

    @pytest.mark.asyncio
    async def test_on_unload(self, tmp_path):
        """on_unload entfernt Event-Handler."""
        plugin, mock_app = _make_plugin(tmp_path)
        await plugin.on_load()
        await plugin.on_unload()

        mock_app.events.unregister.assert_called_once()

    @pytest.mark.asyncio
    async def test_loads_custom_presets(self, tmp_path):
        """on_load laedt Presets aus config.json."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        assert plugin._morpher.get_preset("test_preset") is not None
        assert plugin._morpher.get_preset("robot") is not None


# ============================================================================
# Tests: Audio Morphing
# ============================================================================

class TestAudioMorphing:
    """Tests fuer den _on_tts_completed Handler."""

    @pytest.mark.asyncio
    async def test_morph_modifies_audio(self, tmp_path):
        """Morphing veraendert die Audio-Daten."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        original_hex = _make_sine_audio(440.0, 0.5, 22050)
        event_data = {
            "audio_data": original_hex,
            "sample_rate": 22050,
            "satellite_id": "sat-1",
        }

        await plugin._on_tts_completed("tts_completed", event_data)

        # Audio wurde veraendert
        assert event_data["audio_data"] != original_hex
        # Ist immer noch gueltiges Hex
        bytes.fromhex(event_data["audio_data"])

    @pytest.mark.asyncio
    async def test_morph_skipped_when_disabled(self, tmp_path):
        """Kein Morphing wenn kein Preset aktiv."""
        plugin, _ = _make_plugin(tmp_path, {"active_preset": ""})
        await plugin.on_load()

        original_hex = _make_sine_audio()
        event_data = {"audio_data": original_hex, "sample_rate": 22050}

        await plugin._on_tts_completed("tts_completed", event_data)

        assert event_data["audio_data"] == original_hex

    @pytest.mark.asyncio
    async def test_morph_empty_audio(self, tmp_path):
        """Leeres Audio wird ignoriert."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        event_data = {"audio_data": "", "sample_rate": 22050}
        await plugin._on_tts_completed("tts_completed", event_data)

        assert event_data["audio_data"] == ""

    @pytest.mark.asyncio
    async def test_morph_preserves_other_fields(self, tmp_path):
        """Morphing veraendert nur audio_data, nicht andere Felder."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        event_data = {
            "audio_data": _make_sine_audio(),
            "sample_rate": 22050,
            "satellite_id": "sat-1",
            "provider": "piper",
            "text": "Hallo Welt",
        }

        await plugin._on_tts_completed("tts_completed", event_data)

        assert event_data["satellite_id"] == "sat-1"
        assert event_data["provider"] == "piper"
        assert event_data["text"] == "Hallo Welt"
        assert event_data["sample_rate"] == 22050


# ============================================================================
# Tests: Intent-Handler
# ============================================================================

class TestIntents:
    """Tests fuer die 6 Voice Morph Intent-Handler."""

    @pytest.mark.asyncio
    async def test_enable(self, tmp_path):
        """voice_morph_enable aktiviert das Preset."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()
        plugin._active_preset = None  # Erst deaktivieren

        result = await plugin.handle_enable(_make_data())
        assert result.success is True
        assert plugin._active_preset is not None
        assert "Test Preset" in result.response_text

    @pytest.mark.asyncio
    async def test_enable_no_preset_configured(self, tmp_path):
        """voice_morph_enable ohne konfiguriertes Preset."""
        plugin, _ = _make_plugin(tmp_path, {"active_preset": ""})
        await plugin.on_load()

        result = await plugin.handle_enable(_make_data())
        assert result.success is False

    @pytest.mark.asyncio
    async def test_disable(self, tmp_path):
        """voice_morph_disable deaktiviert das Morphing."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        result = await plugin.handle_disable(_make_data())
        assert result.success is True
        assert plugin._active_preset is None
        assert "deaktiviert" in result.response_text.lower()

    @pytest.mark.asyncio
    async def test_status_active(self, tmp_path):
        """voice_morph_status zeigt aktives Preset."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        result = await plugin.handle_status(_make_data())
        assert result.success is True
        assert "Test Preset" in result.response_text
        assert "100" in result.response_text  # 100 Prozent

    @pytest.mark.asyncio
    async def test_status_inactive(self, tmp_path):
        """voice_morph_status zeigt deaktiviert."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()
        plugin._active_preset = None

        result = await plugin.handle_status(_make_data())
        assert result.success is True
        assert "deaktiviert" in result.response_text.lower()

    @pytest.mark.asyncio
    async def test_set_preset_by_id(self, tmp_path):
        """voice_morph_set_preset wechselt Preset per ID."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()
        plugin.save_config = MagicMock()

        data = _make_data()
        result = await plugin.handle_set_preset(data, preset_name="robot")
        assert result.success is True
        assert plugin._active_preset.preset_id == "robot"

    @pytest.mark.asyncio
    async def test_set_preset_by_label(self, tmp_path):
        """voice_morph_set_preset wechselt Preset per Label."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()
        plugin.save_config = MagicMock()

        result = await plugin.handle_set_preset(_make_data(), preset_name="Robot Stimme")
        assert result.success is True
        assert plugin._active_preset.preset_id == "robot"

    @pytest.mark.asyncio
    async def test_set_preset_not_found(self, tmp_path):
        """voice_morph_set_preset mit unbekanntem Preset."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        result = await plugin.handle_set_preset(_make_data(), preset_name="nonexistent")
        assert result.success is False

    @pytest.mark.asyncio
    async def test_set_preset_no_name(self, tmp_path):
        """voice_morph_set_preset ohne Preset-Name."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        result = await plugin.handle_set_preset(_make_data(), preset_name="")
        assert result.success is False

    @pytest.mark.asyncio
    async def test_list_presets(self, tmp_path):
        """voice_morph_list_presets listet alle Presets."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        result = await plugin.handle_list_presets(_make_data())
        assert result.success is True
        assert "Test Preset" in result.response_text
        assert "Robot Stimme" in result.response_text

    @pytest.mark.asyncio
    async def test_set_intensity_percent(self, tmp_path):
        """voice_morph_set_intensity mit Prozentwert."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()
        plugin.save_config = MagicMock()

        result = await plugin.handle_set_intensity(_make_data(), value="80 Prozent")
        assert result.success is True
        assert abs(plugin._intensity - 0.8) < 0.01
        assert "80" in result.response_text

    @pytest.mark.asyncio
    async def test_set_intensity_decimal(self, tmp_path):
        """voice_morph_set_intensity mit Dezimalwert."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()
        plugin.save_config = MagicMock()

        result = await plugin.handle_set_intensity(_make_data(), value="0.5")
        assert result.success is True
        assert abs(plugin._intensity - 0.5) < 0.01

    @pytest.mark.asyncio
    async def test_set_intensity_clamps(self, tmp_path):
        """voice_morph_set_intensity clampt auf 0-100%."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()
        plugin.save_config = MagicMock()

        result = await plugin.handle_set_intensity(_make_data(), value="150")
        assert result.success is True
        assert plugin._intensity == 1.0

    @pytest.mark.asyncio
    async def test_set_intensity_invalid(self, tmp_path):
        """voice_morph_set_intensity mit ungueltigem Wert."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        result = await plugin.handle_set_intensity(_make_data(), value="abc")
        assert result.success is False

    @pytest.mark.asyncio
    async def test_set_intensity_empty(self, tmp_path):
        """voice_morph_set_intensity ohne Wert."""
        plugin, _ = _make_plugin(tmp_path)
        await plugin.on_load()

        result = await plugin.handle_set_intensity(_make_data(), value="")
        assert result.success is False