# -*- coding: utf-8 -*- """ Tests für das Assets-Management-System. Testet: - AssetManager: Pfad-Auflösung, Fallback, Caching - AssetService: ServiceContainer-Integration - AssetInfo: Metadaten und Eigenschaften - AssetType: Typ-Erkennung """ import pytest import tempfile import time from pathlib import Path from unittest.mock import Mock, AsyncMock, patch from trixy_core.assets import ( AssetManager, AssetService, AssetType, AssetInfo, AssetNotFoundError, ) # ============================================================================= # Fixtures # ============================================================================= @pytest.fixture def temp_assets_dir(tmp_path): """Erstellt ein temporäres Assets-Verzeichnis mit Test-Struktur.""" # Default-Profil default_dir = tmp_path / "default" (default_dir / "audio").mkdir(parents=True) (default_dir / "music").mkdir(parents=True) (default_dir / "images").mkdir(parents=True) # Default-Assets erstellen (default_dir / "audio" / "success.wav").write_bytes(b"RIFF" + b"\x00" * 100) (default_dir / "audio" / "error.wav").write_bytes(b"RIFF" + b"\x00" * 50) (default_dir / "music" / "song1.mp3").write_bytes(b"ID3" + b"\x00" * 200) (default_dir / "images" / "logo.png").write_bytes(b"\x89PNG" + b"\x00" * 100) (default_dir / "config.json").write_text('{"key": "value"}') # Custom-Profil custom_dir = tmp_path / "custom" (custom_dir / "audio").mkdir(parents=True) # Custom-Asset (überschreibt default) (custom_dir / "audio" / "success.wav").write_bytes(b"RIFF" + b"\x00" * 150) # Zusätzliches Custom-Asset (custom_dir / "audio" / "custom_sound.wav").write_bytes(b"RIFF" + b"\x00" * 75) return tmp_path @pytest.fixture def asset_manager(temp_assets_dir): """Erstellt einen AssetManager mit Test-Verzeichnis.""" return AssetManager( base_dir=temp_assets_dir, profile="custom", cache_enabled=True, cache_ttl=60.0, ) @pytest.fixture def mock_application(): """Erstellt eine Mock-Application für AssetService-Tests.""" app = Mock() app.config = Mock() app.config.assets_directory = "assets" app.config.profile = "default" return app # ============================================================================= # AssetType Tests # ============================================================================= class TestAssetType: """Tests für AssetType-Enum.""" def test_audio_type(self): """Audio-Typ existiert.""" assert AssetType.AUDIO is not None assert AssetType.AUDIO.name == "AUDIO" def test_music_type(self): """Musik-Typ existiert.""" assert AssetType.MUSIC is not None def test_image_type(self): """Bild-Typ existiert.""" assert AssetType.IMAGE is not None def test_all_types(self): """Alle Typen sind definiert.""" expected_types = ["AUDIO", "MUSIC", "IMAGE", "VIDEO", "CONFIG", "DATA", "MODEL", "UNKNOWN"] for type_name in expected_types: assert hasattr(AssetType, type_name) # ============================================================================= # AssetInfo Tests # ============================================================================= class TestAssetInfo: """Tests für AssetInfo-Dataclass.""" def test_basic_properties(self, temp_assets_dir): """Grundlegende Eigenschaften werden korrekt gesetzt.""" path = temp_assets_dir / "default" / "audio" / "success.wav" info = AssetInfo( path=path, relative_path="audio/success.wav", profile="default", is_default=True, asset_type=AssetType.AUDIO, size=104, modified=time.time(), ) assert info.path == path assert info.relative_path == "audio/success.wav" assert info.profile == "default" assert info.is_default is True assert info.asset_type == AssetType.AUDIO assert info.size == 104 def test_extension_property(self, temp_assets_dir): """Extension-Property funktioniert.""" path = temp_assets_dir / "default" / "audio" / "success.wav" info = AssetInfo( path=path, relative_path="audio/success.wav", profile="default", is_default=True, asset_type=AssetType.AUDIO, size=100, modified=time.time(), ) assert info.extension == ".wav" def test_name_property(self, temp_assets_dir): """Name-Property funktioniert.""" path = temp_assets_dir / "default" / "audio" / "success.wav" info = AssetInfo( path=path, relative_path="audio/success.wav", profile="default", is_default=True, asset_type=AssetType.AUDIO, size=100, modified=time.time(), ) assert info.name == "success.wav" assert info.stem == "success" def test_exists_property(self, temp_assets_dir): """Exists-Property prüft Datei-Existenz.""" path = temp_assets_dir / "default" / "audio" / "success.wav" info = AssetInfo( path=path, relative_path="audio/success.wav", profile="default", is_default=True, asset_type=AssetType.AUDIO, size=100, modified=time.time(), ) assert info.exists is True # Datei löschen path.unlink() assert info.exists is False def test_checksum_calculation(self, temp_assets_dir): """Checksum wird korrekt berechnet.""" path = temp_assets_dir / "default" / "audio" / "success.wav" info = AssetInfo( path=path, relative_path="audio/success.wav", profile="default", is_default=True, asset_type=AssetType.AUDIO, size=104, modified=time.time(), ) checksum = info.checksum assert len(checksum) == 32 # MD5 hex length # Zweiter Aufruf sollte gecached sein assert info.checksum == checksum # ============================================================================= # AssetManager Tests # ============================================================================= class TestAssetManagerInit: """Tests für AssetManager-Initialisierung.""" def test_init_creates_directories(self, tmp_path): """Initialisierung erstellt Verzeichnisse.""" manager = AssetManager( base_dir=tmp_path / "assets", profile="test_profile", ) assert (tmp_path / "assets" / "default").exists() assert (tmp_path / "assets" / "test_profile").exists() def test_init_default_values(self, tmp_path): """Standard-Werte werden gesetzt.""" manager = AssetManager(base_dir=tmp_path) assert manager.profile == "default" assert manager._cache_enabled is True assert manager._cache_ttl == 60.0 def test_base_dir_property(self, asset_manager, temp_assets_dir): """Base-Dir Property funktioniert.""" assert asset_manager.base_dir == temp_assets_dir def test_profile_property(self, asset_manager): """Profile Property funktioniert.""" assert asset_manager.profile == "custom" def test_profile_dir_property(self, asset_manager, temp_assets_dir): """Profile-Dir Property funktioniert.""" assert asset_manager.profile_dir == temp_assets_dir / "custom" def test_default_dir_property(self, asset_manager, temp_assets_dir): """Default-Dir Property funktioniert.""" assert asset_manager.default_dir == temp_assets_dir / "default" class TestAssetManagerGet: """Tests für Asset-Abruf.""" def test_get_from_profile(self, asset_manager, temp_assets_dir): """Asset wird aus Profil geladen.""" path = asset_manager.get("audio/success.wav") assert path is not None assert path == temp_assets_dir / "custom" / "audio" / "success.wav" def test_get_fallback_to_default(self, asset_manager, temp_assets_dir): """Fallback auf Default funktioniert.""" path = asset_manager.get("audio/error.wav") assert path is not None assert path == temp_assets_dir / "default" / "audio" / "error.wav" def test_get_not_found_returns_none(self, asset_manager): """Nicht gefundenes Asset gibt None zurück.""" path = asset_manager.get("audio/nonexistent.wav") assert path is None def test_get_raises_on_missing(self, asset_manager): """AssetNotFoundError wird bei raise_on_missing geworfen.""" with pytest.raises(AssetNotFoundError) as exc_info: asset_manager.get("audio/nonexistent.wav", raise_on_missing=True) assert "nonexistent.wav" in str(exc_info.value) assert exc_info.value.profile == "custom" def test_get_info_returns_details(self, asset_manager, temp_assets_dir): """get_info gibt detaillierte Informationen zurück.""" info = asset_manager.get_info("audio/success.wav") assert info is not None assert info.profile == "custom" assert info.is_default is False assert info.asset_type == AssetType.AUDIO assert info.size == 154 # RIFF (4 bytes) + 150 bytes (custom profile) def test_get_info_default_fallback(self, asset_manager, temp_assets_dir): """get_info nutzt Default-Fallback.""" info = asset_manager.get_info("audio/error.wav") assert info is not None assert info.profile == "default" assert info.is_default is True def test_get_info_not_found(self, asset_manager): """get_info gibt None zurück wenn nicht gefunden.""" info = asset_manager.get_info("nonexistent.file") assert info is None def test_exists(self, asset_manager): """exists prüft korrekt.""" assert asset_manager.exists("audio/success.wav") is True assert asset_manager.exists("audio/error.wav") is True assert asset_manager.exists("nonexistent.file") is False class TestAssetManagerTypedAccess: """Tests für typ-spezifische Zugriffe.""" def test_get_audio(self, asset_manager, temp_assets_dir): """get_audio funktioniert.""" path = asset_manager.get_audio("success.wav") assert path is not None assert "audio" in str(path) def test_get_music(self, asset_manager, temp_assets_dir): """get_music funktioniert.""" path = asset_manager.get_music("song1.mp3") assert path is not None assert "music" in str(path) def test_get_image(self, asset_manager, temp_assets_dir): """get_image funktioniert.""" path = asset_manager.get_image("logo.png") assert path is not None assert "images" in str(path) def test_get_nonexistent_typed(self, asset_manager): """Typ-Methoden geben None für nicht existente Assets.""" assert asset_manager.get_audio("nonexistent.wav") is None assert asset_manager.get_music("nonexistent.mp3") is None assert asset_manager.get_image("nonexistent.png") is None class TestAssetManagerCaching: """Tests für das Cache-System.""" def test_cache_hit(self, asset_manager): """Cache wird genutzt.""" # Erster Aufruf info1 = asset_manager.get_info("audio/success.wav") # Zweiter Aufruf (sollte gecached sein) info2 = asset_manager.get_info("audio/success.wav") assert info1 is info2 # Gleiche Instanz def test_cache_disabled(self, temp_assets_dir): """Cache kann deaktiviert werden.""" manager = AssetManager( base_dir=temp_assets_dir, profile="custom", cache_enabled=False, ) info1 = manager.get_info("audio/success.wav") info2 = manager.get_info("audio/success.wav") # Nicht gleiche Instanz assert info1 is not info2 def test_cache_clear(self, asset_manager): """Cache kann geleert werden.""" # Cache füllen asset_manager.get_info("audio/success.wav") assert len(asset_manager._cache) > 0 # Cache leeren asset_manager.clear_cache() assert len(asset_manager._cache) == 0 def test_cache_ttl_expiration(self, temp_assets_dir): """Cache-Einträge verfallen nach TTL.""" manager = AssetManager( base_dir=temp_assets_dir, profile="custom", cache_enabled=True, cache_ttl=0.1, # 100ms ) # Cache füllen info1 = manager.get_info("audio/success.wav") assert "audio/success.wav" in manager._cache # Warten bis TTL abgelaufen time.sleep(0.15) # Neuer Abruf sollte Cache-Miss sein info2 = manager.get_info("audio/success.wav") # Neue Instanz assert info1 is not info2 class TestAssetManagerListing: """Tests für Asset-Listing.""" def test_list_assets_all(self, asset_manager): """Alle Assets werden gelistet.""" assets = list(asset_manager.list_assets()) # Sollte mindestens die bekannten Assets enthalten assert len(assets) >= 5 # success.wav, error.wav, custom_sound.wav, song1.mp3, logo.png, config.json def test_list_assets_by_type(self, asset_manager): """Filterung nach Typ funktioniert.""" audio_assets = list(asset_manager.list_assets(AssetType.AUDIO)) # Alle sollten Audio sein for info in audio_assets: assert info.asset_type == AssetType.AUDIO def test_list_assets_by_directory(self, asset_manager): """Filterung nach Verzeichnis funktioniert.""" audio_assets = list(asset_manager.list_assets(directory="audio")) for info in audio_assets: assert "audio" in info.relative_path def test_list_assets_profile_priority(self, asset_manager, temp_assets_dir): """Profil-Assets haben Priorität über Default.""" audio_assets = list(asset_manager.list_assets(directory="audio")) # Finde success.wav success_info = None for info in audio_assets: if info.name == "success.wav": success_info = info break assert success_info is not None assert success_info.profile == "custom" assert success_info.is_default is False def test_list_assets_exclude_default(self, asset_manager): """Default kann ausgeschlossen werden.""" assets = list(asset_manager.list_assets(include_default=False)) for info in assets: assert info.is_default is False def test_list_audio(self, asset_manager): """list_audio funktioniert.""" audio_assets = list(asset_manager.list_audio()) assert len(audio_assets) >= 3 # success, error, custom_sound def test_list_music(self, asset_manager): """list_music funktioniert.""" music_assets = list(asset_manager.list_music()) assert len(music_assets) >= 1 # song1.mp3 def test_list_profiles(self, asset_manager): """list_profiles gibt alle Profile zurück.""" profiles = asset_manager.list_profiles() assert "default" in profiles assert "custom" in profiles # Default sollte zuerst sein assert profiles[0] == "default" class TestAssetManagerProfileSwitch: """Tests für Profil-Wechsel.""" def test_profile_switch_clears_cache(self, asset_manager): """Profil-Wechsel leert den Cache.""" # Cache füllen asset_manager.get_info("audio/success.wav") assert len(asset_manager._cache) > 0 # Profil wechseln asset_manager.profile = "default" # Cache sollte leer sein assert len(asset_manager._cache) == 0 def test_profile_switch_creates_directory(self, temp_assets_dir): """Profil-Wechsel erstellt Verzeichnis.""" manager = AssetManager(base_dir=temp_assets_dir, profile="default") # Neues Profil manager.profile = "new_profile" assert (temp_assets_dir / "new_profile").exists() def test_profile_switch_affects_resolution(self, asset_manager, temp_assets_dir): """Profil-Wechsel beeinflusst Pfad-Auflösung.""" # Im custom-Profil info1 = asset_manager.get_info("audio/success.wav") assert info1.profile == "custom" # Wechsel zu default asset_manager.profile = "default" # Jetzt sollte default verwendet werden info2 = asset_manager.get_info("audio/success.wav") assert info2.profile == "default" class TestAssetManagerCopy: """Tests für Asset-Kopieren.""" def test_copy_to_profile(self, asset_manager, temp_assets_dir): """Asset kann in Profil kopiert werden.""" # error.wav ist nur in default assert asset_manager.get_info("audio/error.wav").is_default is True # Kopieren new_path = asset_manager.copy_to_profile("audio/error.wav") assert new_path is not None assert new_path.exists() assert "custom" in str(new_path) # Cache sollte invalidiert sein info = asset_manager.get_info("audio/error.wav") assert info.profile == "custom" assert info.is_default is False def test_copy_nonexistent_returns_none(self, asset_manager): """Kopieren nicht existenter Assets gibt None.""" result = asset_manager.copy_to_profile("nonexistent.file") assert result is None def test_copy_already_in_profile(self, asset_manager, temp_assets_dir): """Kopieren eines bereits im Profil vorhandenen Assets.""" # success.wav ist im custom-Profil result = asset_manager.copy_to_profile("audio/success.wav") # Sollte bestehenden Pfad zurückgeben assert result is not None assert "custom" in str(result) class TestAssetManagerStats: """Tests für Statistiken.""" def test_get_stats(self, asset_manager): """get_stats gibt korrektes Dictionary zurück.""" stats = asset_manager.get_stats() assert stats["profile"] == "custom" assert "default" in stats["profiles"] assert "custom" in stats["profiles"] assert stats["counts"]["total"] > 0 assert stats["sizes"]["total"] > 0 def test_stats_counts_types(self, asset_manager): """Stats zählen Asset-Typen.""" stats = asset_manager.get_stats() assert stats["counts"]["audio"] >= 3 assert stats["counts"]["music"] >= 1 class TestAssetManagerTypeDetection: """Tests für Typ-Erkennung.""" def test_detect_audio_type(self, asset_manager): """Audio-Dateien werden erkannt.""" info = asset_manager.get_info("audio/success.wav") assert info.asset_type == AssetType.AUDIO def test_detect_music_type(self, asset_manager): """Musik-Dateien werden erkannt.""" info = asset_manager.get_info("music/song1.mp3") # music/ Verzeichnis hat Vorrang assert info.asset_type == AssetType.MUSIC def test_detect_image_type(self, asset_manager): """Bild-Dateien werden erkannt.""" info = asset_manager.get_info("images/logo.png") assert info.asset_type == AssetType.IMAGE def test_detect_config_type(self, asset_manager): """Konfig-Dateien werden erkannt.""" info = asset_manager.get_info("config.json") assert info.asset_type == AssetType.CONFIG class TestAssetNotFoundError: """Tests für AssetNotFoundError.""" def test_error_contains_details(self, asset_manager): """Error enthält hilfreiche Details.""" try: asset_manager.get("nonexistent/file.wav", raise_on_missing=True) except AssetNotFoundError as e: assert e.asset_path == "nonexistent/file.wav" assert e.profile == "custom" assert len(e.searched_paths) == 2 # ============================================================================= # AssetService Tests # ============================================================================= class TestAssetServiceInit: """Tests für AssetService-Initialisierung.""" def test_init(self, mock_application): """Service kann initialisiert werden.""" service = AssetService(mock_application) assert service._manager is None assert service.state.name == "STOPPED" def test_service_metadata(self, mock_application): """Service-Metadaten sind korrekt.""" service = AssetService(mock_application) assert service.NAME == "AssetService" assert service.PRIORITY.name == "CRITICAL" assert service.GROUP.name == "UTILITY" @pytest.mark.asyncio class TestAssetServiceLifecycle: """Tests für Service-Lebenszyklus.""" async def test_start(self, mock_application, tmp_path): """Service kann gestartet werden.""" mock_application.config.assets_directory = str(tmp_path / "assets") mock_application.config.profile = "test_profile" service = AssetService(mock_application) await service.start() assert service.is_running assert service._manager is not None assert service.profile == "test_profile" async def test_stop(self, mock_application, tmp_path): """Service kann gestoppt werden.""" mock_application.config.assets_directory = str(tmp_path / "assets") mock_application.config.profile = "default" service = AssetService(mock_application) await service.start() await service.stop() assert not service.is_running assert service._manager is None async def test_health_check(self, mock_application, tmp_path): """Health-Check funktioniert.""" mock_application.config.assets_directory = str(tmp_path / "assets") mock_application.config.profile = "default" service = AssetService(mock_application) # Vor Start assert await service.health_check() is False await service.start() # Nach Start assert await service.health_check() is True await service.stop() async def test_manager_access_before_start(self, mock_application): """Manager-Zugriff vor Start wirft Fehler.""" service = AssetService(mock_application) with pytest.raises(RuntimeError, match="nicht gestartet"): _ = service.manager @pytest.mark.asyncio class TestAssetServiceDelegation: """Tests für delegierte Methoden.""" async def test_get_delegation(self, mock_application, temp_assets_dir): """get wird delegiert.""" mock_application.config.assets_directory = str(temp_assets_dir) mock_application.config.profile = "custom" service = AssetService(mock_application) await service.start() path = service.get("audio/success.wav") assert path is not None await service.stop() async def test_exists_delegation(self, mock_application, temp_assets_dir): """exists wird delegiert.""" mock_application.config.assets_directory = str(temp_assets_dir) mock_application.config.profile = "custom" service = AssetService(mock_application) await service.start() assert service.exists("audio/success.wav") is True assert service.exists("nonexistent.file") is False await service.stop() async def test_profile_property(self, mock_application, temp_assets_dir): """Profile-Property wird delegiert.""" mock_application.config.assets_directory = str(temp_assets_dir) mock_application.config.profile = "custom" service = AssetService(mock_application) await service.start() assert service.profile == "custom" service.profile = "default" assert service.profile == "default" await service.stop() async def test_list_assets_delegation(self, mock_application, temp_assets_dir): """list_assets wird delegiert.""" mock_application.config.assets_directory = str(temp_assets_dir) mock_application.config.profile = "custom" service = AssetService(mock_application) await service.start() assets = list(service.list_assets()) assert len(assets) > 0 await service.stop() async def test_get_stats_delegation(self, mock_application, temp_assets_dir): """get_stats wird delegiert.""" mock_application.config.assets_directory = str(temp_assets_dir) mock_application.config.profile = "custom" service = AssetService(mock_application) await service.start() stats = service.get_stats() assert "profile" in stats assert "counts" in stats await service.stop() # ============================================================================= # Integration mit echten Assets # ============================================================================= class TestRealAssets: """Tests mit echten Assets aus dem Projekt.""" @pytest.fixture def real_asset_manager(self): """AssetManager mit echten Projekt-Assets.""" assets_path = Path(__file__).parent.parent / "assets" if not assets_path.exists(): pytest.skip("Assets-Verzeichnis nicht gefunden") return AssetManager(base_dir=assets_path, profile="default") def test_real_audio_exists(self, real_asset_manager): """Echte Audio-Assets existieren.""" path = real_asset_manager.get("audio/success.wav") if path is None: pytest.skip("success.wav nicht gefunden") assert path.exists() def test_real_music_listing(self, real_asset_manager): """Echte Musik-Assets können gelistet werden.""" music = list(real_asset_manager.list_music()) if len(music) == 0: pytest.skip("Keine Musik-Assets gefunden") for info in music: assert info.asset_type == AssetType.MUSIC assert info.exists def test_real_asset_info(self, real_asset_manager): """AssetInfo für echte Assets funktioniert.""" info = real_asset_manager.get_info("audio/success.wav") if info is None: pytest.skip("success.wav nicht gefunden") assert info.size > 0 assert info.modified > 0 assert len(info.checksum) == 32 # ============================================================================= # Edge Cases # ============================================================================= class TestEdgeCases: """Tests für Grenzfälle.""" def test_path_normalization(self, asset_manager): """Pfade werden normalisiert.""" # Backslashes path1 = asset_manager.get("audio\\success.wav") # Leading slash path2 = asset_manager.get("/audio/success.wav") assert path1 == path2 def test_empty_profile(self, temp_assets_dir): """Leeres Profil verwendet Default.""" manager = AssetManager(base_dir=temp_assets_dir, profile="") # Sollte trotzdem funktionieren assert manager.profile == "" assert manager.profile_dir == temp_assets_dir / "" def test_special_characters_in_filename(self, temp_assets_dir): """Sonderzeichen in Dateinamen werden unterstützt.""" # Asset mit Sonderzeichen erstellen special_file = temp_assets_dir / "default" / "audio" / "test file (1).wav" special_file.write_bytes(b"RIFF" + b"\x00" * 50) manager = AssetManager(base_dir=temp_assets_dir, profile="default") path = manager.get("audio/test file (1).wav") assert path is not None assert path.exists() def test_unicode_in_filename(self, temp_assets_dir): """Unicode in Dateinamen wird unterstützt.""" # Asset mit Unicode erstellen unicode_file = temp_assets_dir / "default" / "audio" / "müsik.wav" unicode_file.write_bytes(b"RIFF" + b"\x00" * 50) manager = AssetManager(base_dir=temp_assets_dir, profile="default") path = manager.get("audio/müsik.wav") assert path is not None assert path.exists() def test_nested_directories(self, temp_assets_dir): """Verschachtelte Verzeichnisse werden unterstützt.""" # Verschachteltes Asset erstellen nested_dir = temp_assets_dir / "default" / "audio" / "effects" / "explosions" nested_dir.mkdir(parents=True) (nested_dir / "boom.wav").write_bytes(b"RIFF" + b"\x00" * 50) manager = AssetManager(base_dir=temp_assets_dir, profile="default") path = manager.get("audio/effects/explosions/boom.wav") assert path is not None assert path.exists() def test_concurrent_access(self, asset_manager): """Concurrent Zugriff ist thread-safe.""" import concurrent.futures def access_asset(i): return asset_manager.get_info("audio/success.wav") with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: futures = [executor.submit(access_asset, i) for i in range(50)] results = [f.result() for f in futures] # Alle sollten erfolgreich sein assert all(r is not None for r in results) if __name__ == "__main__": pytest.main([__file__, "-v"])