Warry
f413240b51
- Improved playlist item fetching by adding robust pagination and low-level request hooks for better testing and flexibility. - Refactored track ID extraction logic into reusable helpers for clarity. - Improved initialization for dialog components to avoid circular dependencies in the GUI. - Added new unit tests to validate playlist operations, including track addition/removal and API integration. - Updated metadata functions to simplify dictionary access using `.get` for better readability. - Ensured PySide6 system dependencies are installed in GitHub Actions to address CI compatibility issues. - Streamlined UI package initialization to improve modularity and robustness of imports.
569 lines
20 KiB
Python
569 lines
20 KiB
Python
"""Tests for Playlist Membership Manager.
|
|
|
|
Comprehensive test suite for the playlist management system including:
|
|
- Cache thread-safety and operations
|
|
- Worker thread loading and API integration
|
|
- Dialog UI state management
|
|
- Transaction rollback on errors
|
|
"""
|
|
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from PySide6 import QtCore, QtWidgets
|
|
from tidalapi import Session, Track
|
|
|
|
from tidal_dl_ng.gui.playlist_membership import (
|
|
PlaylistCellState,
|
|
PlaylistColumnDelegate,
|
|
PlaylistContextLoader,
|
|
ThreadSafePlaylistCache,
|
|
)
|
|
from tidal_dl_ng.ui.dialog_playlist_manager import PlaylistManagerDialog
|
|
|
|
|
|
class TestThreadSafePlaylistCache(unittest.TestCase):
|
|
"""Tests for ThreadSafePlaylistCache.
|
|
|
|
Verifies thread-safety, O(1) performance, and data integrity.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up test fixtures."""
|
|
self.cache = ThreadSafePlaylistCache()
|
|
|
|
def test_add_track_to_playlist(self) -> None:
|
|
"""Test adding a track to a playlist."""
|
|
self.cache.add_track_to_playlist("track_1", "playlist_1")
|
|
playlists = self.cache.get_playlists_for_track("track_1")
|
|
self.assertIn("playlist_1", playlists)
|
|
|
|
def test_remove_track_from_playlist(self) -> None:
|
|
"""Test removing a track from a playlist."""
|
|
self.cache.add_track_to_playlist("track_1", "playlist_1")
|
|
self.cache.remove_track_from_playlist("track_1", "playlist_1")
|
|
playlists = self.cache.get_playlists_for_track("track_1")
|
|
self.assertNotIn("playlist_1", playlists)
|
|
|
|
def test_is_track_in_playlist(self) -> None:
|
|
"""Test checking if track is in playlist."""
|
|
self.cache.add_track_to_playlist("track_1", "playlist_1")
|
|
self.assertTrue(self.cache.is_track_in_playlist("track_1", "playlist_1"))
|
|
self.assertFalse(self.cache.is_track_in_playlist("track_1", "playlist_2"))
|
|
|
|
def test_get_nonexistent_track(self) -> None:
|
|
"""Test getting playlists for nonexistent track returns empty set."""
|
|
playlists = self.cache.get_playlists_for_track("nonexistent")
|
|
self.assertEqual(playlists, set())
|
|
|
|
def test_multiple_playlists_per_track(self) -> None:
|
|
"""Test track can be in multiple playlists."""
|
|
self.cache.add_track_to_playlist("track_1", "playlist_1")
|
|
self.cache.add_track_to_playlist("track_1", "playlist_2")
|
|
self.cache.add_track_to_playlist("track_1", "playlist_3")
|
|
|
|
playlists = self.cache.get_playlists_for_track("track_1")
|
|
self.assertEqual(len(playlists), 3)
|
|
self.assertIn("playlist_1", playlists)
|
|
self.assertIn("playlist_2", playlists)
|
|
self.assertIn("playlist_3", playlists)
|
|
|
|
def test_concurrent_reads_no_deadlock(self) -> None:
|
|
"""Test concurrent reads don't cause deadlock."""
|
|
self.cache.add_track_to_playlist("track_1", "playlist_1")
|
|
|
|
results = []
|
|
|
|
def read_cache() -> None:
|
|
for _ in range(100):
|
|
result = self.cache.get_playlists_for_track("track_1")
|
|
results.append(result)
|
|
|
|
threads = [threading.Thread(target=read_cache) for _ in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# All reads should succeed
|
|
self.assertEqual(len(results), 500)
|
|
|
|
def test_concurrent_writes_thread_safe(self) -> None:
|
|
"""Test concurrent writes maintain consistency."""
|
|
|
|
def write_cache(track_id: str, playlist_ids: list[str]) -> None:
|
|
for pid in playlist_ids:
|
|
self.cache.add_track_to_playlist(track_id, pid)
|
|
|
|
threads = []
|
|
for i in range(10):
|
|
t = threading.Thread(target=write_cache, args=(f"track_{i}", [f"p_{i}_{j}" for j in range(5)]))
|
|
threads.append(t)
|
|
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# Verify all writes succeeded
|
|
total_tracks = sum(len(self.cache.get_playlists_for_track(f"track_{i}")) for i in range(10))
|
|
self.assertEqual(total_tracks, 50)
|
|
|
|
def test_update_from_dict(self) -> None:
|
|
"""Test batch update from dictionary."""
|
|
data = {
|
|
"track_1": {"playlist_1", "playlist_2"},
|
|
"track_2": {"playlist_2", "playlist_3"},
|
|
}
|
|
self.cache.update_from_dict(data)
|
|
|
|
self.assertEqual(self.cache.get_playlists_for_track("track_1"), {"playlist_1", "playlist_2"})
|
|
self.assertEqual(self.cache.get_playlists_for_track("track_2"), {"playlist_2", "playlist_3"})
|
|
|
|
def test_clear(self) -> None:
|
|
"""Test clearing the cache."""
|
|
self.cache.add_track_to_playlist("track_1", "playlist_1")
|
|
self.cache.clear()
|
|
self.assertEqual(self.cache.get_playlists_for_track("track_1"), set())
|
|
|
|
def test_set_and_get_playlist_metadata(self) -> None:
|
|
"""Test storing and retrieving playlist metadata."""
|
|
self.cache.set_playlist_metadata("playlist_1", "My Favorites", 42)
|
|
metadata = self.cache.get_playlist_metadata("playlist_1")
|
|
|
|
self.assertIsNotNone(metadata)
|
|
self.assertEqual(metadata["name"], "My Favorites")
|
|
self.assertEqual(metadata["item_count"], 42)
|
|
|
|
def test_get_all_playlists(self) -> None:
|
|
"""Test getting all playlist IDs."""
|
|
self.cache.add_track_to_playlist("track_1", "playlist_1")
|
|
self.cache.add_track_to_playlist("track_2", "playlist_2")
|
|
self.cache.add_track_to_playlist("track_3", "playlist_1")
|
|
|
|
all_playlists = self.cache.get_all_playlists()
|
|
self.assertEqual(all_playlists, {"playlist_1", "playlist_2"})
|
|
|
|
|
|
class TestPlaylistContextLoader(unittest.TestCase):
|
|
"""Tests for PlaylistContextLoader worker.
|
|
|
|
Verifies API calls, pagination, and cache building.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up test fixtures."""
|
|
self.app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
|
|
|
|
@patch("tidal_dl_ng.gui.playlist_membership.Session")
|
|
def test_fetch_user_playlists_pagination(self, mock_session_class: Mock) -> None:
|
|
"""Test handling of playlist pagination."""
|
|
mock_session = Mock()
|
|
|
|
# Mock paginated responses
|
|
mock_session.request.side_effect = [
|
|
Mock(
|
|
json=lambda: {
|
|
"items": [
|
|
{"uuid": f"playlist_{i}", "title": f"Playlist {i}", "numberOfItems": 10} for i in range(50)
|
|
],
|
|
"totalNumberOfItems": 75,
|
|
}
|
|
),
|
|
Mock(
|
|
json=lambda: {
|
|
"items": [
|
|
{"uuid": f"playlist_{i}", "title": f"Playlist {i}", "numberOfItems": 10} for i in range(50, 75)
|
|
],
|
|
"totalNumberOfItems": 75,
|
|
}
|
|
),
|
|
]
|
|
|
|
loader = PlaylistContextLoader(mock_session, user_id="user_123")
|
|
playlists = loader._fetch_user_playlists()
|
|
|
|
# Should have fetched all playlists across two pages
|
|
self.assertEqual(len(playlists), 75)
|
|
|
|
@patch("tidal_dl_ng.gui.playlist_membership.Session")
|
|
def test_fetch_playlist_items_pagination(self, mock_session_class: Mock) -> None:
|
|
"""Test handling of playlist items pagination."""
|
|
mock_session = Mock()
|
|
|
|
# Mock paginated item responses
|
|
mock_session.request.side_effect = [
|
|
Mock(
|
|
json=lambda: {
|
|
"items": [{"item": {"id": f"track_{i}"}} for i in range(300)],
|
|
"totalNumberOfItems": 450,
|
|
}
|
|
),
|
|
Mock(
|
|
json=lambda: {
|
|
"items": [{"item": {"id": f"track_{i}"}} for i in range(300, 450)],
|
|
"totalNumberOfItems": 450,
|
|
}
|
|
),
|
|
]
|
|
|
|
loader = PlaylistContextLoader(mock_session, user_id="user_123")
|
|
track_ids = loader._fetch_playlist_items("playlist_1")
|
|
|
|
self.assertEqual(len(track_ids), 450)
|
|
|
|
@patch("tidal_dl_ng.gui.playlist_membership.Session")
|
|
def test_build_cache_structure(self, mock_session_class: Mock) -> None:
|
|
"""Test correct cache structure building."""
|
|
mock_session = Mock()
|
|
|
|
# Mock responses
|
|
playlists_response = Mock(
|
|
json=lambda: {
|
|
"items": [
|
|
{"uuid": "playlist_1", "title": "Favorites", "numberOfItems": 2},
|
|
{"uuid": "playlist_2", "title": "Workout", "numberOfItems": 1},
|
|
],
|
|
"totalNumberOfItems": 2,
|
|
}
|
|
)
|
|
|
|
items_responses = [
|
|
Mock(
|
|
json=lambda: {
|
|
"items": [{"item": {"id": "track_1"}}, {"item": {"id": "track_2"}}],
|
|
"totalNumberOfItems": 2,
|
|
}
|
|
),
|
|
Mock(
|
|
json=lambda: {
|
|
"items": [{"item": {"id": "track_2"}}],
|
|
"totalNumberOfItems": 1,
|
|
}
|
|
),
|
|
]
|
|
|
|
mock_session.request.side_effect = [playlists_response, *items_responses]
|
|
|
|
loader = PlaylistContextLoader(mock_session, user_id="user_123")
|
|
|
|
# Manually test cache building (since run() needs event loop)
|
|
playlists = loader._fetch_user_playlists()
|
|
cache = loader._fetch_all_playlist_contents(playlists)
|
|
|
|
# Verify cache structure
|
|
self.assertIn("track_1", cache)
|
|
self.assertIn("track_2", cache)
|
|
self.assertEqual(cache["track_1"], {"playlist_1"})
|
|
self.assertEqual(cache["track_2"], {"playlist_1", "playlist_2"})
|
|
|
|
@patch("tidal_dl_ng.gui.playlist_membership.Session")
|
|
def test_abort_request(self, mock_session_class: Mock) -> None:
|
|
"""Test aborting the loader."""
|
|
mock_session = Mock()
|
|
loader = PlaylistContextLoader(mock_session, user_id="user_123")
|
|
|
|
# Request abort
|
|
loader.request_abort()
|
|
|
|
# Verify abort flag is set
|
|
self.assertTrue(loader._abort_requested.is_set())
|
|
|
|
def test_max_workers_clamped(self) -> None:
|
|
"""Test max_workers is clamped to valid range."""
|
|
mock_session = Mock()
|
|
|
|
loader1 = PlaylistContextLoader(mock_session, user_id="user_123", max_workers=10)
|
|
self.assertLessEqual(loader1.max_workers, 5)
|
|
|
|
loader2 = PlaylistContextLoader(mock_session, user_id="user_123", max_workers=0)
|
|
self.assertEqual(loader2.max_workers, 1)
|
|
|
|
|
|
class TestPlaylistColumnDelegate(unittest.TestCase):
|
|
"""Tests for PlaylistColumnDelegate.
|
|
|
|
Verifies state rendering and transitions.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up test fixtures."""
|
|
self.app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
|
|
self.delegate = PlaylistColumnDelegate()
|
|
|
|
def test_initial_state_pending(self) -> None:
|
|
"""Test delegate starts in PENDING state."""
|
|
# Cell should render as PENDING by default
|
|
self.assertFalse(self.delegate._cache_ready)
|
|
|
|
def test_set_cache_ready_transitions_states(self) -> None:
|
|
"""Test state transitions when cache becomes ready."""
|
|
self.delegate.set_cell_state(0, PlaylistCellState.PENDING)
|
|
self.delegate.set_cell_state(1, PlaylistCellState.PENDING)
|
|
|
|
# Mark cache as ready
|
|
self.delegate.set_cache_ready(True)
|
|
|
|
# States should transition to READY
|
|
self.assertTrue(self.delegate._cache_ready)
|
|
|
|
def test_set_cell_state(self) -> None:
|
|
"""Test setting cell state."""
|
|
self.delegate.set_cell_state(5, PlaylistCellState.READY)
|
|
self.assertEqual(self.delegate._cell_states.get("5"), PlaylistCellState.READY)
|
|
|
|
self.delegate.set_cell_state(5, PlaylistCellState.ERROR)
|
|
self.assertEqual(self.delegate._cell_states.get("5"), PlaylistCellState.ERROR)
|
|
|
|
def test_state_rendering_pending(self) -> None:
|
|
"""Test rendering of PENDING state (spinner)."""
|
|
# Create mock painter and option
|
|
painter = Mock()
|
|
painter.fillRect = Mock()
|
|
painter.setPen = Mock()
|
|
painter.drawText = Mock()
|
|
|
|
option = Mock()
|
|
option.rect = Mock(adjusted=Mock(return_value=Mock(center=Mock(return_value=QtCore.QPoint(50, 50)))))
|
|
option.state = 0
|
|
|
|
index = Mock()
|
|
index.row = Mock(return_value=0)
|
|
|
|
# Set state to PENDING
|
|
self.delegate.set_cell_state(0, PlaylistCellState.PENDING)
|
|
|
|
# Paint should be called without errors
|
|
# (Actual painting verification would require QPixmap/screen rendering)
|
|
|
|
def test_button_click_event(self) -> None:
|
|
"""Test that READY state cells emit button_clicked signal."""
|
|
self.delegate.set_cell_state(0, PlaylistCellState.READY)
|
|
|
|
# Create mock event and model
|
|
event = Mock()
|
|
event.type = Mock(return_value=QtCore.QEvent.Type.MouseButtonRelease)
|
|
|
|
model = Mock()
|
|
option = Mock()
|
|
index = Mock()
|
|
index.row = Mock(return_value=0)
|
|
|
|
# Track signal emissions
|
|
signal_emitted = []
|
|
|
|
def on_button_clicked(idx) -> None:
|
|
signal_emitted.append(idx)
|
|
|
|
self.delegate.button_clicked.connect(on_button_clicked)
|
|
|
|
# Simulate click event
|
|
handled = self.delegate.editorEvent(event, model, option, index)
|
|
|
|
# Should handle the event and emit signal
|
|
self.assertTrue(handled)
|
|
|
|
|
|
class TestPlaylistManagerDialog(unittest.TestCase):
|
|
"""Tests for PlaylistManagerDialog.
|
|
|
|
Verifies UI state, transaction logic, and error handling.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up test fixtures."""
|
|
self.app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
|
|
self.threadpool = QtCore.QThreadPool()
|
|
|
|
# Create mock objects
|
|
self.mock_track = Mock(spec=Track)
|
|
self.mock_track.id = "track_uuid_1"
|
|
self.mock_track.name = "Test Track"
|
|
|
|
self.cache = ThreadSafePlaylistCache()
|
|
|
|
# Add test data to cache
|
|
self.cache.set_playlist_metadata("playlist_1", "Favorites", 42)
|
|
self.cache.set_playlist_metadata("playlist_2", "Workout Mix", 15)
|
|
self.cache.add_track_to_playlist("track_uuid_1", "playlist_1")
|
|
|
|
self.mock_session = Mock(spec=Session)
|
|
|
|
def test_dialog_initialization(self) -> None:
|
|
"""Test dialog initializes correctly."""
|
|
dialog = PlaylistManagerDialog(
|
|
track=self.mock_track,
|
|
cache=self.cache,
|
|
session=self.mock_session,
|
|
threadpool=self.threadpool,
|
|
)
|
|
|
|
self.assertIsNotNone(dialog)
|
|
self.assertEqual(dialog.track.name, "Test Track")
|
|
|
|
def test_playlists_populated_from_cache(self) -> None:
|
|
"""Test dialog populates playlists from cache."""
|
|
dialog = PlaylistManagerDialog(
|
|
track=self.mock_track,
|
|
cache=self.cache,
|
|
session=self.mock_session,
|
|
threadpool=self.threadpool,
|
|
)
|
|
|
|
# Count checkbox widgets (should be 2 playlists)
|
|
checkboxes = dialog.container_layout.count()
|
|
self.assertGreaterEqual(checkboxes, 2)
|
|
|
|
def test_checkbox_initial_state_from_cache(self) -> None:
|
|
"""Test checkboxes reflect initial cache state."""
|
|
dialog = PlaylistManagerDialog(
|
|
track=self.mock_track,
|
|
cache=self.cache,
|
|
session=self.mock_session,
|
|
threadpool=self.threadpool,
|
|
)
|
|
|
|
# Track is in playlist_1, so that checkbox should be checked
|
|
# Track is NOT in playlist_2, so that checkbox should be unchecked
|
|
original_states = dialog._original_states
|
|
|
|
self.assertTrue(original_states.get("playlist_1"))
|
|
self.assertFalse(original_states.get("playlist_2"))
|
|
|
|
@patch("tidal_dl_ng.gui.dialog_playlist_manager.Worker")
|
|
def test_checkbox_change_triggers_worker(self, mock_worker_class: Mock) -> None:
|
|
"""Test checkbox change triggers worker thread."""
|
|
PlaylistManagerDialog(
|
|
track=self.mock_track,
|
|
cache=self.cache,
|
|
session=self.mock_session,
|
|
threadpool=self.threadpool,
|
|
)
|
|
|
|
# Mock worker
|
|
mock_worker_class.return_value = Mock()
|
|
|
|
# Simulate checkbox change would trigger worker
|
|
# (Actual checkbox interaction would require Qt event simulation)
|
|
|
|
def test_api_add_track_success(self) -> None:
|
|
"""Test successful track addition."""
|
|
dialog = PlaylistManagerDialog(
|
|
track=self.mock_track,
|
|
cache=self.cache,
|
|
session=self.mock_session,
|
|
threadpool=self.threadpool,
|
|
)
|
|
|
|
# Mock playlist object for add_track_to_playlist
|
|
mock_playlist = Mock()
|
|
mock_playlist.id = "playlist_2"
|
|
mock_playlist.add = Mock()
|
|
|
|
# Mock session.playlist() to return our mock playlist
|
|
self.mock_session.playlist.return_value = mock_playlist
|
|
|
|
# Mock session.request for the test hook in add_track_to_playlist
|
|
mock_response = Mock()
|
|
mock_response.raise_for_status = Mock()
|
|
self.mock_session.request.return_value = mock_response
|
|
|
|
# Create checkbox mock
|
|
mock_checkbox = Mock()
|
|
|
|
# Simulate add operation
|
|
dialog._api_add_track_to_playlist("track_uuid_1", "playlist_2", mock_checkbox, False)
|
|
|
|
# Cache should be updated
|
|
self.assertTrue(self.cache.is_track_in_playlist("track_uuid_1", "playlist_2"))
|
|
|
|
# Checkbox should be re-enabled
|
|
mock_checkbox.setEnabled.assert_called_with(True)
|
|
|
|
# Verify API was called
|
|
self.mock_session.playlist.assert_called_once_with("playlist_2")
|
|
self.mock_session.request.assert_called_once_with("POST", "/playlists/playlist_2/tracks")
|
|
|
|
def test_api_add_track_error_rollback(self) -> None:
|
|
"""Test rollback on add failure."""
|
|
dialog = PlaylistManagerDialog(
|
|
track=self.mock_track,
|
|
cache=self.cache,
|
|
session=self.mock_session,
|
|
threadpool=self.threadpool,
|
|
)
|
|
|
|
# Mock playlist object
|
|
mock_playlist = Mock()
|
|
mock_playlist.id = "playlist_2"
|
|
self.mock_session.playlist.return_value = mock_playlist
|
|
|
|
# Mock failed API response
|
|
mock_response = Mock()
|
|
mock_response.raise_for_status = Mock(side_effect=Exception("API Error"))
|
|
self.mock_session.request.return_value = mock_response
|
|
|
|
# Create checkbox mock
|
|
mock_checkbox = Mock()
|
|
|
|
# Original state: not checked
|
|
original_state = False
|
|
|
|
# Simulate failed add operation
|
|
dialog._api_add_track_to_playlist("track_uuid_1", "playlist_2", mock_checkbox, original_state)
|
|
|
|
# Checkbox should be restored to original state
|
|
mock_checkbox.setChecked.assert_called()
|
|
|
|
# Checkbox should be re-enabled
|
|
mock_checkbox.setEnabled.assert_called_with(True)
|
|
|
|
def test_api_remove_track_success(self) -> None:
|
|
"""Test successful track removal."""
|
|
dialog = PlaylistManagerDialog(
|
|
track=self.mock_track,
|
|
cache=self.cache,
|
|
session=self.mock_session,
|
|
threadpool=self.threadpool,
|
|
)
|
|
|
|
# Mock playlist object and its methods for remove_track_from_playlist
|
|
mock_playlist = Mock()
|
|
mock_playlist.id = "playlist_1"
|
|
mock_playlist._items = None
|
|
|
|
# Mock playlist.items() to return a track
|
|
mock_track_item = Mock()
|
|
mock_track_item.id = "track_uuid_1"
|
|
mock_playlist.items = Mock(return_value=[mock_track_item])
|
|
|
|
# Mock remove_by_index
|
|
mock_playlist.remove_by_index = Mock()
|
|
|
|
# Mock session.playlist() to return our mock playlist
|
|
self.mock_session.playlist.return_value = mock_playlist
|
|
|
|
# Create checkbox mock
|
|
mock_checkbox = Mock()
|
|
|
|
# Verify track is in cache initially
|
|
self.assertTrue(self.cache.is_track_in_playlist("track_uuid_1", "playlist_1"))
|
|
|
|
# Simulate remove operation
|
|
dialog._api_remove_track_from_playlist("track_uuid_1", "playlist_1", mock_checkbox, True)
|
|
|
|
# Cache should be updated (track removed)
|
|
self.assertFalse(self.cache.is_track_in_playlist("track_uuid_1", "playlist_1"))
|
|
|
|
# Checkbox should be re-enabled
|
|
mock_checkbox.setEnabled.assert_called_with(True)
|
|
|
|
# Verify playlist methods were called
|
|
self.mock_session.playlist.assert_called_once_with("playlist_1")
|
|
mock_playlist.remove_by_index.assert_called_once_with(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|