304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""Regression coverage for background work, cache isolation, and playback."""
|
|
import logging
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from PySide6.QtCore import Qt
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
from PySide6.QtWidgets import QApplication, QDialog, QTableWidget, QTreeWidgetItem, QWidget
|
|
|
|
from src.api.client import SubsonicClient
|
|
from src.api.models import Song, Playlist
|
|
from src.cache import CachedClient, CacheDatabase, CacheManager
|
|
from src.main_window import MainWindow
|
|
from src.managers.playback_manager import PlaybackManager
|
|
from src.widgets.search_dialog import SearchDialog
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def app():
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
def wait_for(app, predicate):
|
|
deadline = time.monotonic() + 3
|
|
while not predicate() and time.monotonic() < deadline:
|
|
app.processEvents()
|
|
time.sleep(0.005)
|
|
assert predicate()
|
|
|
|
|
|
def song(identifier):
|
|
return Song(id=identifier, title=identifier, album="Album", albumId="a", artist="Artist")
|
|
|
|
|
|
def test_background_error_survives_until_ui_delivery(app):
|
|
callbacks, errors = [], []
|
|
with ThreadPoolExecutor(1) as executor:
|
|
harness = SimpleNamespace(_executor=executor, logger=logging.getLogger("test"),
|
|
_on_ui=callbacks.append)
|
|
def fail():
|
|
raise ValueError("synthetic failure")
|
|
MainWindow._run_background(harness, "test", fail, on_error=errors.append)
|
|
assert len(callbacks) == 1
|
|
callbacks[0]()
|
|
assert str(errors[0]) == "synthetic failure"
|
|
|
|
|
|
def test_keyboard_shortcuts_dialog_has_accessible_shortcut_table(app, monkeypatch):
|
|
parent = QWidget()
|
|
|
|
def inspect_dialog(dialog):
|
|
table = dialog.findChild(QTableWidget)
|
|
assert dialog.windowTitle() == "Keyboard Shortcuts"
|
|
assert table.accessibleName() == "Keyboard shortcuts"
|
|
assert table.horizontalHeaderItem(0).text() == "Shortcut"
|
|
assert table.horizontalHeaderItem(1).text() == "Action"
|
|
assert table.rowCount() >= 20
|
|
assert table.item(15, 0).text() == "Ctrl+Down or 9"
|
|
assert table.item(15, 1).text() == "Decrease volume"
|
|
return QDialog.Accepted
|
|
|
|
monkeypatch.setattr(QDialog, "exec", inspect_dialog)
|
|
MainWindow.showKeyboardShortcuts(parent)
|
|
|
|
|
|
@pytest.mark.parametrize("refresh", ["forced", "full", "incremental"])
|
|
def test_refresh_bypasses_api_memory_cache(tmp_path, refresh):
|
|
raw = SubsonicClient("https://fake.example", "user", "fake")
|
|
state = {"name": "Old"}
|
|
def request(endpoint, params=None):
|
|
if endpoint == "getArtists":
|
|
return {"artists": {"index": [{"artist": [{"id": "a", "name": state["name"], "albumCount": 0}]}]}}
|
|
if endpoint == "getArtist":
|
|
return {"artist": {"id": "a", "name": state["name"]}}
|
|
return {}
|
|
raw._makeRequest = Mock(side_effect=request)
|
|
cached = CachedClient(raw, tmp_path)
|
|
try:
|
|
assert cached.getArtists()[0].name == "Old"
|
|
state["name"] = "New"
|
|
if refresh == "forced":
|
|
result = cached.getArtists(forceRefresh=True)
|
|
else:
|
|
getattr(cached, "syncFull" if refresh == "full" else "syncIncremental")()
|
|
result = cached.getArtists()
|
|
assert result[0].name == "New"
|
|
finally:
|
|
cached.db.close()
|
|
|
|
|
|
def test_accounts_have_separate_persistent_caches(tmp_path):
|
|
first = CachedClient(SubsonicClient("https://fake.example", "alice", "fake"), tmp_path)
|
|
second = CachedClient(SubsonicClient("https://fake.example", "bob", "fake"), tmp_path)
|
|
again = CachedClient(SubsonicClient("https://fake.example", "alice", "new password"), tmp_path)
|
|
try:
|
|
first.cache.setPlaylist(Playlist(id="p", name="Private", songs=[song("s")]))
|
|
assert second.cache.getPlaylists() == []
|
|
assert again.cache.getPlaylist("p").name == "Private"
|
|
assert first.db.dbPath != second.db.dbPath
|
|
finally:
|
|
for client in (first, second, again):
|
|
client.db.close()
|
|
|
|
|
|
def test_playlist_metadata_preserves_order_and_empty_detail_clears(tmp_path):
|
|
db = CacheDatabase("https://fake.example", tmp_path)
|
|
cache = CacheManager(db)
|
|
try:
|
|
cache.setPlaylist(Playlist(id="p", name="Before", songs=[song("b"), song("a"), song("b")]))
|
|
cache.setPlaylists([Playlist(id="p", name="After", songCount=3)])
|
|
assert [s.id for s in cache.getPlaylist("p").songs] == ["b", "a", "b"]
|
|
assert cache.getPlaylist("p").name == "After"
|
|
cache.setPlaylist(Playlist(id="p", name="Empty", songs=[]))
|
|
assert cache.getPlaylist("p").songs == []
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_search_activation_enqueues_once(app):
|
|
parent = QWidget()
|
|
parent.playback = SimpleNamespace(enqueue=Mock())
|
|
parent.enablePlaybackActions = Mock()
|
|
parent.statusBar = SimpleNamespace(showMessage=Mock())
|
|
dialog = SearchDialog(Mock(), parent)
|
|
dialog.songActivated.connect(lambda s: MainWindow.onSearchSongChosen(parent, s))
|
|
chosen = song("s")
|
|
item = QTreeWidgetItem(["song"])
|
|
item.setData(0, Qt.UserRole, {"type": "song", "song": chosen})
|
|
dialog.handleActivate(item)
|
|
parent.playback.enqueue.assert_called_once_with(chosen, playNow=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def playback(app, monkeypatch):
|
|
# Exercise the real manager with a fake media backend; no audio or network.
|
|
import src.managers.playback_manager as module
|
|
factory = Mock()
|
|
factory.PlayingState = QMediaPlayer.PlayingState
|
|
factory.PausedState = QMediaPlayer.PausedState
|
|
factory.StoppedState = QMediaPlayer.StoppedState
|
|
monkeypatch.setattr(module, "QMediaPlayer", factory)
|
|
monkeypatch.setattr(module, "QAudioOutput", Mock())
|
|
manager = PlaybackManager()
|
|
manager.player.playbackState.return_value = QMediaPlayer.PlayingState
|
|
manager.setStreamResolver(lambda s: "https://fake.example/" + s.id)
|
|
manager.queue = [song(str(i)) for i in range(4)]
|
|
manager.currentIndex = 1
|
|
manager.shuffleEnabled = True
|
|
manager._shuffleOrder = [2, 1, 3, 0]
|
|
manager._shufflePosition = 1
|
|
return manager
|
|
|
|
|
|
@pytest.mark.parametrize("removed", [0, 2, 3])
|
|
def test_queue_removal_preserves_current_and_valid_shuffle(playback, removed):
|
|
current = playback.currentSong()
|
|
playback.removeIndex(removed)
|
|
assert playback.currentSong() is current
|
|
assert sorted(playback._shuffleOrder) == list(range(3))
|
|
assert playback._shuffleOrder[playback._shufflePosition] == playback.currentIndex
|
|
playback.player.setSource.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("state", [QMediaPlayer.PlayingState, QMediaPlayer.PausedState, QMediaPlayer.StoppedState])
|
|
def test_removing_current_selects_shuffle_successor(playback, state):
|
|
playback.player.playbackState.return_value = state
|
|
changes = []
|
|
playback.trackChanged.connect(changes.append)
|
|
playback.removeIndex(1)
|
|
assert playback.currentSong().id == "3"
|
|
assert changes[-1].id == "3"
|
|
assert sorted(playback._shuffleOrder) == [0, 1, 2]
|
|
if state == QMediaPlayer.PausedState:
|
|
playback.player.pause.assert_called_once()
|
|
elif state == QMediaPlayer.StoppedState:
|
|
playback.player.stop.assert_called_once()
|
|
|
|
|
|
def test_removing_last_track_clears_source(playback):
|
|
playback.queue = [song("only")]
|
|
playback.currentIndex = 0
|
|
playback.removeIndex(0)
|
|
assert playback.currentSong() is None
|
|
assert playback.queue == []
|
|
assert playback._shuffleOrder == []
|
|
assert playback.player.setSource.call_args.args[0].isEmpty()
|
|
|
|
|
|
def test_search_is_nonblocking_and_discards_old_results(app):
|
|
release = threading.Event()
|
|
started = threading.Event()
|
|
def search(query, **kwargs):
|
|
if query == "old":
|
|
started.set()
|
|
assert release.wait(3)
|
|
return {"songs": [song(query)]}
|
|
dialog = SearchDialog(SimpleNamespace(search=search))
|
|
try:
|
|
dialog.queryEdit.setText("old")
|
|
dialog.runSearch()
|
|
assert started.wait(1)
|
|
dialog.queryEdit.setText("new")
|
|
dialog.runSearch()
|
|
wait_for(app, lambda: dialog._resultsLoaded)
|
|
assert dialog.resultsTree.topLevelItem(0).child(0).data(0, Qt.UserRole)["song"].id == "new"
|
|
release.set()
|
|
dialog._executor.shutdown(wait=True)
|
|
app.processEvents()
|
|
assert dialog.resultsTree.topLevelItem(0).child(0).data(0, Qt.UserRole)["song"].id == "new"
|
|
finally:
|
|
release.set()
|
|
dialog.reject()
|
|
|
|
|
|
@pytest.mark.parametrize("invalidate", ["search", "close"])
|
|
def test_expansion_results_are_ignored_after_search_or_close(app, invalidate):
|
|
release = threading.Event()
|
|
started = threading.Event()
|
|
def get_album(identifier):
|
|
started.set()
|
|
assert release.wait(3)
|
|
return {"songs": [song("late")]}
|
|
client = SimpleNamespace(getAlbum=get_album, search=lambda *args, **kw: {})
|
|
dialog = SearchDialog(client)
|
|
item = QTreeWidgetItem(["album"])
|
|
item.setData(0, Qt.UserRole, {"type": "album", "id": "a", "loaded": False})
|
|
dialog.resultsTree.addTopLevelItem(item)
|
|
try:
|
|
dialog.onItemExpanded(item)
|
|
assert started.wait(1)
|
|
if invalidate == "search":
|
|
dialog.queryEdit.setText("new")
|
|
dialog.runSearch()
|
|
else:
|
|
dialog.reject()
|
|
release.set()
|
|
dialog._executor.shutdown(wait=True)
|
|
app.processEvents()
|
|
assert dialog.resultsTree.topLevelItemCount() == (0 if invalidate == "search" else 1)
|
|
finally:
|
|
release.set()
|
|
dialog.reject()
|
|
|
|
|
|
def test_unchanged_playlist_keeps_tracks_across_incremental_sync(tmp_path):
|
|
from test_cache_basic import FakeClient
|
|
client = CachedClient(FakeClient(), tmp_path)
|
|
try:
|
|
client.syncIncremental()
|
|
client.syncIncremental()
|
|
assert [s.id for s in client.cache.getPlaylist("p1").songs] == ["s1"]
|
|
finally:
|
|
client.db.close()
|
|
|
|
|
|
@pytest.mark.parametrize("index, expected", [(1, "2"), (3, "2")])
|
|
def test_removing_current_without_shuffle(playback, index, expected):
|
|
playback.setShuffle(False)
|
|
playback.currentIndex = index
|
|
playback.removeIndex(index)
|
|
assert playback.currentSong().id == expected
|
|
assert playback._shuffleOrder == []
|
|
|
|
|
|
def test_expansion_fetches_off_thread_and_updates_ui(app):
|
|
main_thread = threading.get_ident()
|
|
worker_threads = []
|
|
def get_album(identifier):
|
|
worker_threads.append(threading.get_ident())
|
|
return {"songs": [song("loaded")]}
|
|
dialog = SearchDialog(SimpleNamespace(getAlbum=get_album))
|
|
item = QTreeWidgetItem(["album"])
|
|
item.setData(0, Qt.UserRole, {"type": "album", "id": "a", "loaded": False})
|
|
dialog.resultsTree.addTopLevelItem(item)
|
|
try:
|
|
dialog.onItemExpanded(item)
|
|
wait_for(app, lambda: item.data(0, Qt.UserRole).get("loaded"))
|
|
assert item.child(0).data(0, Qt.UserRole)["song"].id == "loaded"
|
|
assert worker_threads[0] != main_thread
|
|
finally:
|
|
dialog.reject()
|
|
|
|
|
|
def test_search_error_delivered_on_ui_thread(app, monkeypatch):
|
|
from src.widgets.accessible_text_dialog import AccessibleTextDialog
|
|
errors = []
|
|
monkeypatch.setattr(AccessibleTextDialog, "showError",
|
|
lambda title, message, parent: errors.append((message, threading.get_ident())))
|
|
def fail(*args, **kwargs):
|
|
raise ValueError("synthetic search failure")
|
|
dialog = SearchDialog(SimpleNamespace(search=fail))
|
|
try:
|
|
dialog.queryEdit.setText("query")
|
|
dialog.runSearch()
|
|
wait_for(app, lambda: bool(errors))
|
|
assert errors == [("synthetic search failure", threading.get_ident())]
|
|
assert dialog.statusLabel.text() == "Search failed."
|
|
finally:
|
|
dialog.reject()
|