trav
8dc33fa65e
Features:
- Rating hover dots: hovering a rating cell shows 5 clickable slots
(RatingDelegate); click slot k sets k stars, clicking the current count
clears. Undoable, library-only (ratings never rewrite music files).
- Search bar moved into the Library header strip, right of the Library
button, so the tracklist top aligns with the playlist tree.
- Right-click "Download Album Art…": iTunes Search API (no key), off-thread
fetch, preview/confirm dialog with Next Result, embeds via write_artwork
+ size refresh, invalidates the MPRIS art cache.
Fixes:
- MPRIS media keys: PropertiesChanged sent invalidated_properties as "av"
instead of "as", so gsd-media-keys dropped it and never MRU-bumped
lintunes (why the play/pause key kept waking stale players). Now an
explicit QDBusArgument string array; loopback-verified sa{sv}as.
- Exit segfault: ordered Player.shutdown() (stop, clear source, detach
buffer/audio outputs) from closeEvent/aboutToQuit; scripted run exits 0.
- BT zero-volume after pause/resume: volume re-applied on resume, device
swap, and BufferedMedia (needs verify on the affected machine).
Cruft sweep:
- Tag writes filtered to EDITABLE_FIELDS; failures logged + surfaced in
the status bar (was a swallowed print).
- O(n²) import fixed (location index + cached max track id); O(1)
refresh/reveal via TrackTableModel row index.
- lastfm: scrobble-queue thread lock, login prefs write marshalled to the
GUI thread, JSON/raise_for_status order fixed.
- Shared read_json/write_json; TableSettingsMixin dedupes view settings.
TASKS.md rewritten as a resumable board; tests in tests/test_round17.py
(251 total pass).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
394 lines
16 KiB
Python
394 lines
16 KiB
Python
"""Round 17: cruft cleanup + rating clicks, search-bar move, player shutdown,
|
|
volume re-apply, MPRIS marshalling fix, album-art download plumbing."""
|
|
import threading
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from PyQt6.QtCore import QCoreApplication, Qt, QPoint, QPointF, QUrl
|
|
from PyQt6.QtGui import QFontMetrics, QMouseEvent
|
|
|
|
from lintunes import player as player_module
|
|
from lintunes import tagging
|
|
from lintunes.lastfm import LastFm
|
|
from lintunes.library_manager import LibraryManager
|
|
from lintunes.models import Track
|
|
from lintunes.models.library import Library
|
|
from lintunes.player import Player
|
|
|
|
|
|
def _manager(tmp_path, tracks=()):
|
|
library = Library()
|
|
for track in tracks:
|
|
library.tracks[track.track_id] = track
|
|
return LibraryManager(library, tmp_path)
|
|
|
|
|
|
# ---- 1a: tag-write filter (rating/size edits never rewrite music files) ----
|
|
|
|
class TestTagWriteFilter:
|
|
def test_rating_edit_succeeds_without_file(self, qapp, tmp_path):
|
|
track = Track(track_id=1, name="Song",
|
|
location=str(tmp_path / "missing.mp3"))
|
|
manager = _manager(tmp_path, [track])
|
|
manager.edit_track_fields(1, {"rating": 80})
|
|
assert track.rating == 80
|
|
manager.undo_stack.undo()
|
|
assert track.rating == 0
|
|
|
|
def test_rating_edit_leaves_file_untouched(self, qapp, tmp_path, mp3_file):
|
|
track = Track(track_id=1, name="Song", location=str(mp3_file))
|
|
manager = _manager(tmp_path, [track])
|
|
before = mp3_file.stat().st_mtime_ns
|
|
manager.edit_track_fields(1, {"rating": 60})
|
|
assert track.rating == 60
|
|
assert mp3_file.stat().st_mtime_ns == before
|
|
|
|
def test_mixed_edit_still_writes_tags(self, qapp, tmp_path, mp3_file):
|
|
track = Track(track_id=1, name="Old", location=str(mp3_file))
|
|
manager = _manager(tmp_path, [track])
|
|
manager.edit_track_fields(1, {"name": "New", "rating": 40})
|
|
assert track.name == "New" and track.rating == 40
|
|
assert tagging.read_tags(mp3_file)["name"] == "New"
|
|
|
|
def test_failed_tag_write_emits_signal(self, qapp, tmp_path, monkeypatch):
|
|
track = Track(track_id=1, name="Song", location=str(tmp_path / "x.mp3"))
|
|
manager = _manager(tmp_path, [track])
|
|
failures = []
|
|
manager.tag_write_failed.connect(lambda name, err: failures.append(name))
|
|
|
|
def boom(path, fields):
|
|
raise OSError("disk full")
|
|
monkeypatch.setattr("lintunes.tagging.write_tags", boom)
|
|
manager.edit_track_fields(1, {"name": "New"})
|
|
assert failures == ["Song"]
|
|
assert track.name == "Song" # edit not applied when the write failed
|
|
|
|
|
|
# ---- 1c: import dedup index + cached max track id ----
|
|
|
|
class TestImportIndex:
|
|
def test_duplicate_import_returns_existing(self, qapp, tmp_path, mp3_file):
|
|
from lintunes.importers.file_importer import import_paths
|
|
manager = _manager(tmp_path)
|
|
music = tmp_path / "music"
|
|
first = import_paths([mp3_file], music, manager)
|
|
again = import_paths([mp3_file], music, manager)
|
|
assert len(manager.library.tracks) == 1
|
|
assert first[0].track_id == again[0].track_id
|
|
|
|
def test_new_track_id_tracks_additions(self, qapp, tmp_path):
|
|
manager = _manager(tmp_path, [Track(track_id=7, name="A")])
|
|
assert manager.new_track_id() == 8
|
|
manager.add_track(Track(name="B"))
|
|
assert manager.new_track_id() == 9
|
|
|
|
def test_new_track_id_recomputed_on_reload(self, qapp, tmp_path):
|
|
from lintunes.storage import json_storage
|
|
manager = _manager(tmp_path, [Track(track_id=1, name="A")])
|
|
manager.flush()
|
|
# Another machine synced in a track with a higher id.
|
|
disk = Library()
|
|
disk.tracks[1] = Track(track_id=1, name="A")
|
|
disk.tracks[42] = Track(track_id=42, name="B")
|
|
json_storage.save_tracks(disk, tmp_path)
|
|
manager.reload_from_disk()
|
|
assert manager.new_track_id() == 43
|
|
|
|
|
|
# ---- 1d: model id→row index ----
|
|
|
|
class TestModelRowIndex:
|
|
def test_refresh_and_row_for_id(self, qapp):
|
|
from lintunes.gui.track_table import TrackTableModel
|
|
model = TrackTableModel()
|
|
a, b = Track(track_id=1, name="A"), Track(track_id=2, name="B")
|
|
model.set_tracks([a, b, a]) # duplicate row, as in a playlist
|
|
assert model.row_for_id(1) == 0
|
|
assert model.row_for_id(2) == 1
|
|
assert model.row_for_id(99) is None
|
|
changed = []
|
|
model.dataChanged.connect(lambda tl, br: changed.append(tl.row()))
|
|
model.refresh_track(1)
|
|
assert changed == [0, 2] # both occurrences repainted
|
|
|
|
|
|
# ---- 1e: scrobble queue lock ----
|
|
|
|
class TestScrobbleQueue:
|
|
def test_concurrent_enqueue_keeps_all(self, qapp, tmp_path):
|
|
prefs = MagicMock()
|
|
prefs.lastfm = {}
|
|
lastfm = LastFm(prefs, tmp_path)
|
|
threads = [threading.Thread(
|
|
target=lastfm._enqueue, args=({"artist": "A", "track": str(i),
|
|
"timestamp": i},))
|
|
for i in range(20)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
assert len(lastfm._load_queue()) == 20
|
|
|
|
|
|
# ---- 1f: login prefs write marshalled to the GUI thread ----
|
|
|
|
class TestLoginMarshal:
|
|
def test_login_updates_prefs_on_gui_thread(self, qapp, tmp_path, monkeypatch):
|
|
prefs = MagicMock()
|
|
prefs.lastfm = {}
|
|
lastfm = LastFm(prefs, tmp_path)
|
|
monkeypatch.setattr(
|
|
LastFm, "_call",
|
|
staticmethod(lambda *a, **k: {"session": {"name": "trav",
|
|
"key": "sk123"}}))
|
|
results = []
|
|
lastfm.login_finished.connect(lambda ok, msg: results.append((ok, msg)))
|
|
lastfm.login("key", "secret", "trav", "pw")
|
|
for _ in range(100):
|
|
QCoreApplication.processEvents()
|
|
if results:
|
|
break
|
|
threading.Event().wait(0.02)
|
|
assert results and results[0][0] is True
|
|
prefs.update_lastfm.assert_called_once_with(
|
|
api_key="key", api_secret="secret", username="trav",
|
|
session_key="sk123")
|
|
|
|
|
|
# ---- Task C: MPRIS PropertiesChanged marshalled as "as" ----
|
|
|
|
class TestMprisSignal:
|
|
def test_signal_marshals_as_sa_sv_as(self, qapp):
|
|
"""Send the signal over the real session bus and check the signature
|
|
it goes out with. A plain [] marshals invalidated_properties as "av"
|
|
and GDBus consumers (gsd-media-keys) validate + drop the signal; the
|
|
spec requires "as"."""
|
|
import pytest
|
|
from PyQt6.QtDBus import QDBusConnection
|
|
from lintunes.mpris import OBJECT_PATH, build_properties_changed
|
|
from PyQt6.QtCore import QObject, pyqtSlot
|
|
from PyQt6.QtDBus import QDBusMessage
|
|
|
|
class Receiver(QObject):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.messages = []
|
|
|
|
@pyqtSlot(QDBusMessage)
|
|
def on_signal(self, msg):
|
|
self.messages.append(msg)
|
|
|
|
bus = QDBusConnection.sessionBus()
|
|
if not bus.isConnected():
|
|
pytest.skip("no D-Bus session bus in this environment")
|
|
receiver = Receiver()
|
|
assert bus.connect(
|
|
"", OBJECT_PATH, "org.freedesktop.DBus.Properties",
|
|
"PropertiesChanged", receiver.on_signal)
|
|
assert bus.send(build_properties_changed(
|
|
{"PlaybackStatus": "Playing"}))
|
|
for _ in range(100):
|
|
QCoreApplication.processEvents()
|
|
if receiver.messages:
|
|
break
|
|
threading.Event().wait(0.02)
|
|
assert receiver.messages, "signal was not delivered back over the bus"
|
|
assert receiver.messages[0].signature() == "sa{sv}as"
|
|
|
|
def test_invalidate_artwork_removes_cache_file(self, tmp_path, monkeypatch):
|
|
from lintunes import mpris
|
|
monkeypatch.setattr(mpris, "_art_cache_dir", lambda: tmp_path)
|
|
cached = tmp_path / "ABCD1234.jpg"
|
|
cached.write_bytes(b"old art")
|
|
mpris.invalidate_artwork(Track(track_id=1, persistent_id="ABCD1234"))
|
|
assert not cached.exists()
|
|
# Missing file / no persistent id are silent no-ops.
|
|
mpris.invalidate_artwork(Track(track_id=1, persistent_id="ABCD1234"))
|
|
mpris.invalidate_artwork(None)
|
|
|
|
|
|
# ---- Tasks E + F: player shutdown order and volume re-apply ----
|
|
|
|
def _mock_player(qapp, tmp_path):
|
|
manager = _manager(tmp_path)
|
|
with patch.multiple(player_module,
|
|
QMediaPlayer=MagicMock(),
|
|
QAudioOutput=MagicMock(),
|
|
QAudioBufferOutput=MagicMock(),
|
|
QMediaDevices=MagicMock()):
|
|
player = Player(manager)
|
|
return player
|
|
|
|
|
|
class TestPlayerShutdown:
|
|
def test_shutdown_detaches_in_order_and_is_idempotent(self, qapp, tmp_path):
|
|
player = _mock_player(qapp, tmp_path)
|
|
player.shutdown()
|
|
player._media.stop.assert_called_once()
|
|
player._media.setSource.assert_called_once_with(QUrl())
|
|
player._media.setAudioBufferOutput.assert_called_with(None)
|
|
player._media.setAudioOutput.assert_called_with(None)
|
|
player.shutdown() # second call must be a no-op
|
|
player._media.stop.assert_called_once()
|
|
|
|
|
|
class TestVolumeReapply:
|
|
def test_set_volume_applies_converted_gain(self, qapp, tmp_path):
|
|
from PyQt6.QtMultimedia import QAudio
|
|
player = _mock_player(qapp, tmp_path)
|
|
player.set_volume(0.5)
|
|
expected = QAudio.convertVolume(
|
|
0.5, QAudio.VolumeScale.LogarithmicVolumeScale,
|
|
QAudio.VolumeScale.LinearVolumeScale)
|
|
player._audio.setVolume.assert_called_with(expected)
|
|
|
|
def test_resume_reapplies_volume(self, qapp, tmp_path, monkeypatch):
|
|
player = _mock_player(qapp, tmp_path)
|
|
player._current_track = Track(track_id=1, name="A")
|
|
applied = []
|
|
monkeypatch.setattr(player, "_apply_volume",
|
|
lambda: applied.append(True))
|
|
monkeypatch.setattr(player, "is_playing", lambda: False)
|
|
player.toggle_play() # paused with a cued track → resume
|
|
assert applied == [True]
|
|
player._media.play.assert_called_once()
|
|
|
|
|
|
# ---- Task B: search bar lives in the header strip ----
|
|
|
|
class TestSearchBarPlacement:
|
|
def test_search_parent_is_top_strip(self, qapp, tmp_path):
|
|
from lintunes.gui.library_view import LibraryView
|
|
manager = _manager(tmp_path, [Track(track_id=1, name="A")])
|
|
view = LibraryView(manager)
|
|
assert view._search.parentWidget() is view._top_strip
|
|
view.set_header_height(34)
|
|
assert view._top_strip.height() >= 0 # metrics path still works
|
|
|
|
def test_search_still_filters(self, qapp, tmp_path):
|
|
from lintunes.gui.library_view import LibraryView
|
|
manager = _manager(tmp_path, [
|
|
Track(track_id=1, name="Apple Song", artist="X"),
|
|
Track(track_id=2, name="Banana Song", artist="Y")])
|
|
view = LibraryView(manager)
|
|
view._search.setText("banana")
|
|
view._apply_search()
|
|
assert [t.name for t in view.table.model_.tracks] == ["Banana Song"]
|
|
|
|
|
|
# ---- Task A: rating hover dots + click to rate ----
|
|
|
|
class TestRatingHelpers:
|
|
def test_rating_from_click(self):
|
|
from lintunes.gui.track_table import rating_from_click
|
|
assert rating_from_click(0, 3) == 60 # set 3 stars
|
|
assert rating_from_click(60, 5) == 100 # raise to 5
|
|
assert rating_from_click(60, 3) == 0 # click current count clears
|
|
assert rating_from_click(100, 1) == 20 # lower to 1
|
|
|
|
def test_rating_slot_at_bounds(self):
|
|
from lintunes.gui.track_table import rating_slot_at
|
|
assert rating_slot_at(0, 14) == 1
|
|
assert rating_slot_at(-5, 14) == 1 # left edge clamps
|
|
assert rating_slot_at(3 + 14 * 2, 14) == 3
|
|
assert rating_slot_at(9999, 14) == 5 # right edge clamps
|
|
assert rating_slot_at(10, 0) == 1 # degenerate width
|
|
|
|
|
|
class TestRatingClicks:
|
|
def _view(self, qapp, tracks):
|
|
from lintunes.gui.track_table import TrackTableView
|
|
from lintunes.models.playlist import PlaylistSettings
|
|
view = TrackTableView(playlist_mode=False)
|
|
view.apply_settings(PlaylistSettings(
|
|
visible_columns=["name", "rating"], sort_column="name"))
|
|
view.set_tracks(tracks)
|
|
view.resize(500, 300)
|
|
return view
|
|
|
|
def test_click_on_rating_cell_emits(self, qapp):
|
|
from lintunes.gui.track_table import (
|
|
RATING_LEFT_PAD, rating_slot_width)
|
|
track = Track(track_id=1, name="A", rating=40)
|
|
view = self._view(qapp, [track])
|
|
emitted = []
|
|
view.rating_edited.connect(lambda tid, r: emitted.append((tid, r)))
|
|
col = view.model_.fields.index("rating")
|
|
rect = view.visualRect(view.proxy.index(0, col))
|
|
slot_w = rating_slot_width(QFontMetrics(view.font()))
|
|
# Click the 4th slot → 80.
|
|
x = rect.left() + RATING_LEFT_PAD + int(slot_w * 3.5)
|
|
pos = QPointF(x, rect.center().y())
|
|
event = QMouseEvent(QMouseEvent.Type.MouseButtonPress, pos,
|
|
Qt.MouseButton.LeftButton,
|
|
Qt.MouseButton.LeftButton,
|
|
Qt.KeyboardModifier.NoModifier)
|
|
view.mousePressEvent(event)
|
|
assert emitted == [(1, 80)]
|
|
|
|
def test_library_view_wiring_edits_and_undoes(self, qapp, tmp_path):
|
|
from lintunes.gui.library_view import LibraryView
|
|
track = Track(track_id=1, name="A")
|
|
manager = _manager(tmp_path, [track])
|
|
view = LibraryView(manager)
|
|
view.table.rating_edited.emit(1, 80)
|
|
assert track.rating == 80
|
|
manager.undo_stack.undo()
|
|
assert track.rating == 0
|
|
|
|
def test_hover_state_tracks_cell(self, qapp):
|
|
track = Track(track_id=1, name="A", rating=40)
|
|
view = self._view(qapp, [track])
|
|
col = view.model_.fields.index("rating")
|
|
index = view.proxy.index(0, col)
|
|
view._set_rating_hover(index)
|
|
assert view.is_rating_hovered(index)
|
|
assert not view.is_rating_hovered(view.proxy.index(0, 0))
|
|
view._set_rating_hover(None)
|
|
assert not view.is_rating_hovered(index)
|
|
|
|
|
|
# ---- Task D: album art search plumbing (offline) ----
|
|
|
|
class TestArtSearch:
|
|
def test_upgrade_artwork_url(self):
|
|
from lintunes.art_search import upgrade_artwork_url
|
|
url = "https://is1.mzstatic.com/image/thumb/x/100x100bb.jpg"
|
|
assert "600x600bb.jpg" in upgrade_artwork_url(url)
|
|
assert "100x100" not in upgrade_artwork_url(url)
|
|
|
|
def test_parse_results(self):
|
|
from lintunes.art_search import parse_results
|
|
payload = {"results": [
|
|
{"artistName": "Artist", "collectionName": "Album",
|
|
"artworkUrl100": "https://x/100x100bb.jpg"},
|
|
{"artistName": "No Art"}, # skipped: no artworkUrl100
|
|
]}
|
|
candidates = parse_results(payload)
|
|
assert len(candidates) == 1
|
|
assert candidates[0].artist == "Artist"
|
|
assert candidates[0].album == "Album"
|
|
assert "600x600" in candidates[0].art_url
|
|
|
|
def test_embed_updates_file_and_size(self, qapp, tmp_path, mp3_file,
|
|
jpeg_bytes):
|
|
# The Task D accept path: write_artwork + size refresh, no undo entry.
|
|
track = Track(track_id=1, name="A", location=str(mp3_file), size=1)
|
|
manager = _manager(tmp_path, [track])
|
|
tagging.write_artwork(track.location, jpeg_bytes, "image/jpeg")
|
|
manager.update_track_fields(1, {"size": mp3_file.stat().st_size})
|
|
assert tagging.read_embedded_artwork(track.location) == jpeg_bytes
|
|
assert track.size == mp3_file.stat().st_size
|
|
assert not manager.undo_stack.can_undo()
|
|
|
|
|
|
# ---- context menu exposes the new action's signal ----
|
|
|
|
class TestDownloadArtSignal:
|
|
def test_signal_exists_and_carries_ids(self, qapp):
|
|
from lintunes.gui.track_table import TrackTableView
|
|
view = TrackTableView()
|
|
got = []
|
|
view.download_art_requested.connect(got.append)
|
|
view.download_art_requested.emit([1, 2])
|
|
assert got == [[1, 2]]
|