Trixy7/plugins/music/tests/test_music_plugin.py

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

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

from trixy_core.music.track import Track, TrackMetadata, TrackState
from trixy_core.nlp.handler import IntentReceivedData, IntentResult


# =============================================================================
# YouTubeSource Tests
# =============================================================================

class TestYouTubeSource:
    """Tests fuer YouTubeSource."""

    def _make_source(self, **kwargs):
        from plugins.music.sources.youtube import YouTubeSource
        return YouTubeSource(
            cache_dir=kwargs.get("cache_dir", "/tmp/trixy_test_cache"),
            max_cache_size_mb=kwargs.get("max_cache_size_mb", 100),
            **{k: v for k, v in kwargs.items()
               if k not in ("cache_dir", "max_cache_size_mb")},
        )

    @pytest.mark.asyncio
    async def test_connect_without_yt_dlp(self):
        """connect() gibt False zurueck wenn yt-dlp nicht installiert."""
        with patch("plugins.music.sources.youtube._YT_DLP_AVAILABLE", False):
            source = self._make_source()
            result = await source.connect()
            assert result is False
            assert source.is_available is False

    @pytest.mark.asyncio
    async def test_connect_with_yt_dlp(self, tmp_path):
        """connect() gibt True zurueck wenn yt-dlp verfuegbar."""
        with patch("plugins.music.sources.youtube._YT_DLP_AVAILABLE", True):
            source = self._make_source(cache_dir=str(tmp_path / "cache"))
            result = await source.connect()
            assert result is True
            assert source.is_available is True
            assert (tmp_path / "cache").exists()

    @pytest.mark.asyncio
    async def test_search_without_yt_dlp(self):
        """search() gibt leere Liste zurueck ohne yt-dlp."""
        with patch("plugins.music.sources.youtube._YT_DLP_AVAILABLE", False):
            source = self._make_source()
            results = await source.search("test query")
            assert results == []

    @pytest.mark.asyncio
    async def test_prepare_track_cache_hit(self, tmp_path):
        """prepare_track() nutzt Cache wenn vorhanden."""
        cache_dir = tmp_path / "cache"
        cache_dir.mkdir()
        cached_file = cache_dir / "test123.mp3"
        cached_file.write_bytes(b"fake audio data")

        with patch("plugins.music.sources.youtube._YT_DLP_AVAILABLE", True):
            source = self._make_source(cache_dir=str(cache_dir))
            await source.connect()

            track = Track(
                source_type="youtube",
                source_id="test123",
                extra={"video_id": "test123"},
            )

            result = await source.prepare_track(track)
            assert result is True
            assert track.path == cached_file
            assert track.state == TrackState.AVAILABLE

    def test_parse_title_with_dash(self):
        """Parst 'Artist - Title' Format."""
        source = self._make_source()
        artist, title = source._parse_title("Queen - Bohemian Rhapsody", "QueenVEVO")
        assert artist == "Queen"
        assert title == "Bohemian Rhapsody"

    def test_parse_title_without_dash(self):
        """Verwendet Uploader als Artist wenn kein Trennzeichen."""
        source = self._make_source()
        artist, title = source._parse_title("Bohemian Rhapsody", "QueenVEVO")
        assert artist == "QueenVEVO"

    def test_parse_title_with_pipe(self):
        """Parst 'Artist | Title' Format."""
        source = self._make_source()
        artist, title = source._parse_title("Queen | Bohemian Rhapsody", "QueenVEVO")
        assert artist == "Queen"
        assert title == "Bohemian Rhapsody"

    @pytest.mark.asyncio
    async def test_cleanup_cache(self, tmp_path):
        """cleanup_cache() entfernt aelteste Dateien."""
        cache_dir = tmp_path / "cache"
        cache_dir.mkdir()

        # 3 Dateien erstellen (je 1 MB simuliert)
        for i in range(3):
            f = cache_dir / f"video{i}.mp3"
            f.write_bytes(b"x" * (1024 * 1024))  # 1 MB

        with patch("plugins.music.sources.youtube._YT_DLP_AVAILABLE", True):
            source = self._make_source(
                cache_dir=str(cache_dir),
                max_cache_size_mb=2,  # Max 2 MB → 1 Datei muss weg
            )
            await source.connect()
            removed = await source.cleanup_cache()
            assert removed >= 1

    @pytest.mark.asyncio
    async def test_scan_returns_zero(self):
        """scan() gibt immer 0 zurueck (YouTube unterstuetzt kein Scanning)."""
        source = self._make_source()
        result = await source.scan()
        assert result == 0


# =============================================================================
# MusicPlugin Tests
# =============================================================================

class TestMusicPlugin:
    """Tests fuer MusicPlugin."""

    def _make_plugin(self):
        """Erstellt ein MusicPlugin mit Mock-Application."""
        mock_app = MagicMock()
        mock_app.events = MagicMock()
        mock_app.events.emit = AsyncMock()
        mock_app.events.trigger = AsyncMock()

        # Mock ServiceContainer
        mock_service = MagicMock()
        mock_service.player = MagicMock()
        mock_service.is_playing = False
        mock_service.is_paused = False
        mock_service.current_track = None
        mock_service.play_track = AsyncMock(return_value=True)
        mock_service.play_playlist = AsyncMock(return_value=True)
        mock_app.services.get_service = MagicMock(return_value=mock_service)

        # Mock SatelliteManager
        mock_sat = MagicMock()
        mock_sat.id = "sat_1"
        mock_sat.is_connected = True
        mock_sat.sockets.music_out = MagicMock()
        mock_app.satellites = MagicMock()
        mock_app.satellites.get = MagicMock(return_value=mock_sat)
        mock_app.satellites.all = MagicMock(return_value=[mock_sat])

        # Mock ServerConfig
        mock_app.server_config.assets_directory = "/tmp/assets"
        mock_app.server_config.profile = "default"

        plugin_path = Path("/tmp/test_music_plugin")
        config = {
            "enabled": True,
            "youtube": {"enabled": False},
            "local": {"enabled": False, "music_dirs": [], "scan_on_load": False},
        }

        from plugins.music.main import MusicPlugin
        plugin = MusicPlugin(mock_app, plugin_path, config)

        return plugin, mock_app, mock_service

    @pytest.mark.asyncio
    async def test_handle_play_music_resume(self):
        """play_music ohne Query bei pausierter Musik → Resume."""
        plugin, mock_app, mock_service = self._make_plugin()
        mock_service.is_paused = True

        data = IntentReceivedData(
            intent="play_music",
            slots={},
            satellite_id="sat_1",
        )

        result = await plugin.handle_play_music(data)
        assert result.success is True
        assert "fortgesetzt" in result.response_text.lower()
        mock_app.events.emit.assert_called_once()

    @pytest.mark.asyncio
    async def test_handle_what_is_playing_no_track(self):
        """what_is_playing ohne aktiven Track."""
        plugin, _, mock_service = self._make_plugin()
        mock_service.current_track = None

        data = IntentReceivedData(
            intent="what_is_playing",
            satellite_id="sat_1",
        )

        result = await plugin.handle_what_is_playing(data)
        assert result.success is True
        assert "keine musik" in result.response_text.lower()

    @pytest.mark.asyncio
    async def test_handle_what_is_playing_with_track(self):
        """what_is_playing mit aktivem Track."""
        plugin, _, mock_service = self._make_plugin()
        mock_service.current_track = Track(
            source_type="local",
            metadata=TrackMetadata(
                title="Bohemian Rhapsody",
                artist="Queen",
                duration_ms=354000,
            ),
        )

        data = IntentReceivedData(
            intent="what_is_playing",
            satellite_id="sat_1",
        )

        result = await plugin.handle_what_is_playing(data)
        assert result.success is True
        assert "Bohemian Rhapsody" in result.response_text
        assert "Queen" in result.response_text

    @pytest.mark.asyncio
    async def test_handle_next_track(self):
        """next_track emittiert music_next Event."""
        plugin, mock_app, _ = self._make_plugin()

        data = IntentReceivedData(
            intent="next_track",
            satellite_id="sat_1",
        )

        result = await plugin.handle_next_track(data)
        assert result.success is True
        mock_app.events.emit.assert_called_once_with(
            "music_next", {"satellite_id": "sat_1"}
        )

    @pytest.mark.asyncio
    async def test_handle_previous_track(self):
        """previous_track emittiert music_previous Event."""
        plugin, mock_app, _ = self._make_plugin()

        data = IntentReceivedData(
            intent="previous_track",
            satellite_id="sat_1",
        )

        result = await plugin.handle_previous_track(data)
        assert result.success is True
        mock_app.events.emit.assert_called_once_with(
            "music_previous", {"satellite_id": "sat_1"}
        )

    @pytest.mark.asyncio
    async def test_handle_play_no_service(self):
        """play_music gibt Fehler zurueck wenn Service fehlt."""
        plugin, mock_app, _ = self._make_plugin()
        mock_app.services.get_service = MagicMock(return_value=None)

        data = IntentReceivedData(
            intent="play_music",
            slots={"query": "test"},
            satellite_id="sat_1",
        )

        result = await plugin.handle_play_music(data)
        assert result.success is False

    @pytest.mark.asyncio
    async def test_get_target_satellites(self):
        """Ermittelt korrekte Ziel-Satellites."""
        plugin, _, _ = self._make_plugin()

        targets = plugin._get_target_satellites("sat_1")
        assert targets == ["sat_1"]

    @pytest.mark.asyncio
    async def test_graceful_degradation_no_youtube(self):
        """Plugin funktioniert ohne yt-dlp."""
        plugin, _, _ = self._make_plugin()

        # YouTube deaktiviert in Config
        await plugin.on_load()

        assert plugin._youtube is None
        # Plugin sollte trotzdem geladen sein

        await plugin.on_unload()