"""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() def test_long_pause_resume_seeks_in_place(self, qapp, tmp_path, monkeypatch): import time as time_module player = _mock_player(qapp, tmp_path) player._current_track = Track(track_id=1, name="A") monkeypatch.setattr(player, "is_playing", lambda: False) # Short pause: no nudge. player._paused_at = time_module.monotonic() - 5 player.toggle_play() player._media.setPosition.assert_not_called() # Walk-away pause: seek-in-place re-primes the suspended sink. player._media.position.return_value = 12345 player._paused_at = time_module.monotonic() - 600 player.toggle_play() player._media.setPosition.assert_called_once_with(12345) assert player._paused_at is None # ---- 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() class TestAlbumWideArt: def test_album_tracks_matches_album_case_insensitively(self): from lintunes.gui.main_window import album_tracks library = Library() library.tracks = { 1: Track(track_id=1, artist="Artist", album="Great Album"), 2: Track(track_id=2, artist="artist", album="great album"), 3: Track(track_id=3, artist="Someone Else", album="Great Album", album_artist="Artist"), # album_artist wins the grouping 4: Track(track_id=4, artist="Artist", album="Other Album"), 5: Track(track_id=5, artist="Other Artist", album="Great Album"), } ids = {t.track_id for t in album_tracks(library, "Artist", "Great Album")} assert ids == {1, 2, 3} def _dialog(self, monkeypatch, image, track_count, album_count): from lintunes.art_search import AlbumArtFetcher, ArtCandidate from lintunes.gui.album_art_dialog import AlbumArtDialog # No network: neuter the fetch and feed the image back by hand. monkeypatch.setattr(AlbumArtFetcher, "fetch", lambda self, c: None) candidate = ArtCandidate(artist="A", album="B", art_url="https://x") dialog = AlbumArtDialog([candidate], track_count, album_count=album_count) dialog._on_image({"candidate": candidate, "image": image, "mime": "image/jpeg"}) return dialog def test_album_button_applies_to_whole_album(self, qapp, monkeypatch, jpeg_bytes): dialog = self._dialog(monkeypatch, jpeg_bytes, track_count=1, album_count=12) assert dialog._album_btn is not None assert dialog._album_btn.isEnabled() assert "12" in dialog._album_btn.text() assert not dialog.apply_to_album dialog._accept_album() assert dialog.apply_to_album assert dialog.selected_image == jpeg_bytes assert dialog.selected_mime == "image/jpeg" def test_no_album_button_when_selection_covers_album(self, qapp, monkeypatch, jpeg_bytes): dialog = self._dialog(monkeypatch, jpeg_bytes, track_count=5, album_count=5) assert dialog._album_btn is None dialog._accept_current() assert not dialog.apply_to_album assert dialog.selected_image == jpeg_bytes # ---- 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]]