path: root/tests/test_history_service.py

"""
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"])