# -*- coding: utf-8 -*- """ Tests für das Musik-System. Testet: - Track und TrackMetadata Datenklassen - LocalFileSource (Dateien scannen) - AudioDecoder (MP3, Format-Erkennung) - AudioStreamer und StreamSession - PlayQueue und Playlist """ import asyncio import pytest import shutil from pathlib import Path from datetime import datetime, timedelta from unittest.mock import AsyncMock, MagicMock from trixy_core.music.track import Track, TrackMetadata, TrackState from trixy_core.music.sources.local import LocalFileSource, SUPPORTED_EXTENSIONS from trixy_core.music.sources.base import SourceState, SourceType from trixy_core.music.formats.base import AudioFormat, AudioInfo, DecodedAudio, DecoderError from trixy_core.music.formats.mp3 import MP3Decoder from trixy_core.music.formats.registry import DecoderRegistry from trixy_core.music.stream.config import StreamConfig from trixy_core.music.stream.session import StreamSession, StreamState, StreamStats from trixy_core.music.stream.streamer import AudioStreamer from trixy_core.music.queue import PlayQueue, QueueItem, QueueState from trixy_core.music.playlist.playlist import Playlist, RepeatMode, ShuffleMode # ============================================================================= # Test Data # ============================================================================= # Pfad zu echten Musikdateien MUSIC_DIR = Path(__file__).parent.parent / "assets" / "default" / "music" def has_music_files() -> bool: """Prüft ob Musikdateien vorhanden sind.""" return MUSIC_DIR.exists() and any(MUSIC_DIR.glob("*.mp3")) def get_first_mp3() -> Path | None: """Holt die erste MP3-Datei.""" if MUSIC_DIR.exists(): mp3_files = list(MUSIC_DIR.glob("*.mp3")) if mp3_files: return mp3_files[0] return None def has_pydub() -> bool: """Prüft ob pydub und ffmpeg/ffprobe verfügbar sind.""" try: from pydub import AudioSegment import subprocess # Prüfe auch ob ffprobe verfügbar ist (benötigt für MP3-Dekodierung) result = subprocess.run( ["ffprobe", "-version"], capture_output=True, timeout=5, ) return result.returncode == 0 except (ImportError, FileNotFoundError, subprocess.TimeoutExpired): return False def has_mutagen() -> bool: """Prüft ob mutagen verfügbar ist.""" try: import mutagen return True except ImportError: return False def has_ffmpeg() -> bool: """Prüft ob ffmpeg verfügbar ist.""" return shutil.which("ffmpeg") is not None # ============================================================================= # Tests für TrackMetadata # ============================================================================= class TestTrackMetadata: """Tests für TrackMetadata Datenklasse.""" def test_default_values(self): """Testet Standard-Werte.""" metadata = TrackMetadata() assert metadata.title == "" assert metadata.artist == "" assert metadata.album == "" assert metadata.sample_rate == 44100 assert metadata.channels == 2 assert metadata.duration_ms == 0 def test_duration_property(self): """Testet duration Property.""" metadata = TrackMetadata(duration_ms=180000) # 3 Minuten assert metadata.duration == timedelta(milliseconds=180000) assert metadata.duration.total_seconds() == 180.0 def test_duration_formatted(self): """Testet formatierte Dauer.""" metadata = TrackMetadata(duration_ms=185000) # 3:05 assert metadata.duration_formatted == "3:05" def test_duration_formatted_short(self): """Testet formatierte Dauer unter einer Minute.""" metadata = TrackMetadata(duration_ms=45000) # 45 Sekunden assert metadata.duration_formatted == "0:45" def test_display_artist(self): """Testet display_artist.""" # Ohne Artist metadata = TrackMetadata() assert metadata.display_artist == "Unbekannter Künstler" # Mit Artist metadata = TrackMetadata(artist="Test Artist") assert metadata.display_artist == "Test Artist" # Album-Artist überschreibt metadata = TrackMetadata(artist="Track Artist", album_artist="Album Artist") assert metadata.display_artist == "Album Artist" def test_display_title(self): """Testet display_title.""" metadata = TrackMetadata() assert metadata.display_title == "Unbekannter Titel" metadata = TrackMetadata(title="Test Song") assert metadata.display_title == "Test Song" def test_to_dict(self): """Testet to_dict Konvertierung.""" metadata = TrackMetadata( title="Test", artist="Artist", duration_ms=120000, ) result = metadata.to_dict() assert result["title"] == "Test" assert result["artist"] == "Artist" assert result["duration_ms"] == 120000 assert result["duration_formatted"] == "2:00" assert "artwork_data" not in result # Binärdaten ausgeschlossen # ============================================================================= # Tests für Track # ============================================================================= class TestTrack: """Tests für Track Datenklasse.""" def test_auto_id(self): """Testet automatische ID-Generierung.""" track = Track() assert track.id.startswith("track_") assert len(track.id) == 18 # "track_" + 12 hex chars def test_uri_from_path(self): """Testet URI-Generierung aus Pfad.""" track = Track(path=Path("/music/test.mp3")) assert track.uri.startswith("file://") assert "test.mp3" in track.uri def test_uri_from_source(self): """Testet URI-Generierung aus Source.""" track = Track(source_type="spotify", source_id="abc123") assert track.uri == "spotify:abc123" def test_state_transitions(self): """Testet Zustandsübergänge.""" track = Track() assert track.state == TrackState.UNKNOWN track.state = TrackState.AVAILABLE assert track.is_available is True assert track.is_playing is False track.state = TrackState.PLAYING assert track.is_playing is True def test_is_local(self): """Testet is_local Property.""" # Kein lokaler Track track = Track(source_type="spotify") assert track.is_local is False # Lokaler Track ohne Pfad track = Track(source_type="local") assert track.is_local is False # Lokaler Track mit Pfad track = Track(source_type="local", path=Path("/music/test.mp3")) assert track.is_local is True def test_mark_played(self): """Testet mark_played.""" track = Track() assert track.play_count == 0 assert track.last_played is None track.mark_played() assert track.play_count == 1 assert track.last_played is not None assert isinstance(track.last_played, datetime) def test_set_error(self): """Testet Fehler-Zustand.""" track = Track() track.set_error("Datei nicht gefunden") assert track.state == TrackState.ERROR assert track.error_message == "Datei nicht gefunden" def test_shortcut_properties(self): """Testet Kurzschluss-Properties.""" track = Track( metadata=TrackMetadata( title="Test Song", artist="Test Artist", duration_ms=180000, ) ) assert track.title == "Test Song" assert track.artist == "Test Artist" assert track.duration_ms == 180000 def test_to_dict(self): """Testet to_dict Konvertierung.""" track = Track( source_type="local", path=Path("/music/test.mp3"), metadata=TrackMetadata(title="Test"), ) result = track.to_dict() assert result["id"] == track.id assert result["source_type"] == "local" assert result["path"] == "/music/test.mp3" assert result["metadata"]["title"] == "Test" # ============================================================================= # Tests für AudioFormat # ============================================================================= class TestAudioFormat: """Tests für AudioFormat Enum.""" def test_from_extension(self): """Testet Format-Erkennung aus Dateiendung.""" assert AudioFormat.from_extension("mp3") == AudioFormat.MP3 assert AudioFormat.from_extension(".MP3") == AudioFormat.MP3 assert AudioFormat.from_extension("wav") == AudioFormat.WAV assert AudioFormat.from_extension("ogg") == AudioFormat.OGG assert AudioFormat.from_extension("flac") == AudioFormat.FLAC assert AudioFormat.from_extension("unknown") == AudioFormat.UNKNOWN def test_from_mime(self): """Testet Format-Erkennung aus MIME-Type.""" assert AudioFormat.from_mime("audio/mpeg") == AudioFormat.MP3 assert AudioFormat.from_mime("audio/mp3") == AudioFormat.MP3 assert AudioFormat.from_mime("audio/wav") == AudioFormat.WAV assert AudioFormat.from_mime("audio/ogg") == AudioFormat.OGG assert AudioFormat.from_mime("audio/flac") == AudioFormat.FLAC assert AudioFormat.from_mime("text/html") == AudioFormat.UNKNOWN # ============================================================================= # Tests für AudioInfo # ============================================================================= class TestAudioInfo: """Tests für AudioInfo Datenklasse.""" def test_defaults(self): """Testet Standard-Werte.""" info = AudioInfo(format=AudioFormat.MP3) assert info.format == AudioFormat.MP3 assert info.sample_rate == 44100 assert info.channels == 2 assert info.bit_depth == 16 def test_duration_seconds(self): """Testet duration_seconds Property.""" info = AudioInfo(format=AudioFormat.MP3, duration_ms=180000) assert info.duration_seconds == 180.0 def test_is_stereo(self): """Testet is_stereo Property.""" mono = AudioInfo(format=AudioFormat.MP3, channels=1) stereo = AudioInfo(format=AudioFormat.MP3, channels=2) assert mono.is_stereo is False assert stereo.is_stereo is True # ============================================================================= # Tests für DecodedAudio # ============================================================================= class TestDecodedAudio: """Tests für DecodedAudio Datenklasse.""" def test_duration_calculation(self): """Testet Dauer-Berechnung.""" # 1 Sekunde @ 44100 Hz, Stereo, 16-bit = 44100 * 2 * 2 = 176400 bytes pcm_data = bytes(176400) audio = DecodedAudio( pcm_data=pcm_data, sample_rate=44100, channels=2, bit_depth=16, ) assert audio.duration_ms == 1000 assert audio.sample_count == 44100 def test_sample_count(self): """Testet Sample-Zählung.""" # 100 Samples @ Stereo, 16-bit = 100 * 2 * 2 = 400 bytes pcm_data = bytes(400) audio = DecodedAudio( pcm_data=pcm_data, sample_rate=44100, channels=2, bit_depth=16, ) assert audio.sample_count == 100 # ============================================================================= # Tests für LocalFileSource # ============================================================================= class TestLocalFileSource: """Tests für LocalFileSource.""" def test_initialization(self): """Testet Initialisierung.""" source = LocalFileSource(name="Test Music") assert source.name == "Test Music" assert source.source_type == SourceType.LOCAL assert source.paths == [] assert source.recursive is True def test_add_path(self, tmp_path): """Testet Hinzufügen von Pfaden.""" source = LocalFileSource() # Existierendes Verzeichnis assert source.add_path(tmp_path) is True assert tmp_path in source.paths # Nicht existierendes Verzeichnis fake_path = tmp_path / "nonexistent" assert source.add_path(fake_path) is False def test_remove_path(self, tmp_path): """Testet Entfernen von Pfaden.""" source = LocalFileSource(paths=[tmp_path]) assert source.remove_path(tmp_path) is True assert tmp_path not in source.paths # Nicht vorhandener Pfad assert source.remove_path(tmp_path) is False @pytest.mark.asyncio async def test_connect(self, tmp_path): """Testet connect.""" source = LocalFileSource(paths=[tmp_path]) result = await source.connect() assert result is True assert source.state == SourceState.AVAILABLE @pytest.mark.asyncio async def test_connect_no_paths(self): """Testet connect ohne Pfade.""" source = LocalFileSource() result = await source.connect() assert result is False assert source.state == SourceState.UNAVAILABLE @pytest.mark.asyncio async def test_scan_empty_directory(self, tmp_path): """Testet Scan eines leeren Verzeichnisses.""" source = LocalFileSource(paths=[tmp_path]) count = await source.scan() assert count == 0 assert source.track_count == 0 @pytest.mark.asyncio async def test_scan_with_files(self, tmp_path): """Testet Scan mit Dummy-Dateien.""" # Dummy MP3-Dateien erstellen (tmp_path / "song1.mp3").write_bytes(b"fake mp3 data") (tmp_path / "song2.mp3").write_bytes(b"fake mp3 data") (tmp_path / "not_music.txt").write_bytes(b"text file") source = LocalFileSource(paths=[tmp_path]) count = await source.scan() # Sollte nur MP3-Dateien finden assert count == 2 assert source.track_count == 2 @pytest.mark.asyncio async def test_scan_recursive(self, tmp_path): """Testet rekursiven Scan.""" # Unterverzeichnis mit Dateien subdir = tmp_path / "subdir" subdir.mkdir() (subdir / "song.mp3").write_bytes(b"fake mp3") (tmp_path / "root.mp3").write_bytes(b"fake mp3") source = LocalFileSource(paths=[tmp_path], recursive=True) count = await source.scan() assert count == 2 # Nicht-rekursiv source2 = LocalFileSource(paths=[tmp_path], recursive=False) count2 = await source2.scan() assert count2 == 1 @pytest.mark.asyncio @pytest.mark.skipif(not has_music_files(), reason="Keine Musikdateien vorhanden") async def test_scan_real_music_files(self): """Testet Scan mit echten Musikdateien.""" source = LocalFileSource(paths=[MUSIC_DIR]) count = await source.scan() assert count > 0 assert source.track_count > 0 # Prüfe ersten Track tracks = list(source.tracks) assert len(tracks) > 0 track = tracks[0] assert track.path is not None assert track.path.exists() assert track.state == TrackState.AVAILABLE def test_supported_extensions(self): """Testet unterstützte Dateiendungen.""" assert ".mp3" in SUPPORTED_EXTENSIONS assert ".wav" in SUPPORTED_EXTENSIONS assert ".ogg" in SUPPORTED_EXTENSIONS assert ".flac" in SUPPORTED_EXTENSIONS # ============================================================================= # Tests für MP3Decoder # ============================================================================= class TestMP3Decoder: """Tests für MP3Decoder.""" def test_format(self): """Testet Format-Konstanten.""" decoder = MP3Decoder() assert decoder.FORMAT == AudioFormat.MP3 assert "mp3" in decoder.EXTENSIONS assert "audio/mpeg" in decoder.MIME_TYPES def test_supports_extension(self): """Testet Extension-Support.""" decoder = MP3Decoder() assert decoder.supports_extension("mp3") is True assert decoder.supports_extension(".mp3") is True assert decoder.supports_extension("wav") is False def test_supports_mime(self): """Testet MIME-Type-Support.""" decoder = MP3Decoder() assert decoder.supports_mime("audio/mpeg") is True assert decoder.supports_mime("audio/mp3") is True assert decoder.supports_mime("audio/wav") is False @pytest.mark.asyncio @pytest.mark.skipif(not has_music_files(), reason="Keine Musikdateien vorhanden") async def test_get_info_real_file(self): """Testet Info-Extraktion aus echter Datei.""" mp3_file = get_first_mp3() if mp3_file is None: pytest.skip("Keine MP3-Datei gefunden") decoder = MP3Decoder() # Sollte auch ohne Mutagen funktionieren (Header-Parsing) info = await decoder.get_info(mp3_file) assert info.format == AudioFormat.MP3 assert info.sample_rate > 0 assert info.channels in (1, 2) assert info.file_size > 0 @pytest.mark.asyncio @pytest.mark.skipif(not (has_pydub() or has_ffmpeg()), reason="Kein Decoder verfügbar") @pytest.mark.skipif(not has_music_files(), reason="Keine Musikdateien vorhanden") async def test_decode_real_file(self): """Testet Dekodierung einer echten Datei.""" mp3_file = get_first_mp3() if mp3_file is None: pytest.skip("Keine MP3-Datei gefunden") decoder = MP3Decoder() if not decoder.is_available: pytest.skip("Decoder nicht verfügbar") decoded = await decoder.decode(mp3_file) assert len(decoded.pcm_data) > 0 assert decoded.sample_rate == 44100 # Standard Output assert decoded.channels == 2 assert decoded.bit_depth == 16 @pytest.mark.asyncio async def test_decode_without_backend(self, tmp_path): """Testet Fehler ohne Backend.""" # Fake MP3-Datei fake_mp3 = tmp_path / "fake.mp3" fake_mp3.write_bytes(b"not a real mp3") decoder = MP3Decoder() if decoder.is_available: pytest.skip("Decoder ist verfügbar") with pytest.raises(DecoderError, match="Kein MP3-Decoder verfügbar"): await decoder.decode(fake_mp3) # ============================================================================= # Tests für DecoderRegistry # ============================================================================= class TestDecoderRegistry: """Tests für DecoderRegistry.""" def test_get_decoder_for_extension(self): """Testet Decoder-Lookup nach Extension.""" registry = DecoderRegistry() registry.register_default_decoders() # WAV sollte immer verfügbar sein decoder = registry.get_decoder_for_extension("wav") assert decoder is not None assert decoder.FORMAT == AudioFormat.WAV # Unbekannte Extension decoder = registry.get_decoder_for_extension("unknown") assert decoder is None def test_get_decoder_for_format(self): """Testet Decoder-Lookup nach Format.""" registry = DecoderRegistry() registry.register_default_decoders() # WAV sollte immer verfügbar sein decoder = registry.get_decoder(AudioFormat.WAV) assert decoder is not None def test_mp3_decoder_if_available(self): """Testet MP3-Decoder wenn verfügbar.""" registry = DecoderRegistry() registry.register_default_decoders() decoder = registry.get_decoder_for_extension("mp3") if has_pydub() or has_ffmpeg(): assert decoder is not None assert isinstance(decoder, MP3Decoder) else: # MP3-Decoder nicht verfügbar ohne pydub/ffmpeg # Das ist erwartetes Verhalten pass def test_extensions(self): """Testet Liste unterstützter Extensions.""" registry = DecoderRegistry() registry.register_default_decoders() extensions = registry.extensions # WAV sollte immer verfügbar sein assert "wav" in extensions or "wave" in extensions # ============================================================================= # Tests für StreamConfig # ============================================================================= class TestStreamConfig: """Tests für StreamConfig.""" def test_for_music(self): """Testet Musik-Konfiguration.""" config = StreamConfig.for_music() assert config.sample_rate == 48000 assert config.channels == 2 assert config.bit_depth == 16 def test_for_voice(self): """Testet Voice-Konfiguration.""" config = StreamConfig.for_voice() assert config.sample_rate == 16000 assert config.channels == 1 assert config.bit_depth == 16 def test_bytes_per_second(self): """Testet bytes_per_second Berechnung.""" config = StreamConfig( sample_rate=44100, channels=2, bit_depth=16, ) # 44100 * 2 * 2 = 176400 assert config.bytes_per_second == 176400 # ============================================================================= # Tests für StreamStats # ============================================================================= class TestStreamStats: """Tests für StreamStats.""" def test_initial_values(self): """Testet Initialwerte.""" stats = StreamStats() assert stats.bytes_sent == 0 assert stats.packets_sent == 0 assert stats.packets_lost == 0 assert stats.loss_rate == 0.0 def test_loss_rate(self): """Testet Loss-Rate-Berechnung.""" stats = StreamStats(packets_sent=90, packets_lost=10) assert stats.loss_rate == 0.1 # 10% def test_add_latency_sample(self): """Testet Latenz-Tracking.""" stats = StreamStats() stats.add_latency_sample(10.0) stats.add_latency_sample(20.0) stats.add_latency_sample(30.0) assert stats.min_latency_ms == 10.0 assert stats.max_latency_ms == 30.0 assert stats.avg_latency_ms == 20.0 def test_to_dict(self): """Testet to_dict Konvertierung.""" stats = StreamStats(bytes_sent=1000, packets_sent=10) result = stats.to_dict() assert result["bytes_sent"] == 1000 assert result["packets_sent"] == 10 assert "latency" in result # ============================================================================= # Tests für StreamSession # ============================================================================= class TestStreamSession: """Tests für StreamSession.""" def test_initialization(self): """Testet Initialisierung.""" session = StreamSession(satellite_id="sat_001") assert session.satellite_id == "sat_001" assert session.state == StreamState.IDLE assert session.is_streaming is False assert session.position_ms == 0 @pytest.mark.asyncio async def test_start_and_stop(self): """Testet Start und Stop.""" session = StreamSession(satellite_id="sat_001") track = Track(metadata=TrackMetadata(title="Test", duration_ms=180000)) success = await session.start(track) assert success is True assert session.current_track == track assert session.state in (StreamState.BUFFERING, StreamState.CONNECTING) await session.stop() assert session.state == StreamState.IDLE assert session.current_track is None @pytest.mark.asyncio async def test_pause_resume(self): """Testet Pause und Resume.""" send_callback = AsyncMock(return_value=True) session = StreamSession(satellite_id="sat_001", send_callback=send_callback) track = Track(metadata=TrackMetadata(title="Test", duration_ms=180000)) await session.start(track) # Füttere genug Daten um zu STREAMING zu wechseln data = bytes(session.config.prebuffer_bytes) await session.feed(data) # Warte kurz auf State-Update await asyncio.sleep(0.1) if session.state == StreamState.STREAMING: await session.pause() assert session.state == StreamState.PAUSED await session.resume() assert session.state == StreamState.STREAMING await session.stop() @pytest.mark.asyncio async def test_feed_data(self): """Testet Daten-Fütterung.""" session = StreamSession(satellite_id="sat_001") track = Track(metadata=TrackMetadata(duration_ms=10000)) await session.start(track) # Kleine Daten einfügen data = bytes(1000) result = await session.feed(data) assert result is True assert session.buffer_ms > 0 await session.stop() @pytest.mark.asyncio async def test_buffer_overflow(self): """Testet Puffer-Überlauf.""" config = StreamConfig(max_buffer_ms=100) # Sehr kleiner Puffer session = StreamSession(satellite_id="sat_001", config=config) track = Track(metadata=TrackMetadata(duration_ms=10000)) await session.start(track) # Versuche mehr Daten einzufügen als der Puffer fasst large_data = bytes(config.max_buffer_ms * config.bytes_per_second) result1 = await session.feed(large_data) result2 = await session.feed(large_data) # Zweites Einfügen sollte abgelehnt werden assert result2 is False assert session.stats.buffer_overruns > 0 await session.stop() @pytest.mark.asyncio async def test_seek(self): """Testet Seek-Funktion.""" session = StreamSession(satellite_id="sat_001") track = Track(metadata=TrackMetadata(duration_ms=180000)) await session.start(track) result = await session.seek(90000) assert result is True assert session.position_ms == 90000 assert session.state == StreamState.BUFFERING # Puffer geleert await session.stop() def test_get_status(self): """Testet Status-Abfrage.""" session = StreamSession(satellite_id="sat_001") status = session.get_status() assert status["satellite_id"] == "sat_001" assert status["state"] == "idle" assert "position_ms" in status assert "buffer_level" in status # ============================================================================= # Tests für AudioStreamer # ============================================================================= class TestAudioStreamer: """Tests für AudioStreamer.""" def test_initialization(self): """Testet Initialisierung.""" streamer = AudioStreamer() assert streamer.session_count == 0 assert streamer.is_streaming is False assert streamer.current_track is None def test_add_satellite(self): """Testet Hinzufügen von Satellites.""" streamer = AudioStreamer() callback = AsyncMock(return_value=True) session = streamer.add_satellite("sat_001", callback) assert session is not None assert "sat_001" in streamer assert streamer.session_count == 1 @pytest.mark.asyncio async def test_remove_satellite(self): """Testet Entfernen von Satellites.""" streamer = AudioStreamer() callback = AsyncMock(return_value=True) streamer.add_satellite("sat_001", callback) result = await streamer.remove_satellite("sat_001") assert result is True assert "sat_001" not in streamer assert streamer.session_count == 0 @pytest.mark.asyncio async def test_start_streaming(self): """Testet Streaming-Start.""" streamer = AudioStreamer() callback = AsyncMock(return_value=True) streamer.add_satellite("sat_001", callback) track = Track(metadata=TrackMetadata(title="Test", duration_ms=180000)) results = await streamer.start(track) assert results["sat_001"] is True assert streamer.current_track == track await streamer.stop() @pytest.mark.asyncio async def test_feed_multiple_satellites(self): """Testet Daten-Fütterung an mehrere Satellites.""" streamer = AudioStreamer() callbacks = [AsyncMock(return_value=True) for _ in range(3)] for i, cb in enumerate(callbacks): streamer.add_satellite(f"sat_{i}", cb) track = Track(metadata=TrackMetadata(duration_ms=10000)) await streamer.start(track) data = bytes(1000) results = await streamer.feed(data) assert len(results) == 3 assert all(results.values()) await streamer.stop() @pytest.mark.asyncio async def test_pause_resume_all(self): """Testet Pause/Resume für alle Sessions.""" streamer = AudioStreamer() callback = AsyncMock(return_value=True) streamer.add_satellite("sat_001", callback) track = Track(metadata=TrackMetadata(duration_ms=180000)) await streamer.start(track) await streamer.pause() session = streamer.get_session("sat_001") # Session wechselt zu PAUSED nur wenn sie STREAMING war await streamer.resume() # Session sollte wieder aktiv sein await streamer.stop() def test_get_status(self): """Testet Status-Abfrage.""" streamer = AudioStreamer() callback = AsyncMock(return_value=True) streamer.add_satellite("sat_001", callback) status = streamer.get_status() assert status["session_count"] == 1 assert "sat_001" in status["sessions"] # ============================================================================= # Tests für PlayQueue # ============================================================================= class TestPlayQueue: """Tests für PlayQueue.""" def test_initialization(self): """Testet Initialisierung.""" queue = PlayQueue() assert queue.count == 0 assert queue.is_empty is True assert queue.has_next is False assert queue.state == QueueState.IDLE def test_add_track(self): """Testet Hinzufügen von Tracks.""" queue = PlayQueue() track = Track(metadata=TrackMetadata(title="Test")) item = queue.add(track) assert item is not None assert queue.count == 1 assert queue.has_next is True def test_add_play_next(self): """Testet Play-Next-Funktion.""" queue = PlayQueue() track1 = Track(id="t1", metadata=TrackMetadata(title="First")) track2 = Track(id="t2", metadata=TrackMetadata(title="Second")) track3 = Track(id="t3", metadata=TrackMetadata(title="Play Next")) queue.add(track1) queue.add(track2) queue.add(track3, play_next=True) # Play-Next sollte am Anfang sein assert queue.items[0].track.id == "t3" assert queue.items[0].priority is True def test_remove_by_index(self): """Testet Entfernen nach Index.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.add(Track(id="t2")) item = queue.remove(0) assert item is not None assert item.track.id == "t1" assert queue.count == 1 def test_remove_by_id(self): """Testet Entfernen nach ID.""" queue = PlayQueue() item = queue.add(Track(id="t1")) removed = queue.remove_by_id(item.id) assert removed is not None assert queue.count == 0 def test_next(self): """Testet next() Navigation.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.add(Track(id="t2")) item = queue.next() assert item is not None assert item.track.id == "t1" assert queue.current_item == item assert queue.has_previous is False # Noch keine History # Nächster Track item2 = queue.next() assert item2.track.id == "t2" assert queue.has_previous is True # Jetzt gibt es History def test_previous(self): """Testet previous() Navigation.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.add(Track(id="t2")) queue.next() # t1 queue.next() # t2 prev = queue.previous() assert prev is not None assert prev.track.id == "t1" assert queue.current_item == prev def test_clear(self): """Testet Queue leeren.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.add(Track(id="t2")) queue.next() # Setzt current count = queue.clear() assert count == 1 # Nur der verbleibende Track assert queue.is_empty is True assert queue.current_item is not None # Current bleibt def test_clear_all(self): """Testet komplettes Leeren.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.next() count = queue.clear_all() assert queue.current_item is None assert queue.state == QueueState.IDLE def test_total_duration(self): """Testet Gesamt-Dauer-Berechnung.""" queue = PlayQueue() queue.add(Track(metadata=TrackMetadata(duration_ms=60000))) queue.add(Track(metadata=TrackMetadata(duration_ms=120000))) assert queue.total_duration_ms == 180000 def test_move(self): """Testet Verschieben von Items.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.add(Track(id="t2")) queue.add(Track(id="t3")) result = queue.move(2, 0) assert result is True assert queue.items[0].track.id == "t3" def test_peek(self): """Testet Vorschau ohne Entfernen.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.add(Track(id="t2")) items = queue.peek(2) assert len(items) == 2 assert queue.count == 2 # Nichts entfernt def test_peek_next(self): """Testet peek_next.""" queue = PlayQueue() assert queue.peek_next() is None queue.add(Track(id="t1")) item = queue.peek_next() assert item is not None assert item.track.id == "t1" def test_find_track(self): """Testet Track-Suche.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.add(Track(id="t2")) result = queue.find_track("t2") assert result is not None index, item = result assert index == 1 assert item.track.id == "t2" def test_contains_track(self): """Testet contains_track.""" queue = PlayQueue() queue.add(Track(id="t1")) assert queue.contains_track("t1") is True assert queue.contains_track("t2") is False def test_state_management(self): """Testet Zustands-Management.""" queue = PlayQueue() queue.add(Track(id="t1")) queue.next() queue.set_playing() assert queue.state == QueueState.PLAYING queue.set_paused() assert queue.state == QueueState.PAUSED queue.set_loading() assert queue.state == QueueState.LOADING def test_to_dict(self): """Testet to_dict Konvertierung.""" queue = PlayQueue() queue.add(Track(id="t1", metadata=TrackMetadata(title="Test"))) result = queue.to_dict() assert result["count"] == 1 assert len(result["items"]) == 1 assert result["items"][0]["track"]["id"] == "t1" # ============================================================================= # Tests für Playlist # ============================================================================= class TestPlaylist: """Tests für Playlist.""" def test_initialization(self): """Testet Initialisierung.""" playlist = Playlist(name="Test Playlist") assert playlist.name == "Test Playlist" assert playlist.count == 0 assert playlist.is_empty is True def test_add_track(self): """Testet Hinzufügen von Tracks.""" playlist = Playlist(name="Test") track = Track(metadata=TrackMetadata(title="Song")) playlist.add(track) assert playlist.count == 1 def test_remove_track(self): """Testet Entfernen von Tracks.""" playlist = Playlist(name="Test") track = Track(id="t1") playlist.add(track) result = playlist.remove(0) assert result is not None assert playlist.count == 0 def test_next_sequential(self): """Testet sequentielle Navigation.""" playlist = Playlist(name="Test", shuffle_mode=ShuffleMode.OFF) playlist.add(Track(id="t1")) playlist.add(Track(id="t2")) playlist.add(Track(id="t3")) track1 = playlist.next() assert track1.id == "t1" track2 = playlist.next() assert track2.id == "t2" track3 = playlist.next() assert track3.id == "t3" def test_previous(self): """Testet Zurück-Navigation.""" playlist = Playlist(name="Test") playlist.add(Track(id="t1")) playlist.add(Track(id="t2")) playlist.next() # t1 playlist.next() # t2 prev = playlist.previous() assert prev.id == "t1" def test_repeat_off(self): """Testet Repeat-Modus OFF.""" playlist = Playlist(name="Test", repeat_mode=RepeatMode.OFF) playlist.add(Track(id="t1")) playlist.next() # t1 track = playlist.next() # Ende assert track is None def test_repeat_all(self): """Testet Repeat-Modus ALL.""" playlist = Playlist(name="Test", repeat_mode=RepeatMode.ALL) playlist.add(Track(id="t1")) playlist.add(Track(id="t2")) playlist.next() # t1 playlist.next() # t2 track = playlist.next() # Zurück zum Anfang assert track.id == "t1" def test_repeat_one(self): """Testet Repeat-Modus ONE.""" playlist = Playlist(name="Test", repeat_mode=RepeatMode.ONE) playlist.add(Track(id="t1")) playlist.add(Track(id="t2")) # Bei REPEAT_ONE bleibt next() beim aktuellen Track # Zuerst müssen wir zum ersten Track navigieren playlist.jump_to(0) assert playlist.current_track.id == "t1" track = playlist.next() # Sollte wieder t1 sein assert track.id == "t1" track = playlist.next() # Immer noch t1 assert track.id == "t1" def test_shuffle(self): """Testet Shuffle-Modus.""" playlist = Playlist(name="Test", shuffle_mode=ShuffleMode.ON) for i in range(10): playlist.add(Track(id=f"t{i}")) # Mehrfach durch die Playlist navigieren # Bei Shuffle sollte die Reihenfolge zufällig sein order = [] for _ in range(10): track = playlist.next() if track: order.append(track.id) # Bei Shuffle sollten alle Tracks vorkommen (ohne Repeat) assert len(set(order)) == len(order) # Keine Duplikate def test_jump_to(self): """Testet Sprung zu Index.""" playlist = Playlist(name="Test") playlist.add(Track(id="t1")) playlist.add(Track(id="t2")) playlist.add(Track(id="t3")) track = playlist.jump_to(2) assert track.id == "t3" assert playlist.current_index == 2 def test_clear(self): """Testet Playlist leeren.""" playlist = Playlist(name="Test") playlist.add(Track(id="t1")) playlist.add(Track(id="t2")) playlist.clear() assert playlist.count == 0 assert playlist.is_empty is True def test_total_duration(self): """Testet Gesamt-Dauer.""" playlist = Playlist(name="Test") playlist.add(Track(metadata=TrackMetadata(duration_ms=60000))) playlist.add(Track(metadata=TrackMetadata(duration_ms=120000))) assert playlist.total_duration_ms == 180000 def test_to_dict(self): """Testet to_dict Konvertierung.""" playlist = Playlist(name="Test Playlist") playlist.add(Track(id="t1")) result = playlist.to_dict() assert result["name"] == "Test Playlist" assert result["count"] == 1 assert len(result["items"]) == 1 def test_peek_next(self): """Testet peek_next.""" playlist = Playlist(name="Test", shuffle_mode=ShuffleMode.OFF) playlist.add(Track(id="t1")) playlist.add(Track(id="t2")) # Am Anfang - current_index ist -1, also ist nächster Track t1 next_track = playlist.peek_next() assert next_track.id == "t1" # Nach einem next() playlist.next() next_track = playlist.peek_next() assert next_track.id == "t2" # ============================================================================= # Integrationstests # ============================================================================= class TestMusicIntegration: """Integrationstests für das Musik-System.""" @pytest.mark.asyncio @pytest.mark.skipif(not has_music_files(), reason="Keine Musikdateien vorhanden") async def test_scan_and_queue(self): """Testet Scannen und Einreihen von Tracks.""" # Musikdateien scannen source = LocalFileSource(paths=[MUSIC_DIR]) await source.scan() assert source.track_count > 0 # Tracks in Queue einfügen queue = PlayQueue() for track in source.tracks: queue.add(track) assert queue.count == source.track_count # Durch Queue navigieren first = queue.next() assert first is not None assert first.track.path.exists() @pytest.mark.asyncio @pytest.mark.skipif(not has_music_files(), reason="Keine Musikdateien vorhanden") async def test_scan_and_playlist(self): """Testet Scannen und Playlist-Erstellung.""" # Musikdateien scannen source = LocalFileSource(paths=[MUSIC_DIR]) await source.scan() # Playlist erstellen playlist = Playlist(name="Test Mix") for track in source.tracks: playlist.add(track) assert playlist.count == source.track_count # Shuffle aktivieren und durchspielen playlist.set_shuffle(ShuffleMode.ON) played = [] # start() gibt den ersten Track zurück first = playlist.start() if first: played.append(first.id) # Dann mit next() fortfahren while True: track = playlist.next() if track is None: break played.append(track.id) assert len(played) == source.track_count @pytest.mark.asyncio @pytest.mark.skipif(not has_music_files(), reason="Keine Musikdateien vorhanden") async def test_full_streaming_pipeline(self): """Testet die komplette Streaming-Pipeline.""" # Scan source = LocalFileSource(paths=[MUSIC_DIR]) await source.scan() track = list(source.tracks)[0] # Streamer einrichten streamer = AudioStreamer() received_data = [] async def receive_callback(data: bytes) -> bool: received_data.append(data) return True streamer.add_satellite("test_client", receive_callback) # Streaming starten await streamer.start(track) # Daten füttern test_data = bytes(4096) await streamer.feed(test_data) # Kurz warten für Streaming-Loop await asyncio.sleep(0.2) # Stoppen await streamer.stop() # Prüfen ob Daten empfangen wurden # (Hängt davon ab ob genug gepuffert war) session = streamer.get_session("test_client") assert session is None or session.stats.bytes_sent >= 0 # ============================================================================= # Performance Tests # ============================================================================= class TestMusicPerformance: """Performance-Tests für das Musik-System.""" @pytest.mark.asyncio @pytest.mark.skipif(not has_music_files(), reason="Keine Musikdateien vorhanden") async def test_scan_performance(self): """Testet Scan-Performance.""" import time source = LocalFileSource(paths=[MUSIC_DIR]) start = time.monotonic() count = await source.scan() duration = time.monotonic() - start # Sollte schnell sein (< 1s für übliche Verzeichnisse) assert duration < 5.0 assert count > 0 def test_queue_performance(self): """Testet Queue-Performance mit vielen Tracks.""" import time queue = PlayQueue() # 1000 Tracks hinzufügen start = time.monotonic() for i in range(1000): queue.add(Track(id=f"track_{i}")) add_duration = time.monotonic() - start assert add_duration < 1.0 # Sollte < 1s sein # Durch alle navigieren start = time.monotonic() while queue.has_next: queue.next() nav_duration = time.monotonic() - start assert nav_duration < 1.0 # Sollte < 1s sein def test_playlist_shuffle_performance(self): """Testet Shuffle-Performance.""" import time playlist = Playlist(name="Big Playlist", shuffle_mode=ShuffleMode.ON) # 1000 Tracks for i in range(1000): playlist.add(Track(id=f"track_{i}")) # Shuffle und durchspielen mit start() + next() start = time.monotonic() count = 0 # start() gibt den ersten Track if playlist.start() is not None: count = 1 while playlist.next() is not None: count += 1 if count > 1000: break # Sicherheit gegen Endlosschleife duration = time.monotonic() - start assert duration < 2.0 assert count == 1000