diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_download_duplicate_prevention.py | 394 | ||||
| -rw-r--r-- | tests/test_gui_duplicate_prevention.py | 335 | ||||
| -rw-r--r-- | tests/test_history_service.py | 550 | ||||
| -rw-r--r-- | tests/test_logger_configuration.py | 277 |
4 files changed, 1556 insertions, 0 deletions
diff --git a/tests/test_download_duplicate_prevention.py b/tests/test_download_duplicate_prevention.py new file mode 100644 index 0000000..d6caee0 --- /dev/null +++ b/tests/test_download_duplicate_prevention.py @@ -0,0 +1,394 @@ +""" +test_download_duplicate_prevention.py + +Test suite for download duplicate prevention functionality. + +Tests cover: +- Duplicate prevention during downloads +- Integration with HistoryService +- Skipped item logging +- Settings toggle behavior +""" + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from tidalapi import Quality, Track + +from tidal_dl_ng.download import Download +from tidal_dl_ng.history import HistoryService + + +@pytest.fixture +def mock_tidal(): + """Create a mock Tidal object for testing. + + Returns: + Mock Tidal object with session. + """ + mock = MagicMock() + mock.session = MagicMock() + mock.session.audio_quality = Quality.high_lossless + mock.switch_to_atmos_session = MagicMock(return_value=True) + mock.restore_normal_session = MagicMock(return_value=True) + mock.stream_lock = MagicMock() + mock.stream_lock.__enter__ = MagicMock(return_value=None) + mock.stream_lock.__exit__ = MagicMock(return_value=None) + mock.settings = MagicMock() + mock.settings.data = MagicMock() + return mock + + +@pytest.fixture +def mock_logger(): + """Create a mock logger for testing. + + Returns: + Mock logger object. + """ + logger = MagicMock() + logger.info = MagicMock() + logger.debug = MagicMock() + logger.warning = MagicMock() + logger.error = MagicMock() + return logger + + +@pytest.fixture +def temp_download_path(tmp_path: Path) -> Path: + """Create a temporary download directory. + + Args: + tmp_path: pytest fixture providing temporary directory. + + Returns: + Path to temporary download directory. + """ + download_dir = tmp_path / "downloads" + download_dir.mkdir() + return download_dir + + +@pytest.fixture +def history_service_with_temp(tmp_path: Path, monkeypatch) -> HistoryService: + """Create a HistoryService with temporary storage. + + Args: + tmp_path: pytest fixture providing temporary directory. + monkeypatch: pytest monkeypatch fixture. + + Returns: + Configured HistoryService instance. + """ + config_path = tmp_path / "config" + config_path.mkdir(parents=True, exist_ok=True) + + monkeypatch.setattr("tidal_dl_ng.history.path_config_base", lambda: str(config_path)) + + # Reset singleton + if hasattr(HistoryService, "_instances"): + HistoryService._instances = {} + + return HistoryService() + + +@pytest.fixture +def download_instance(mock_tidal, mock_logger, temp_download_path, history_service_with_temp): + """Create a Download instance for testing. + + Args: + mock_tidal: Mock Tidal object. + mock_logger: Mock logger. + temp_download_path: Temporary download directory. + history_service_with_temp: HistoryService with temp storage. + + Returns: + Configured Download instance. + """ + with patch("tidal_dl_ng.download.HistoryService", return_value=history_service_with_temp): + download = Download( + tidal_obj=mock_tidal, path_base=str(temp_download_path), fn_logger=mock_logger, skip_existing=False + ) + return download + + +class TestDuplicatePreventionInDownload: + """Test duplicate prevention during download operations.""" + + def test_skip_download_when_track_in_history_and_prevention_enabled(self, download_instance: Download, mock_logger): + """Test that download is skipped when track is in history and prevention is enabled.""" + # Add track to history + track_id = "123456" + download_instance.history_service.add_track_to_history(track_id) + download_instance.history_service.update_settings(preventDuplicates=True) + + # Create mock track + mock_track = MagicMock(spec=Track) + mock_track.id = int(track_id) + mock_track.name = "Test Track" + mock_track.available = True + mock_track.album = MagicMock() + mock_track.album.name = "Test Album" + + # Attempt download + with patch.object(download_instance.session, "track", return_value=mock_track): + result, path = download_instance.item(file_template="{artist_name} - {track_title}", media=mock_track) + + # Should be skipped (result False means skipped) + assert result is False + # Logger should have logged the skip message + assert mock_logger.info.called + skip_call_found = any( + "Skipped item" in str(call) or "already in history" in str(call) for call in mock_logger.info.call_args_list + ) + assert skip_call_found + + def test_download_proceeds_when_prevention_disabled(self, download_instance: Download, mock_logger): + """Test that download proceeds when prevention is disabled.""" + # Add track to history but disable prevention + track_id = "789012" + download_instance.history_service.add_track_to_history(track_id) + download_instance.history_service.update_settings(preventDuplicates=False) + + # Create mock track + mock_track = MagicMock(spec=Track) + mock_track.id = int(track_id) + mock_track.name = "Test Track 2" + mock_track.available = True + + # Mock the download process to avoid actual download + with ( + patch.object(download_instance, "_validate_and_prepare_media", return_value=mock_track), + patch.object(download_instance, "_prepare_file_paths_and_skip_logic") as mock_paths, + ): + + mock_paths.return_value = (Path("/fake/path.flac"), ".flac", False, False) + + # The download should not be skipped by duplicate prevention + # (it will still fail later in the chain, but that's okay for this test) + result = download_instance.history_service.should_skip_download(track_id) + + # Should NOT be skipped + assert result is False + + def test_download_proceeds_when_track_not_in_history(self, download_instance: Download): + """Test that download proceeds when track is not in history.""" + track_id = "999999" + download_instance.history_service.update_settings(preventDuplicates=True) + + # Track not in history + result = download_instance.history_service.should_skip_download(track_id) + + assert result is False + + +class TestHistoryLoggingMessages: + """Test logging messages for skipped downloads.""" + + def test_skip_message_format(self, download_instance: Download, mock_logger): + """Test that skip message follows the correct format.""" + track_id = "111222" + download_instance.history_service.add_track_to_history(track_id) + download_instance.history_service.update_settings(preventDuplicates=True) + + mock_track = MagicMock(spec=Track) + mock_track.id = int(track_id) + mock_track.name = "Amazing Song" + mock_track.artist = MagicMock() + mock_track.artist.name = "Great Artist" + mock_track.available = True + mock_track.album = MagicMock() + + with patch.object(download_instance.session, "track", return_value=mock_track): + download_instance.item(file_template="{artist_name} - {track_title}", media=mock_track) + + # Check that info logger was called + assert mock_logger.info.called + + # Verify message contains expected parts + log_calls = [str(call) for call in mock_logger.info.call_args_list] + skip_message_found = any("Skipped item" in msg and "already in history" in msg for msg in log_calls) + assert skip_message_found + + +class TestHistoryAfterDownload: + """Test that successful downloads are added to history.""" + + @patch("tidal_dl_ng.download.Download._download_and_process_media") + @patch("tidal_dl_ng.download.Download._adjust_quality_settings") + @patch("tidal_dl_ng.download.Download._perform_post_processing") + def test_track_added_to_history_after_successful_download( + self, mock_post_process, mock_quality, mock_download_process, download_instance: Download + ): + """Test that a track is added to history after successful download.""" + # Mock successful download + mock_download_process.return_value = True + mock_quality.return_value = (None, None) + + track_id = "333444" + mock_track = MagicMock(spec=Track) + mock_track.id = int(track_id) + mock_track.name = "New Track" + mock_track.available = True + mock_track.album = MagicMock() + + with ( + patch.object(download_instance, "_validate_and_prepare_media", return_value=mock_track), + patch.object(download_instance, "_prepare_file_paths_and_skip_logic") as mock_paths, + ): + + mock_paths.return_value = (Path("/fake/path.flac"), ".flac", False, False) + + download_instance.item( + file_template="{artist_name} - {track_title}", + media=mock_track, + source_type="playlist", + source_id="pl-123", + source_name="Test Playlist", + ) + + # Track should be in history + assert download_instance.history_service.is_downloaded(track_id) + + # Check metadata + track_info = download_instance.history_service.get_track_info(track_id) + assert track_info is not None + assert track_info["sourceType"] == "playlist" + assert track_info["sourceId"] == "pl-123" + + @patch("tidal_dl_ng.download.Download._download_and_process_media") + def test_track_not_added_on_failed_download(self, mock_download_process, download_instance: Download): + """Test that track is NOT added to history if download fails.""" + # Mock failed download + mock_download_process.return_value = False + + track_id = "555666" + mock_track = MagicMock(spec=Track) + mock_track.id = int(track_id) + mock_track.available = True + mock_track.album = MagicMock() + + with ( + patch.object(download_instance, "_validate_and_prepare_media", return_value=mock_track), + patch.object(download_instance, "_prepare_file_paths_and_skip_logic") as mock_paths, + patch.object(download_instance, "_adjust_quality_settings") as mock_quality, + ): + + mock_paths.return_value = (Path("/fake/path.flac"), ".flac", False, False) + mock_quality.return_value = (None, None) + + download_instance.item(file_template="{artist_name} - {track_title}", media=mock_track) + + # Track should NOT be in history + assert not download_instance.history_service.is_downloaded(track_id) + + +class TestSettingsIntegration: + """Test integration with settings toggle.""" + + def test_settings_toggle_affects_duplicate_check(self, download_instance: Download): + """Test that toggling settings immediately affects duplicate checking.""" + track_id = "777888" + download_instance.history_service.add_track_to_history(track_id) + + # Enable prevention + download_instance.history_service.update_settings(preventDuplicates=True) + assert download_instance.history_service.should_skip_download(track_id) is True + + # Disable prevention + download_instance.history_service.update_settings(preventDuplicates=False) + assert download_instance.history_service.should_skip_download(track_id) is False + + # Re-enable + download_instance.history_service.update_settings(preventDuplicates=True) + assert download_instance.history_service.should_skip_download(track_id) is True + + def test_settings_persist_across_service_instances(self, tmp_path: Path, monkeypatch): + """Test that settings persist across service instances.""" + config_path = tmp_path / "config" + config_path.mkdir(parents=True, exist_ok=True) + + monkeypatch.setattr("tidal_dl_ng.history.path_config_base", lambda: str(config_path)) + + # Create first instance and disable prevention + if hasattr(HistoryService, "_instances"): + HistoryService._instances = {} + + service1 = HistoryService() + service1.update_settings(preventDuplicates=False) + + # Create second instance (simulating app restart) + HistoryService._instances = {} + service2 = HistoryService() + + # Settings should persist + assert service2.get_settings()["preventDuplicates"] is False + + +class TestVideoNotAffectedByHistory: + """Test that video downloads are not affected by history (only tracks).""" + + def test_video_download_not_checked_against_history(self, download_instance: Download): + """Test that videos are not checked against download history.""" + from tidalapi import Video + + # This test verifies the type check in the duplicate prevention code + # Videos should bypass the history check entirely + + mock_video = MagicMock(spec=Video) + mock_video.id = 123456 + mock_video.name = "Test Video" + mock_video.available = True + + # Even if we somehow had a video ID in history, it shouldn't matter + # because the check is `isinstance(media, Track)` + download_instance.history_service.add_track_to_history("123456") + + with ( + patch.object(download_instance, "_validate_and_prepare_media", return_value=mock_video), + patch.object(download_instance, "_prepare_file_paths_and_skip_logic") as mock_paths, + patch.object(download_instance.history_service, "should_skip_download") as mock_skip, + ): + mock_paths.return_value = (Path("/fake.mp4"), ".mp4", False, False) + + # The key is that should_skip_download is never called for videos + # Expected to fail, we're just checking the skip wasn't called + from contextlib import suppress + + with suppress(Exception): + download_instance.item(file_template="{artist_name} - {track_title}", media=mock_video) + + # should_skip_download should NOT have been called for video + assert not mock_skip.called + + +class TestBulkDownloadWithHistory: + """Test history tracking in bulk download scenarios.""" + + @patch("tidal_dl_ng.download.Download.item") + def test_multiple_tracks_added_to_history(self, mock_item_download, download_instance: Download): + """Test that multiple tracks in a bulk download are tracked.""" + # Simulate successful downloads + mock_item_download.return_value = (True, Path("/fake/path.flac")) + + track_ids = ["t1", "t2", "t3", "t4", "t5"] + + # Manually add tracks to history (simulating what item() would do) + for track_id in track_ids: + download_instance.history_service.add_track_to_history( + track_id=track_id, source_type="album", source_id="al-123", source_name="Test Album" + ) + + # All tracks should be in history + for track_id in track_ids: + assert download_instance.history_service.is_downloaded(track_id) + + # Should be grouped under same album + grouped = download_instance.history_service.get_history_by_source() + assert "album_al-123" in grouped + assert len(grouped["album_al-123"]) == 5 + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_gui_duplicate_prevention.py b/tests/test_gui_duplicate_prevention.py new file mode 100644 index 0000000..52932e7 --- /dev/null +++ b/tests/test_gui_duplicate_prevention.py @@ -0,0 +1,335 @@ +""" +test_gui_duplicate_prevention.py + +Test suite for GUI duplicate prevention features. + +Tests cover: +- Tools menu action creation +- Duplicate prevention toggle +- Settings persistence in GUI +- Status messages +""" + +from unittest.mock import MagicMock, patch + +import pytest + +# Mock Qt before importing GUI modules +with patch("tidal_dl_ng.gui.QtWidgets"), patch("tidal_dl_ng.gui.QtGui"), patch("tidal_dl_ng.gui.QtCore"): + pass + + +class TestToolsMenuIntegration: + """Test Tools menu integration for duplicate prevention.""" + + @patch("tidal_dl_ng.gui.QtWidgets.QMenuBar") + @patch("tidal_dl_ng.gui.QtGui.QAction") + def test_tools_menu_created_if_not_exists(self, mock_action, mock_menubar): + """Test that Tools menu is created if it doesn't exist.""" + # This test verifies the logic in _init_menu_actions + # We're testing the structure, not the actual Qt implementation + + mock_menubar_instance = MagicMock() + mock_menubar_instance.actions.return_value = [] # No existing Tools menu + mock_menubar_instance.addMenu = MagicMock() + + # Simulate the logic from _init_menu_actions + tools_menu = None + for action in mock_menubar_instance.actions(): + if action.text() == "Tools": + tools_menu = action.menu() + break + + if not tools_menu: + tools_menu = mock_menubar_instance.addMenu("Tools") + + # Verify addMenu was called + assert mock_menubar_instance.addMenu.called + + @patch("tidal_dl_ng.gui.QtWidgets.QMenuBar") + @patch("tidal_dl_ng.gui.QtGui.QAction") + def test_tools_menu_found_if_exists(self, mock_action, mock_menubar): + """Test that existing Tools menu is found and reused.""" + mock_menubar_instance = MagicMock() + + # Create mock existing Tools menu + mock_tools_action = MagicMock() + mock_tools_action.text.return_value = "Tools" + mock_tools_menu = MagicMock() + mock_tools_action.menu.return_value = mock_tools_menu + + mock_menubar_instance.actions.return_value = [mock_tools_action] + mock_menubar_instance.addMenu = MagicMock() + + # Simulate the logic + tools_menu = None + for action in mock_menubar_instance.actions(): + if action.text() == "Tools": + tools_menu = action.menu() + break + + # Should have found existing menu + assert tools_menu is mock_tools_menu + # Should NOT have called addMenu + assert not mock_menubar_instance.addMenu.called + + +class TestDuplicatePreventionAction: + """Test the duplicate prevention menu action.""" + + def test_action_is_checkable(self): + """Test that the duplicate prevention action is checkable.""" + # This verifies the action configuration + # In the actual code: self.a_toggle_duplicate_prevention.setCheckable(True) + + mock_action = MagicMock() + mock_action.setCheckable = MagicMock() + + # Simulate action creation + mock_action.setCheckable(True) + + mock_action.setCheckable.assert_called_once_with(True) + + def test_action_initial_state_from_settings(self): + """Test that action initial state reflects settings.""" + mock_history_service = MagicMock() + mock_history_service.get_settings.return_value = {"preventDuplicates": True} + + # Simulate getting initial state + is_preventing = mock_history_service.get_settings().get("preventDuplicates", True) + + assert is_preventing is True + + def test_action_connected_to_handler(self): + """Test that action triggered signal connects to handler.""" + mock_action = MagicMock() + mock_handler = MagicMock() + + # Simulate connection + mock_action.triggered.connect(mock_handler) + + mock_action.triggered.connect.assert_called_once_with(mock_handler) + + +class TestToggleDuplicatePreventionHandler: + """Test the on_toggle_duplicate_prevention handler method.""" + + def test_handler_updates_settings_when_enabled(self): + """Test that handler updates settings to True when enabled.""" + mock_history_service = MagicMock() + + # Simulate the handler logic + enabled = True + mock_history_service.update_settings(preventDuplicates=enabled) + + mock_history_service.update_settings.assert_called_once_with(preventDuplicates=True) + + def test_handler_updates_settings_when_disabled(self): + """Test that handler updates settings to False when disabled.""" + mock_history_service = MagicMock() + + # Simulate the handler logic + enabled = False + mock_history_service.update_settings(preventDuplicates=enabled) + + mock_history_service.update_settings.assert_called_once_with(preventDuplicates=False) + + def test_handler_emits_status_message_enabled(self): + """Test that handler emits status message when enabled.""" + + # Simulate the handler logic + enabled = True + status_msg = "enabled" if enabled else "disabled" + + assert status_msg == "enabled" + + def test_handler_emits_status_message_disabled(self): + """Test that handler emits status message when disabled.""" + enabled = False + status_msg = "enabled" if enabled else "disabled" + + assert status_msg == "disabled" + + def test_handler_logs_change(self): + """Test that handler logs the settings change.""" + from unittest.mock import MagicMock + + mock_logger = MagicMock() + + # Simulate handler logging + enabled = True + status_msg = "enabled" if enabled else "disabled" + mock_logger.info(f"Duplicate download prevention {status_msg}") + + mock_logger.info.assert_called_once() + call_args = str(mock_logger.info.call_args) + assert "enabled" in call_args + + +class TestSettingsPersistence: + """Test that GUI settings changes persist.""" + + def test_toggle_persists_to_json(self): + """Test that toggling the option persists to the JSON file.""" + mock_history_service = MagicMock() + + # Simulate toggle + mock_history_service.update_settings(preventDuplicates=False) + + # Verify update_settings was called (which triggers save) + mock_history_service.update_settings.assert_called_with(preventDuplicates=False) + + def test_settings_restored_on_gui_restart(self): + """Test that settings are restored when GUI restarts.""" + mock_history_service = MagicMock() + mock_history_service.get_settings.return_value = {"preventDuplicates": False} + + # Simulate GUI init reading settings + is_preventing = mock_history_service.get_settings().get("preventDuplicates", True) + + assert is_preventing is False + + +class TestStatusBarMessages: + """Test status bar message display.""" + + def test_status_message_format_enabled(self): + """Test status message format when enabled.""" + from tidal_dl_ng.model.gui_data import StatusbarMessage + + enabled = True + status_msg = "enabled" if enabled else "disabled" + message = StatusbarMessage(message=f"Duplicate prevention {status_msg}.", timeout=2500) + + assert "Duplicate prevention enabled" in message.message + assert message.timeout == 2500 + + def test_status_message_format_disabled(self): + """Test status message format when disabled.""" + from tidal_dl_ng.model.gui_data import StatusbarMessage + + enabled = False + status_msg = "enabled" if enabled else "disabled" + message = StatusbarMessage(message=f"Duplicate prevention {status_msg}.", timeout=2500) + + assert "Duplicate prevention disabled" in message.message + assert message.timeout == 2500 + + +class TestViewHistoryAction: + """Test the View Download History action.""" + + def test_view_history_action_exists(self): + """Test that View History action is created.""" + mock_action = MagicMock() + mock_handler = MagicMock() + + # Simulate action creation + mock_action.setText("View Download History...") + mock_action.triggered.connect(mock_handler) + + assert "View Download History" in str(mock_action.setText.call_args) + + def test_view_history_opens_dialog(self): + """Test that View History action opens the dialog.""" + mock_dialog_class = MagicMock() + mock_history_service = MagicMock() + + # Simulate dialog opening + mock_dialog_class(history_service=mock_history_service, parent=None) + + mock_dialog_class.assert_called_once() + + +class TestMenuSeparator: + """Test that separator is added between menu items.""" + + def test_separator_added_between_items(self): + """Test that a separator is added between View History and Prevent Duplicates.""" + mock_menu = MagicMock() + + # Simulate adding items + mock_menu.addAction(MagicMock()) # View History + mock_menu.addSeparator() + mock_menu.addAction(MagicMock()) # Prevent Duplicates + + # Verify separator was called + mock_menu.addSeparator.assert_called_once() + + +class TestIntegrationWithDownloadQueue: + """Test integration with download queue.""" + + def test_prevention_affects_queue_downloads(self): + """Test that prevention setting affects downloads from queue.""" + mock_history_service = MagicMock() + mock_history_service.should_skip_download.return_value = True + + # Simulate queue download check + track_id = "123456" + should_skip = mock_history_service.should_skip_download(track_id) + + assert should_skip is True + + def test_prevention_disabled_allows_queue_downloads(self): + """Test that disabling prevention allows queue downloads.""" + mock_history_service = MagicMock() + mock_history_service.should_skip_download.return_value = False + + # Simulate queue download check + track_id = "123456" + should_skip = mock_history_service.should_skip_download(track_id) + + assert should_skip is False + + +class TestActionStateSync: + """Test that action state stays synced with settings.""" + + def test_action_checked_when_prevention_enabled(self): + """Test that action appears checked when prevention is enabled.""" + mock_action = MagicMock() + mock_history_service = MagicMock() + mock_history_service.get_settings.return_value = {"preventDuplicates": True} + + # Simulate setting initial state + is_preventing = mock_history_service.get_settings().get("preventDuplicates", True) + mock_action.setChecked(is_preventing) + + mock_action.setChecked.assert_called_once_with(True) + + def test_action_unchecked_when_prevention_disabled(self): + """Test that action appears unchecked when prevention is disabled.""" + mock_action = MagicMock() + mock_history_service = MagicMock() + mock_history_service.get_settings.return_value = {"preventDuplicates": False} + + # Simulate setting initial state + is_preventing = mock_history_service.get_settings().get("preventDuplicates", True) + mock_action.setChecked(is_preventing) + + mock_action.setChecked.assert_called_once_with(False) + + +class TestErrorHandling: + """Test error handling in GUI actions.""" + + def test_handler_continues_on_settings_error(self): + """Test that handler continues gracefully if settings update fails.""" + mock_history_service = MagicMock() + mock_history_service.update_settings.side_effect = Exception("Settings error") + + # Simulate handler with try/except (if implemented) + try: + mock_history_service.update_settings(preventDuplicates=True) + success = True + except Exception: + success = False + + # In current implementation, error would propagate + # This test documents current behavior + assert success is False + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_history_service.py b/tests/test_history_service.py new file mode 100644 index 0000000..0ab4d28 --- /dev/null +++ b/tests/test_history_service.py @@ -0,0 +1,550 @@ +""" +test_history_service.py + +Test suite for the HistoryService class and download history functionality. + +Tests cover: +- History service initialization +- Track addition and removal +- Duplicate prevention logic +- Settings management +- JSON persistence and corruption recovery +- Import/Export functionality +- Statistics calculation +""" + +import json +from datetime import UTC, datetime +from pathlib import Path +from unittest.mock import patch + +import pytest + +from tidal_dl_ng.history import HistoryService + + +@pytest.fixture +def temp_history_file(tmp_path: Path) -> Path: + """Create a temporary directory for history file storage. + + Args: + tmp_path: pytest fixture providing temporary directory. + + Returns: + Path to temporary directory. + """ + return tmp_path + + +@pytest.fixture +def history_service(temp_history_file: Path, monkeypatch) -> HistoryService: + """Create a HistoryService instance with a temporary file. + + Args: + temp_history_file: Temporary directory path. + monkeypatch: pytest monkeypatch fixture. + + Returns: + Configured HistoryService instance. + """ + # Mock the config path to use temp directory + mock_config_path = temp_history_file / "config" + mock_config_path.mkdir(parents=True, exist_ok=True) + + monkeypatch.setattr("tidal_dl_ng.history.path_config_base", lambda: str(mock_config_path)) + + # Reset singleton instance + if hasattr(HistoryService, "_instances"): + HistoryService._instances = {} + + return HistoryService() + + +class TestHistoryServiceInitialization: + """Test history service initialization and file creation.""" + + def test_service_init_creates_empty_history(self, history_service: HistoryService): + """Test that initialization creates an empty history.""" + assert history_service.history_data == {} + assert history_service.settings_data == {"preventDuplicates": True} + assert history_service.file_path.exists() + + def test_service_init_creates_json_file(self, history_service: HistoryService): + """Test that JSON file is created with correct structure.""" + with open(history_service.file_path, encoding="utf-8") as f: + data = json.load(f) + + assert "_schema_version" in data + assert data["_schema_version"] == 1 + assert "settings" in data + assert "tracks" in data + assert data["tracks"] == {} + + def test_service_singleton_pattern(self, history_service: HistoryService): + """Test that HistoryService follows singleton pattern.""" + service2 = HistoryService() + assert service2 is history_service + + +class TestTrackOperations: + """Test track addition, removal, and queries.""" + + def test_add_track_to_history(self, history_service: HistoryService): + """Test adding a track to history.""" + track_id = "12345678" + history_service.add_track_to_history( + track_id=track_id, source_type="playlist", source_id="pl-123", source_name="My Playlist" + ) + + assert track_id in history_service.history_data + assert history_service.history_data[track_id]["sourceType"] == "playlist" + assert history_service.history_data[track_id]["sourceId"] == "pl-123" + assert history_service.history_data[track_id]["sourceName"] == "My Playlist" + assert "downloadDate" in history_service.history_data[track_id] + + def test_add_track_persists_to_file(self, history_service: HistoryService): + """Test that adding a track persists to the JSON file.""" + track_id = "87654321" + history_service.add_track_to_history(track_id=track_id, source_type="album") + + # Reload from file + with open(history_service.file_path, encoding="utf-8") as f: + data = json.load(f) + + assert track_id in data["tracks"] + + def test_is_downloaded_returns_true_for_existing_track(self, history_service: HistoryService): + """Test that is_downloaded returns True for existing tracks.""" + track_id = "11111111" + history_service.add_track_to_history(track_id=track_id) + + assert history_service.is_downloaded(track_id) is True + + def test_is_downloaded_returns_false_for_missing_track(self, history_service: HistoryService): + """Test that is_downloaded returns False for missing tracks.""" + assert history_service.is_downloaded("99999999") is False + + def test_remove_track_from_history(self, history_service: HistoryService): + """Test removing a track from history.""" + track_id = "22222222" + history_service.add_track_to_history(track_id=track_id) + + result = history_service.remove_track_from_history(track_id) + + assert result is True + assert track_id not in history_service.history_data + + def test_remove_nonexistent_track_returns_false(self, history_service: HistoryService): + """Test that removing a nonexistent track returns False.""" + result = history_service.remove_track_from_history("nonexistent") + assert result is False + + def test_get_track_info_returns_correct_data(self, history_service: HistoryService): + """Test getting track info returns complete data.""" + track_id = "33333333" + history_service.add_track_to_history( + track_id=track_id, source_type="mix", source_id="mx-456", source_name="Dance Mix" + ) + + info = history_service.get_track_info(track_id) + + assert info is not None + assert info["sourceType"] == "mix" + assert info["sourceId"] == "mx-456" + assert info["sourceName"] == "Dance Mix" + + def test_get_track_info_returns_none_for_missing(self, history_service: HistoryService): + """Test that get_track_info returns None for missing tracks.""" + info = history_service.get_track_info("missing") + assert info is None + + +class TestDuplicatePrevention: + """Test duplicate download prevention logic.""" + + def test_should_skip_download_when_enabled_and_downloaded(self, history_service: HistoryService): + """Test that should_skip_download returns True when prevention is enabled and track exists.""" + track_id = "44444444" + history_service.add_track_to_history(track_id=track_id) + history_service.update_settings(preventDuplicates=True) + + assert history_service.should_skip_download(track_id) is True + + def test_should_skip_download_when_disabled_and_downloaded(self, history_service: HistoryService): + """Test that should_skip_download returns False when prevention is disabled.""" + track_id = "55555555" + history_service.add_track_to_history(track_id=track_id) + history_service.update_settings(preventDuplicates=False) + + assert history_service.should_skip_download(track_id) is False + + def test_should_skip_download_when_enabled_and_not_downloaded(self, history_service: HistoryService): + """Test that should_skip_download returns False for new tracks.""" + history_service.update_settings(preventDuplicates=True) + + assert history_service.should_skip_download("new_track") is False + + def test_prevent_duplicates_default_enabled(self, history_service: HistoryService): + """Test that duplicate prevention is enabled by default.""" + settings = history_service.get_settings() + assert settings["preventDuplicates"] is True + + +class TestSettingsManagement: + """Test settings management functionality.""" + + def test_get_settings_returns_copy(self, history_service: HistoryService): + """Test that get_settings returns a copy, not the original.""" + settings = history_service.get_settings() + settings["preventDuplicates"] = False + + # Original should be unchanged + assert history_service.settings_data["preventDuplicates"] is True + + def test_update_settings_changes_value(self, history_service: HistoryService): + """Test that update_settings changes the setting value.""" + history_service.update_settings(preventDuplicates=False) + + assert history_service.settings_data["preventDuplicates"] is False + + def test_update_settings_persists_to_file(self, history_service: HistoryService): + """Test that settings updates are persisted to the file.""" + history_service.update_settings(preventDuplicates=False) + + with open(history_service.file_path, encoding="utf-8") as f: + data = json.load(f) + + assert data["settings"]["preventDuplicates"] is False + + def test_update_settings_converts_to_bool(self, history_service: HistoryService): + """Test that update_settings converts values to boolean.""" + history_service.update_settings(preventDuplicates="true") + assert history_service.settings_data["preventDuplicates"] is True + + history_service.update_settings(preventDuplicates=0) + assert history_service.settings_data["preventDuplicates"] is False + + +class TestHistoryBySource: + """Test source-centric history view.""" + + def test_get_history_by_source_groups_correctly(self, history_service: HistoryService): + """Test that tracks are grouped by source correctly.""" + # Add tracks from same playlist + history_service.add_track_to_history("t1", "playlist", "pl-1", "Playlist 1") + history_service.add_track_to_history("t2", "playlist", "pl-1", "Playlist 1") + # Add track from different source + history_service.add_track_to_history("t3", "album", "al-1", "Album 1") + + grouped = history_service.get_history_by_source() + + assert "playlist_pl-1" in grouped + assert len(grouped["playlist_pl-1"]) == 2 + assert "album_al-1" in grouped + assert len(grouped["album_al-1"]) == 1 + + def test_get_history_by_source_manual_tracks(self, history_service: HistoryService): + """Test that manual tracks are grouped separately.""" + history_service.add_track_to_history("t1", "manual", None, None) + history_service.add_track_to_history("t2", "manual", None, None) + + grouped = history_service.get_history_by_source() + + assert "manual_manual" in grouped + assert len(grouped["manual_manual"]) == 2 + + +class TestStatistics: + """Test statistics calculation.""" + + def test_get_statistics_counts_total_tracks(self, history_service: HistoryService): + """Test that statistics include total track count.""" + history_service.add_track_to_history("t1") + history_service.add_track_to_history("t2") + history_service.add_track_to_history("t3") + + stats = history_service.get_statistics() + + assert stats["total_tracks"] == 3 + + def test_get_statistics_counts_by_source_type(self, history_service: HistoryService): + """Test that statistics count tracks by source type.""" + history_service.add_track_to_history("t1", "playlist") + history_service.add_track_to_history("t2", "playlist") + history_service.add_track_to_history("t3", "album") + history_service.add_track_to_history("t4", "mix") + + stats = history_service.get_statistics() + + assert stats["by_source_type"]["playlist"] == 2 + assert stats["by_source_type"]["album"] == 1 + assert stats["by_source_type"]["mix"] == 1 + + def test_get_statistics_finds_oldest_newest(self, history_service: HistoryService): + """Test that statistics include oldest and newest download dates.""" + history_service.add_track_to_history("t1") + history_service.add_track_to_history("t2") + + stats = history_service.get_statistics() + + assert stats["oldest_download"] is not None + assert stats["newest_download"] is not None + + def test_get_statistics_empty_history(self, history_service: HistoryService): + """Test statistics with empty history.""" + stats = history_service.get_statistics() + + assert stats["total_tracks"] == 0 + assert stats["by_source_type"] == {} + assert stats["oldest_download"] is None + assert stats["newest_download"] is None + + +class TestJSONPersistence: + """Test JSON file persistence and corruption recovery.""" + + def test_corrupted_json_creates_backup(self, history_service: HistoryService): + """Test that corrupted JSON files are backed up.""" + # Write corrupted JSON + with open(history_service.file_path, "w", encoding="utf-8") as f: + f.write("{ invalid json }") + + # Force reload + history_service._load_history() + + # Check backup was created + backup_files = list(history_service.file_path.parent.glob("*.bak*")) + assert len(backup_files) > 0 + + def test_corrupted_json_starts_fresh(self, history_service: HistoryService): + """Test that corrupted JSON results in fresh empty history.""" + # Add a track first + history_service.add_track_to_history("t1") + + # Corrupt the file + with open(history_service.file_path, "w", encoding="utf-8") as f: + f.write("corrupted") + + # Force reload + history_service._load_history() + + # Should have fresh empty history + assert history_service.history_data == {} + + def test_atomic_write_on_failure(self, history_service: HistoryService): + """Test that failed writes don't corrupt existing file.""" + from contextlib import suppress + + # Add initial data + history_service.add_track_to_history("t1") + + # Mock a write failure + with patch("builtins.open", side_effect=OSError("Disk full")), suppress(OSError): + history_service.add_track_to_history("t2") + + # Original file should still be intact + with open(history_service.file_path, encoding="utf-8") as f: + data = json.load(f) + + assert "t1" in data["tracks"] + + def test_legacy_format_migration(self, history_service: HistoryService): + """Test that legacy JSON format (without tracks section) is migrated.""" + # Write legacy format (tracks at root level) + legacy_data = { + "_schema_version": 1, + "12345": { + "sourceType": "playlist", + "sourceId": "pl-1", + "sourceName": "Test", + "downloadDate": datetime.now(UTC).isoformat(), + }, + } + + with open(history_service.file_path, "w", encoding="utf-8") as f: + json.dump(legacy_data, f) + + # Force reload + history_service._load_history() + + # Should have migrated the track + assert "12345" in history_service.history_data + + +class TestImportExport: + """Test import and export functionality.""" + + def test_export_history_creates_valid_json(self, history_service: HistoryService, tmp_path: Path): + """Test that export creates a valid JSON file.""" + history_service.add_track_to_history("t1", "playlist", "pl-1", "Test") + + export_path = tmp_path / "export.json" + success, message = history_service.export_history(str(export_path)) + + assert success is True + assert export_path.exists() + + with open(export_path, encoding="utf-8") as f: + data = json.load(f) + + assert "_schema_version" in data + assert "tracks" in data + assert "t1" in data["tracks"] + + def test_import_history_merge_mode(self, history_service: HistoryService, tmp_path: Path): + """Test importing history in merge mode.""" + # Add existing track + history_service.add_track_to_history("existing", "album") + + # Create import file + import_data = { + "tracks": { + "imported": { + "sourceType": "playlist", + "sourceId": "pl-1", + "sourceName": "Imported", + "downloadDate": datetime.now(UTC).isoformat(), + } + } + } + + import_path = tmp_path / "import.json" + with open(import_path, "w", encoding="utf-8") as f: + json.dump(import_data, f) + + success, message = history_service.import_history(str(import_path), merge=True) + + assert success is True + assert "existing" in history_service.history_data + assert "imported" in history_service.history_data + + def test_import_history_replace_mode(self, history_service: HistoryService, tmp_path: Path): + """Test importing history in replace mode.""" + # Add existing track + history_service.add_track_to_history("existing", "album") + + # Create import file + import_data = { + "tracks": { + "imported": { + "sourceType": "playlist", + "sourceId": "pl-1", + "sourceName": "Imported", + "downloadDate": datetime.now(UTC).isoformat(), + } + } + } + + import_path = tmp_path / "import.json" + with open(import_path, "w", encoding="utf-8") as f: + json.dump(import_data, f) + + success, message = history_service.import_history(str(import_path), merge=False) + + assert success is True + assert "existing" not in history_service.history_data + assert "imported" in history_service.history_data + + def test_import_invalid_json_fails(self, history_service: HistoryService, tmp_path: Path): + """Test that importing invalid JSON fails gracefully.""" + import_path = tmp_path / "invalid.json" + with open(import_path, "w", encoding="utf-8") as f: + f.write("{ invalid }") + + success, message = history_service.import_history(str(import_path)) + + assert success is False + assert "Invalid JSON" in message + + def test_import_settings_from_file(self, history_service: HistoryService, tmp_path: Path): + """Test that settings are imported along with tracks.""" + import_data = { + "settings": {"preventDuplicates": False}, + "tracks": {"t1": {"sourceType": "manual", "downloadDate": datetime.now(UTC).isoformat()}}, + } + + import_path = tmp_path / "import.json" + with open(import_path, "w", encoding="utf-8") as f: + json.dump(import_data, f) + + history_service.import_history(str(import_path)) + + assert history_service.settings_data["preventDuplicates"] is False + + +class TestClearHistory: + """Test history clearing functionality.""" + + def test_clear_history_removes_all_tracks(self, history_service: HistoryService): + """Test that clear_history removes all tracks.""" + history_service.add_track_to_history("t1") + history_service.add_track_to_history("t2") + history_service.add_track_to_history("t3") + + history_service.clear_history() + + assert len(history_service.history_data) == 0 + + def test_clear_history_persists_to_file(self, history_service: HistoryService): + """Test that clearing history persists to the file.""" + history_service.add_track_to_history("t1") + history_service.clear_history() + + with open(history_service.file_path, encoding="utf-8") as f: + data = json.load(f) + + assert len(data["tracks"]) == 0 + + +class TestThreadSafety: + """Test thread safety of history operations.""" + + def test_concurrent_track_additions(self, history_service: HistoryService): + """Test that concurrent track additions are thread-safe.""" + import threading + + def add_tracks(start_id: int): + for i in range(10): + history_service.add_track_to_history(f"t{start_id + i}") + + threads = [threading.Thread(target=add_tracks, args=(i * 10,)) for i in range(5)] + + for t in threads: + t.start() + + for t in threads: + t.join() + + # Should have 50 tracks total + assert len(history_service.history_data) == 50 + + def test_concurrent_read_write(self, history_service: HistoryService): + """Test concurrent read and write operations.""" + import threading + + history_service.add_track_to_history("t1") + + results = [] + + def reader(): + for _ in range(100): + results.append(history_service.is_downloaded("t1")) + + def writer(): + for i in range(10): + history_service.add_track_to_history(f"new_{i}") + + threads = [threading.Thread(target=reader), threading.Thread(target=writer)] + + for t in threads: + t.start() + + for t in threads: + t.join() + + # All reads should have succeeded + assert all(results) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_logger_configuration.py b/tests/test_logger_configuration.py new file mode 100644 index 0000000..8cfb2a9 --- /dev/null +++ b/tests/test_logger_configuration.py @@ -0,0 +1,277 @@ +""" +test_logger_configuration.py + +Test suite for logger configuration with colored output. + +Tests cover: +- INFO level color configuration +- coloredlogs integration +- Log message formatting +""" + +import logging +from unittest.mock import MagicMock + +import coloredlogs +import pytest + + +class TestLoggerColorConfiguration: + """Test logger color configuration for INFO messages.""" + + def test_info_level_has_green_color(self): + """Test that INFO level is configured with green color.""" + # Create level styles like in logger.py + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + assert "info" in level_styles + assert level_styles["info"] == {"color": "green"} + + def test_default_info_level_has_no_color(self): + """Test that default INFO level has no color (to verify our fix).""" + default_styles = coloredlogs.DEFAULT_LEVEL_STYLES + + # Default should have no color for INFO + assert default_styles.get("info", {}) == {} + + def test_debug_level_keeps_green_color(self): + """Test that DEBUG level keeps its default green color.""" + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + assert level_styles["debug"] == {"color": "green"} + + def test_level_styles_includes_all_levels(self): + """Test that custom level styles include all necessary levels.""" + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + # Verify all important levels are present + assert "debug" in level_styles + assert "info" in level_styles + assert "warning" in level_styles + assert "error" in level_styles + + +class TestColoredFormatter: + """Test ColoredFormatter configuration.""" + + def test_formatter_uses_custom_level_styles(self): + """Test that formatter is created with custom level styles.""" + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + formatter = coloredlogs.ColoredFormatter(fmt="> %(message)s", level_styles=level_styles) + + # Formatter should exist + assert formatter is not None + + def test_formatter_format_string(self): + """Test that formatter uses correct format string.""" + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + log_fmt = "> %(message)s" + formatter = coloredlogs.ColoredFormatter(fmt=log_fmt, level_styles=level_styles) + + # Format string should be set + assert formatter._fmt == log_fmt or hasattr(formatter, "_style") + + +class TestLoggerHandlers: + """Test logger handler configuration.""" + + def test_gui_logger_has_qt_handler(self): + """Test that GUI logger would be configured with QtHandler.""" + # This is a structural test - verifying the configuration pattern + from tidal_dl_ng.logger import QtHandler + + handler = QtHandler() + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + formatter = coloredlogs.ColoredFormatter(fmt="> %(message)s", level_styles=level_styles) + + handler.setFormatter(formatter) + + # Handler should have formatter + assert handler.formatter is not None + + def test_cli_logger_has_stream_handler(self): + """Test that CLI logger would be configured with StreamHandler.""" + handler = logging.StreamHandler() + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + formatter = coloredlogs.ColoredFormatter(fmt="> %(message)s", level_styles=level_styles) + + handler.setFormatter(formatter) + + # Handler should have formatter + assert handler.formatter is not None + + +class TestLogMessageFormatting: + """Test log message formatting.""" + + def test_info_message_formatted_with_arrow(self): + """Test that INFO messages are formatted with arrow prefix.""" + log_fmt = "> %(message)s" + + # Create a test record + record = logging.LogRecord( + name="test", level=logging.INFO, pathname="", lineno=0, msg="Test message", args=(), exc_info=None + ) + + formatter = logging.Formatter(log_fmt) + formatted = formatter.format(record) + + assert formatted == "> Test message" + + def test_skip_message_format(self): + """Test that skip messages follow the expected format.""" + message_template = "Skipped item '{item}' (already in history)." + item_name = "Test Track" + + message = message_template.format(item=item_name) + + assert "Skipped item" in message + assert item_name in message + assert "already in history" in message + + def test_downloaded_message_format(self): + """Test that downloaded messages follow the expected format.""" + message_template = "Downloaded item '{item}'." + item_name = "Test Track" + + message = message_template.format(item=item_name) + + assert "Downloaded item" in message + assert item_name in message + + +class TestColoredlogsIntegration: + """Test coloredlogs integration.""" + + def test_coloredlogs_available(self): + """Test that coloredlogs module is available.""" + assert coloredlogs is not None + assert hasattr(coloredlogs, "ColoredFormatter") + assert hasattr(coloredlogs, "DEFAULT_LEVEL_STYLES") + + def test_color_codes_in_formatter_output(self): + """Test that formatter produces ANSI color codes.""" + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + formatter = coloredlogs.ColoredFormatter(fmt="%(message)s", level_styles=level_styles) + + record = logging.LogRecord( + name="test", level=logging.INFO, pathname="", lineno=0, msg="Test", args=(), exc_info=None + ) + + formatted = formatter.format(record) + + # ANSI codes should be present (or the message itself if colors disabled) + assert formatted is not None + assert len(formatted) > 0 + + +class TestLoggerMessageTypes: + """Test different logger message types.""" + + def test_info_messages_use_info_level(self): + """Test that informational messages use INFO level.""" + logger = logging.getLogger("test") + handler = logging.Handler() + handler.emit = MagicMock() + logger.addHandler(handler) + logger.setLevel(logging.INFO) + + logger.info("Test info message") + + # Handler should have been called + assert handler.emit.called + record = handler.emit.call_args[0][0] + assert record.levelno == logging.INFO + + def test_debug_messages_use_debug_level(self): + """Test that debug messages use DEBUG level.""" + logger = logging.getLogger("test_debug") + handler = logging.Handler() + handler.emit = MagicMock() + logger.addHandler(handler) + logger.setLevel(logging.DEBUG) + + logger.debug("Test debug message") + + # Handler should have been called + assert handler.emit.called + record = handler.emit.call_args[0][0] + assert record.levelno == logging.DEBUG + + +class TestXStreamIntegration: + """Test XStream integration with colored output.""" + + def test_xstream_writes_formatted_messages(self): + """Test that XStream can write formatted log messages.""" + from tidal_dl_ng.logger import XStream + + stream = XStream.stdout() + + # Should have write method + assert hasattr(stream, "write") + assert callable(stream.write) + + def test_xstream_emits_signal(self): + """Test that XStream emits signal when writing.""" + from tidal_dl_ng.logger import XStream + + stream = XStream() + + # Should have messageWritten signal + assert hasattr(stream, "messageWritten") + + +class TestLoggerLevels: + """Test logger level configuration.""" + + def test_logger_set_to_debug_level(self): + """Test that loggers are set to DEBUG level.""" + logger = logging.getLogger("test_levels") + logger.setLevel(logging.DEBUG) + + assert logger.level == logging.DEBUG + + def test_debug_level_allows_info_messages(self): + """Test that DEBUG level allows INFO messages.""" + logger = logging.getLogger("test_info_allowed") + logger.setLevel(logging.DEBUG) + + # INFO is less severe than DEBUG, so it should be allowed + assert logger.isEnabledFor(logging.INFO) + + +class TestConsistentStyling: + """Test consistent styling across loggers.""" + + def test_both_loggers_use_same_level_styles(self): + """Test that GUI and CLI loggers use same level styles.""" + # Create shared level styles + level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy() + level_styles["info"] = {"color": "green"} + + # Both formatters should use the same styles + gui_formatter = coloredlogs.ColoredFormatter(fmt="> %(message)s", level_styles=level_styles) + + cli_formatter = coloredlogs.ColoredFormatter(fmt="> %(message)s", level_styles=level_styles) + + # Both should exist and be configured + assert gui_formatter is not None + assert cli_formatter is not None + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) |