# -*- coding: utf-8 -*- """ Tests fuer das Download-Modul. """ import pytest from pathlib import Path from unittest.mock import MagicMock, AsyncMock, patch from trixy_core.utils.download import ( DownloadProgress, download_file, download_from_huggingface, download_from_github, download_and_extract, format_size, ) from trixy_core.events.event_data.download import ( BeforeDownload, DownloadCompleted, DownloadFailed, ) class TestFormatSize: """Tests fuer die format_size Funktion.""" def test_bytes(self): """Testet Byte-Formatierung.""" assert format_size(500) == "500.0 B" def test_kilobytes(self): """Testet KB-Formatierung.""" assert format_size(1024) == "1.0 KB" assert format_size(2048) == "2.0 KB" def test_megabytes(self): """Testet MB-Formatierung.""" assert format_size(1024 * 1024) == "1.0 MB" assert format_size(5 * 1024 * 1024) == "5.0 MB" def test_gigabytes(self): """Testet GB-Formatierung.""" assert format_size(1024 * 1024 * 1024) == "1.0 GB" def test_terabytes(self): """Testet TB-Formatierung.""" assert format_size(1024 * 1024 * 1024 * 1024) == "1.0 TB" class TestDownloadProgress: """Tests fuer die DownloadProgress Klasse.""" def test_initial_state(self): """Testet den initialen Zustand.""" progress = DownloadProgress(total_size=1000) assert progress.downloaded == 0 assert progress.total_size == 1000 assert progress.percent == 0.0 def test_update(self): """Testet das Update.""" progress = DownloadProgress(total_size=1000) progress.update(500) assert progress.downloaded == 500 assert progress.percent == 50.0 def test_zero_total_size(self): """Testet mit total_size=0.""" progress = DownloadProgress(total_size=0) progress.update(100) assert progress.percent == 0.0 class TestDownloadFile: """Tests fuer die download_file Funktion.""" @pytest.mark.asyncio async def test_creates_parent_directories(self, tmp_path): """Testet, dass Parent-Verzeichnisse erstellt werden.""" dest_path = tmp_path / "subdir" / "file.txt" # Mock aiohttp with patch("trixy_core.utils.download.download_file") as mock_download: mock_download.return_value = True # Verzeichnis sollte erstellt werden dest_path.parent.mkdir(parents=True, exist_ok=True) assert dest_path.parent.exists() @pytest.mark.asyncio async def test_invalid_url_returns_false(self, tmp_path): """Testet, dass ungueltige URLs False zurueckgeben.""" dest_path = tmp_path / "file.txt" result = await download_file( url="https://invalid.url.that.does.not.exist/file.txt", dest_path=dest_path, timeout=5, ) assert result is False class TestDownloadFromGithub: """Tests fuer download_from_github.""" def test_url_construction(self): """Testet, dass die GitHub URL korrekt konstruiert wird.""" expected_url = "https://github.com/owner/repo/releases/download/v1.0.0/file.zip" assert "github.com" in expected_url assert "releases/download" in expected_url class TestDownloadEvents: """Tests fuer Download-Events mit trigger().""" @pytest.fixture def mock_event_manager(self): """Erstellt einen Mock-EventManager.""" em = MagicMock() em.trigger = AsyncMock() return em @pytest.mark.asyncio async def test_before_download_event(self, tmp_path, mock_event_manager): """Testet, dass before_download Event getriggert wird.""" dest_path = tmp_path / "file.txt" # trigger() gibt das Event-Objekt zurueck (nicht-gecancelled) mock_event_manager.trigger.return_value = BeforeDownload( download_id="test-download", url="https://example.com/nonexistent.txt", filename="file.txt", dest_path=str(dest_path), ) await download_file( url="https://example.com/nonexistent.txt", dest_path=dest_path, event_manager=mock_event_manager, download_id="test-download", timeout=5, ) # Pruefen, dass trigger() aufgerufen wurde calls = mock_event_manager.trigger.call_args_list event_names = [call[0][0] for call in calls] assert "before_download" in event_names # Da der Download fehlschlaegt, sollte auch download_failed getriggert werden assert "download_failed" in event_names @pytest.mark.asyncio async def test_before_download_cancel(self, tmp_path, mock_event_manager): """Testet, dass ein gecancelltes BeforeDownload den Download verhindert.""" dest_path = tmp_path / "file.txt" # trigger() gibt ein gecancelltes Event zurueck cancelled_event = BeforeDownload( download_id="test-download", url="https://example.com/file.txt", filename="file.txt", dest_path=str(dest_path), ) cancelled_event.cancel() mock_event_manager.trigger.return_value = cancelled_event result = await download_file( url="https://example.com/file.txt", dest_path=dest_path, event_manager=mock_event_manager, download_id="test-download", ) assert result is False # Nur before_download sollte getriggert worden sein — kein Download-Versuch calls = mock_event_manager.trigger.call_args_list assert len(calls) == 1 assert calls[0][0][0] == "before_download" @pytest.mark.asyncio async def test_before_download_redirect(self, tmp_path, mock_event_manager): """Testet, dass ein Plugin die URL aendern kann.""" dest_path = tmp_path / "file.txt" new_dest = tmp_path / "redirected.txt" # trigger() gibt Event mit geaenderter URL/Pfad zurueck redirected_event = BeforeDownload( download_id="test-download", url="https://mirror.local/file.txt", filename="redirected.txt", dest_path=str(new_dest), ) mock_event_manager.trigger.return_value = redirected_event await download_file( url="https://example.com/file.txt", dest_path=dest_path, event_manager=mock_event_manager, download_id="test-download", ) # Pruefen, dass download_failed mit der neuen URL getriggert wurde calls = mock_event_manager.trigger.call_args_list failed_calls = [c for c in calls if c[0][0] == "download_failed"] if failed_calls: failed_event = failed_calls[0][0][1] assert isinstance(failed_event, DownloadFailed) assert failed_event.url == "https://mirror.local/file.txt" class TestDownloadAndExtract: """Tests fuer download_and_extract.""" @pytest.mark.asyncio async def test_unsupported_format_fails(self, tmp_path): """Testet, dass nicht unterstuetzte Formate fehlschlagen.""" result = await download_and_extract( url="https://example.com/file.rar", dest_dir=tmp_path, ) assert result is False