"""Round 69: followed playlists in my sidebar. * A followed friend playlist shows in my sidebar with a purple UFO — mixed in alphabetically, or inside a folder named after the friend (a per-friend setting) — and is never a Playlist of mine, so it stays out of the library and its conflict merge. * It plays from the cache: Player queues opaque keys ("cassette::") resolved through Cassette. Plays scrobble but don't count in my play journal. Songs not here yet are grey and skipped. * Its cassette column is hidden by default; pressing one takes the song into my library — straight from the cache if it's there, otherwise by asking. * Unfollowing frees the cache; requests never re-ask for cached songs. """ import shutil from pathlib import Path from unittest.mock import MagicMock import pytest from lintunes.cassette import delivery, matching, publish, share from lintunes.cassette import requests as req from lintunes.cassette.service import CassetteService from lintunes.cassette.state import ( CassetteState, Friend, Selection, cache_dir, in_dir, out_dir, ) from lintunes.gui.followed_view import parse_key, track_key from lintunes.models import Playlist, Track from lintunes.models.library import Library TOKEN = "0123456789abcdef" def _friend_share(root, mp3_file, cached=(10,)): theirs = Library() for tid, name in ((10, "One"), (11, "Two"), (12, "Three")): theirs.tracks[tid] = Track(track_id=tid, name=name, artist="Sam's Band", album="Demo", total_time=1000, size=100) theirs.playlists["P"] = Playlist(name="Road Trip", persistent_id="P", track_ids=[11, 10]) share.publish_share(in_dir(root, TOKEN), theirs, Selection(all_library=True, all_playlists=True)) publish.publish_intro(in_dir(root, TOKEN), "Sam", None) cache = cache_dir(root, TOKEN) cache.mkdir(parents=True, exist_ok=True) for tid in cached: shutil.copyfile(mp3_file, cache / f"{tid}.mp3") return theirs # ---- keys and the player ---- class TestPlayer: def test_keys(self): assert parse_key(track_key(TOKEN, 12)) == (TOKEN, 12) assert parse_key("12") is None and parse_key(12) is None def test_friend_track_scrobbles_but_isnt_counted(self, qapp, tmp_path, mp3_file): from lintunes.player import Player manager = MagicMock() player = Player(manager) friend_track = Track(track_id=5, name="Theirs", location=str(mp3_file)) player.track_resolver = lambda key: friend_track if key == "cassette:t:5" else None finished = [] player.track_finished.connect(finished.append) player._current_track = friend_track player._current_key = "cassette:t:5" player._note_finished(friend_track) player._note_finished(friend_track) manager.record_play.assert_not_called() assert finished == [friend_track] # Last.fm still hears it assert player._resolve("cassette:t:5") is friend_track assert player._resolve("cassette:t:6") is None # ---- the service never re-asks for cached songs ---- class TestRequests: def test_followed_songs_already_cached_arent_requested(self, qapp, tmp_path, mp3_file): root = tmp_path / "root" _friend_share(root, mp3_file, cached=(10,)) service = CassetteService(root=root, config_loader=lambda: {}) service._state.friends[TOKEN] = Friend(token=TOKEN, device_id="D", name="Sam") service.save_requests(TOKEN, {12}, ["P"]) assert req.read_requests(out_dir(root, TOKEN)) == {11, 12} def test_unfollowed_cache_is_pruned(self, tmp_path, mp3_file): root = tmp_path / "root" _friend_share(root, mp3_file, cached=(10, 12)) friend = Friend(token=TOKEN, device_id="D", name="Sam", followed=["P"]) plan = delivery.plan_receive(friend, root, matching.LibraryIndex([])) assert [p.name for p in plan.cache_prune] == ["12.mp3"] friend.followed = [] plan = delivery.plan_receive(friend, root, matching.LibraryIndex([])) assert sorted(p.name for p in plan.cache_prune) == ["10.mp3", "12.mp3"] delivery.run_plans([], [plan], None) assert list(cache_dir(root, TOKEN).iterdir()) == [] def test_no_playlists_file_prunes_nothing(self, tmp_path, mp3_file): root = tmp_path / "root" _friend_share(root, mp3_file, cached=(10,)) (in_dir(root, TOKEN) / share.PLAYLISTS_FILE).unlink() friend = Friend(token=TOKEN, device_id="D", name="Sam") plan = delivery.plan_receive(friend, root, matching.LibraryIndex([])) assert plan.cache_prune == [] # ---- the sidebar and the view ---- class _Window: def __init__(self, qapp): from PyQt6.QtWidgets import QMainWindow, QStackedWidget from lintunes.gui.sidebar import SidebarPanel self.main = QMainWindow() self.content_stack = QStackedWidget() self.main.setCentralWidget(self.content_stack) manager = MagicMock() manager.library.playlists = { "M1": Playlist(name="Morning", persistent_id="M1"), "M2": Playlist(name="Zebra", persistent_id="M2")} self.sidebar = SidebarPanel(manager, MagicMock()) self.player = MagicMock() self.player.current_key = None self.player.is_playing.return_value = False self.now_playing_context = "" self.played = [] self.totals = 0 def statusBar(self): return self.main.statusBar() def play_keys(self, keys, start, context): self.played.append((keys, start, context)) def update_totals(self): self.totals += 1 def show_my_library(self): pass @pytest.fixture def followed(qapp, tmp_path, mp3_file): from lintunes.gui.cassette_ui import CassetteUi from lintunes.library_manager import LibraryManager root = tmp_path / "root" _friend_share(root, mp3_file, cached=(10,)) service = CassetteService(root=root, config_loader=lambda: {}) service._state.friends[TOKEN] = Friend(token=TOKEN, device_id="D", name="Sam", followed=["P"], layout="mixed") manager = LibraryManager(Library(), tmp_path / "data") manager.organize_root = lambda: tmp_path / "Music" window = _Window(qapp) ui = CassetteUi(window.main, MagicMock(), manager) ui._window = window ui.service = service ui._hook_window() return ui, window, service, manager class TestSidebar: def test_mixed_in_alphabetically(self, followed): ui, window, _, _ = followed tree = window.sidebar.tree names = [tree.topLevelItem(i).text(0) for i in range(tree.topLevelItemCount())] assert names == ["Morning", "Road Trip", "Zebra"] def test_in_a_friend_folder(self, followed): ui, window, service, _ = followed service._state.friends[TOKEN].layout = "folder" ui._refresh_followed() tree = window.sidebar.tree names = [tree.topLevelItem(i).text(0) for i in range(tree.topLevelItemCount())] assert names == ["Morning", "Sam", "Zebra"] folder = tree.topLevelItem(1) assert folder.child(0).text(0) == "Road Trip" def test_followed_items_arent_editable_or_draggable(self, followed): from PyQt6.QtCore import Qt _, window, _, _ = followed tree = window.sidebar.tree item = tree._find_item(f"cassette:{TOKEN}:P") assert not item.flags() & Qt.ItemFlag.ItemIsEditable assert not item.flags() & Qt.ItemFlag.ItemIsDragEnabled class TestFollowedView: def test_cached_plays_uncached_is_grey(self, followed): ui, window, _, _ = followed ui.show_followed(f"cassette:{TOKEN}:P") view = ui.followed_view assert [t.track_id for t in view.table.model_.tracks] == [11, 10] assert view._foreground(view.table.model_.tracks[0]) is not None # 11 not here assert view._foreground(view.table.model_.tracks[1]) is None # 10 cached view._on_double_click(view.table.model_.index(1, 0)) assert window.played == [([track_key(TOKEN, 10)], 0, f"followed:cassette:{TOKEN}:P")] view._on_double_click(view.table.model_.index(0, 0)) # not here yet assert len(window.played) == 1 assert "last synced" in view.status_suffix() def test_cassette_column_hidden_by_default(self, followed): from lintunes.gui.track_table import CASSETTE_FIELD ui, _, _, _ = followed ui.show_followed(f"cassette:{TOKEN}:P") assert CASSETTE_FIELD not in ui.followed_view.table.model_.fields assert ui.followed_view.table.cassette_hooks is not None def test_resolve_points_at_the_cache(self, followed): ui, _, service, _ = followed track = ui.resolve_key(track_key(TOKEN, 10)) assert Path(track.location).parent == cache_dir(service.root, TOKEN) assert ui.resolve_key(track_key(TOKEN, 11)) is None def test_cassette_a_cached_song_imports_it_now(self, followed, tmp_path): ui, _, service, manager = followed ui.toggle_cassette(TOKEN, 10) [track] = manager.library.tracks.values() assert tmp_path / "Music" in Path(track.location).parents assert track.play_count == 0 assert service.state().friends[TOKEN].cassetted == [] def test_cassette_an_uncached_song_asks_for_it(self, followed): ui, _, service, manager = followed ui.toggle_cassette(TOKEN, 12) assert service.state().friends[TOKEN].cassetted == [12] assert 12 in req.read_requests(out_dir(service.root, TOKEN)) assert ui.cassette_state(TOKEN, 12) == "on" ui.toggle_cassette(TOKEN, 12) assert ui.cassette_state(TOKEN, 12) == "off" def test_unfollow(self, followed): ui, window, service, _ = followed ui.unfollow(f"cassette:{TOKEN}:P") assert service.state().friends[TOKEN].followed == [] tree = window.sidebar.tree names = [tree.topLevelItem(i).text(0) for i in range(tree.topLevelItemCount())] assert names == ["Morning", "Zebra"] class TestSettings: def test_follow_settings_on_the_friend_page(self, qapp, tmp_path): from lintunes.gui.sync_settings_dialog import SyncSettingsDialog from tests.test_round64 import _NoProbe state = CassetteState() state.friends["t"] = Friend(token="t", device_id="D", name="Sam") dialog = SyncSettingsDialog(state, tmp_path, {}, probe=_NoProbe()) page = dialog.friend_pages["t"] assert page.layout_folder.isChecked() and page.mode_separate.isChecked() page.layout_mixed.setChecked(True) page.mode_library.setChecked(True) friend = dialog.state.friends["t"] assert (friend.layout, friend.followed_mode) == ("mixed", "library")