"""Round 47: LinTunes writes the andTunes library onto the device. The Rabbit's music now lives in one de-duplicated tree, ``Music/andTunes/``, with a ``library.json`` the Android app reads instead of scanning anything. Everything here is testable against tmp_path stand-ins the same way the per-playlist sync is: a fake device root for planning and copying, with the worker driven synchronously via ``_run()`` so progress handlers fire inline. What the tests are really pinning: * a track is stored **once**, however many playlists hold it; * names collide deterministically (order-independent [track_id] suffixes); * the containment guard, because this is the first code that deletes a *directory* on the Rabbit — the old ``Music//`` folders must be structurally unreachable from here; * the index is written **last** and always describes what is really there, including after a cancel. Not covered: the menu actions and the settings dialog are plain Qt in MainWindow, which has no tests anywhere in this suite; verified by hand. """ import json import os import pytest from lintunes import tagging from lintunes.andtunes import art, layout, manifest from lintunes.andtunes.layout import ( DeviceLayoutError, album_key, art_relpath, assert_inside, media_relpath, playlist_relpath, ) from lintunes.andtunes.manifest import build_manifest from lintunes.andtunes.sync import AndTunesSyncWorker, plan_andtunes_sync from lintunes.models import Track from lintunes.models.playlist import Playlist, PlaylistType def _track(tid, name="Song", artist="Band", album="Record", path="", total_time=180_000, **kw): return Track(track_id=tid, name=name, artist=artist, album=album, location=str(path), total_time=total_time, **kw) def _audio(tmp_path, filename, data=b"x" * 100): path = tmp_path / "local" / filename path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) return path def _playlist(name, pid, track_ids): return Playlist(name=name, persistent_id=pid, playlist_type=PlaylistType.REGULAR, track_ids=list(track_ids)) def _library(*tracks): return {t.track_id: t for t in tracks} class TestMediaLayout: def test_artist_album_track_number(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, "Alhambra", "Ahmad Jamal", "Alhambra", src, track_number=4) assert media_relpath(t) == "Media/Ahmad Jamal/Alhambra/04 Alhambra.mp3" def test_album_artist_wins_over_track_artist(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, "Song", "Guest", "Comp", src, album_artist="Host") assert media_relpath(t).startswith("Media/Host/") def test_multi_disc_gets_a_disc_prefix(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, "Song", "Band", "Record", src, track_number=4, disc_number=2, disc_count=2) assert media_relpath(t).endswith("/2-04 Song.mp3") def test_single_disc_has_no_prefix(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, "Song", "Band", "Record", src, track_number=4, disc_number=1, disc_count=1) assert media_relpath(t).endswith("/04 Song.mp3") def test_no_track_number_keeps_just_the_title(self, tmp_path): src = _audio(tmp_path, "x.mp3") assert media_relpath(_track(1, "Song", "Band", "Record", src)).endswith( "/Song.mp3") def test_missing_artist_and_album_fall_back(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, "Song", "", "", src) assert media_relpath(t) == "Media/Unknown Artist/Unknown Album/Song.mp3" def test_forbidden_characters_are_sanitized(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, "A/B", "AC/DC", "Back: In", src) assert media_relpath(t) == "Media/AC DC/Back In/A B.mp3" def test_title_falls_back_to_the_filename_stem(self, tmp_path): src = _audio(tmp_path, "raw_rip.mp3") assert media_relpath(_track(1, "", "", "", src)).endswith("/raw_rip.mp3") class TestAlbumKey: def test_same_album_same_key(self, tmp_path): src = _audio(tmp_path, "x.mp3") a = _track(1, "One", "Band", "Record", src) b = _track(2, "Two", "Band", "Record", src) assert album_key(a) == album_key(b) def test_case_insensitive(self, tmp_path): src = _audio(tmp_path, "x.mp3") a = _track(1, "One", "Band", "Record", src) b = _track(2, "One", "BAND", "record", src) assert album_key(a) == album_key(b) def test_different_albums_differ(self, tmp_path): src = _audio(tmp_path, "x.mp3") a = _track(1, "One", "Band", "Record", src) b = _track(2, "One", "Band", "Other", src) assert album_key(a) != album_key(b) def test_key_is_eight_hex(self, tmp_path): key = album_key(_track(1, path=_audio(tmp_path, "x.mp3"))) assert len(key) == 8 and int(key, 16) >= 0 class TestContainmentGuard: """The blast radius of a bad delete here is somebody's music.""" def test_inside_is_allowed(self, tmp_path): root = tmp_path / "andTunes" assert assert_inside(root, root / "Media" / "a.mp3") def test_the_root_itself_is_refused(self, tmp_path): root = tmp_path / "andTunes" with pytest.raises(DeviceLayoutError): assert_inside(root, root) def test_a_sibling_playlist_folder_is_refused(self, tmp_path): """The old Auxio folders live next door and must be unreachable.""" root = tmp_path / "Music" / "andTunes" with pytest.raises(DeviceLayoutError): assert_inside(root, tmp_path / "Music" / "a fresh master") def test_dot_dot_escape_is_refused(self, tmp_path): root = tmp_path / "andTunes" with pytest.raises(DeviceLayoutError): assert_inside(root, root / ".." / "elsewhere" / "x.mp3") class TestManifest: def test_sparse_rows_omit_empty_fields(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, "Song", "", "", src, total_time=0) row = manifest.track_entry(t, "Media/x.mp3") assert row == {"id": 1, "title": "Song", "path": "Media/x.mp3"} def test_seconds_round_from_milliseconds(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, path=src, total_time=214_400) assert manifest.track_entry(t, "p")["secs"] == 214 def test_art_included_only_when_given(self, tmp_path): src = _audio(tmp_path, "x.mp3") t = _track(1, path=src) assert "art" not in manifest.track_entry(t, "p") assert manifest.track_entry(t, "p", "Art/ab.jpg")["art"] == "Art/ab.jpg" def test_playlist_ids_are_filtered_to_known_tracks(self, tmp_path): src = _audio(tmp_path, "x.mp3") entries = [(_track(1, path=src), "Media/a.mp3", None)] payload = build_manifest(entries, [("Mix", "P1", [1, 99])], generator="lintunes test") assert payload["playlists"][0]["tracks"] == [1] def test_format_version_is_stamped(self, tmp_path): payload = build_manifest([], [], generator="lintunes test") assert payload["format"] == layout.FORMAT_VERSION assert payload["generator"] == "lintunes test" class TestPlan: def test_a_track_in_two_playlists_is_stored_once(self, tmp_path): src = _audio(tmp_path, "a.mp3") track = _track(1, "Song", "Band", "Record", src) plan = plan_andtunes_sync( [_playlist("A", "P1", [1]), _playlist("B", "P2", [1])], _library(track), tmp_path / "dev") assert len(plan.items) == 1 assert len(plan.copies) == 1 assert [p.name for p in plan.playlists] == ["A", "B"] def test_collision_suffixes_are_order_independent(self, tmp_path): # Same artist, album, title and number: only the track id separates them. a = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3")) b = _track(2, "Song", "Band", "Record", _audio(tmp_path, "b.mp3")) forward = plan_andtunes_sync([_playlist("A", "P1", [1, 2])], _library(a, b), tmp_path / "d1") backward = plan_andtunes_sync([_playlist("A", "P1", [2, 1])], _library(a, b), tmp_path / "d2") assert ({i.rel for i in forward.items} == {i.rel for i in backward.items} == {"Media/Band/Record/Song [1].mp3", "Media/Band/Record/Song [2].mp3"}) def test_missing_local_files_are_skipped(self, tmp_path): here = _track(1, "Here", "Band", "Record", _audio(tmp_path, "a.mp3")) gone = _track(2, "Gone", "Band", "Record", tmp_path / "nope.mp3") plan = plan_andtunes_sync([_playlist("A", "P1", [1, 2])], _library(here, gone), tmp_path / "dev") assert plan.skipped == 1 assert [i.track_id for i in plan.items] == [1] assert plan.playlists[0].track_ids == [1] def test_m3u_entries_point_up_into_media(self, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3")) plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), tmp_path / "dev") assert plan.playlists[0].rel == "Playlists/A.m3u" assert plan.playlists[0].entries[0][0] == "../Media/Band/Record/Song.mp3" def test_matching_files_are_kept_and_extras_go_stale(self, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3", b"y" * 40)) root = tmp_path / "dev" dest = root / "Media" / "Band" / "Record" dest.mkdir(parents=True) (dest / "Song.mp3").write_bytes(b"y" * 40) (root / "Media" / "Old" ).mkdir(parents=True) (root / "Media" / "Old" / "gone.mp3").write_bytes(b"z" * 10) plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), root) assert plan.kept == 1 and plan.copies == [] assert plan.stale == ["Media/Old/gone.mp3"] assert plan.bytes_freed == 10 def test_a_size_mismatch_is_recopied(self, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3", b"y" * 40)) root = tmp_path / "dev" dest = root / "Media" / "Band" / "Record" dest.mkdir(parents=True) (dest / "Song.mp3").write_bytes(b"y" * 39) # a partial from last time plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), root) assert plan.kept == 0 and len(plan.copies) == 1 def test_the_devices_own_folders_are_never_scanned_or_pruned(self, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3")) root = tmp_path / "dev" for owned in (layout.BUTTONS_DIR, layout.PLAYS_DIR): (root / owned).mkdir(parents=True) (root / owned / "mine.dat").write_bytes(b"!") plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), root) assert plan.stale == [] def test_the_index_and_art_are_never_stale(self, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3")) root = tmp_path / "dev" root.mkdir() (root / layout.MANIFEST_NAME).write_text("{}") (root / layout.ART_DIR).mkdir() (root / art_relpath(album_key(track))).write_bytes(b"jpeg") plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), root) assert plan.stale == [] def _run(worker): """Drive the worker synchronously so signals fire inline.""" worker._run() def _cancel_on_emit(worker, n: int): """Cancel once the nth progress signal has fired. Each 100-byte fixture is one chunk, so it emits exactly once — n is effectively "cancel while copying file n", the same trick test_round28 uses on the per-playlist worker. """ seen = [] def _tick(*_args): seen.append(1) if len(seen) >= n: worker.cancel() worker.progress.connect(_tick) return worker class TestWorker: def test_end_to_end(self, qapp, tmp_path): a = _track(1, "One", "Band", "Record", _audio(tmp_path, "a.mp3"), track_number=1) b = _track(2, "Two", "Band", "Record", _audio(tmp_path, "b.mp3"), track_number=2) root = tmp_path / "dev" plan = plan_andtunes_sync([_playlist("Mix", "P1", [2, 1])], _library(a, b), root, copy_art=False) done = [] worker = AndTunesSyncWorker(plan) worker.finished.connect(done.append) _run(worker) assert (root / "Media/Band/Record/01 One.mp3").read_bytes() == b"x" * 100 assert (root / "Media/Band/Record/02 Two.mp3").exists() m3u = (root / "Playlists/Mix.m3u").read_text() assert m3u.splitlines()[0] == "#EXTM3U" # Playlist order, not track-number order, and pointing up into Media/. assert "../Media/Band/Record/02 Two.mp3" in m3u.splitlines()[2] payload = json.loads((root / layout.MANIFEST_NAME).read_text()) assert [t["id"] for t in payload["tracks"]] == [1, 2] assert payload["playlists"][0] == {"id": "P1", "name": "Mix", "tracks": [2, 1]} assert done[0]["tracks"] == 2 and done[0]["copied"] == 2 def test_progress_is_reported_in_kib(self, qapp, tmp_path): big = _audio(tmp_path, "big.mp3", b"x" * (2 * 1024 * 1024)) track = _track(1, "Big", "Band", "Record", big) plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), tmp_path / "dev", copy_art=False) seen = [] worker = AndTunesSyncWorker(plan) worker.progress.connect(lambda d, t, label: seen.append((d, t))) _run(worker) assert seen[-1][0] == 2048 and seen[0][1] == 2048 def test_stale_files_are_removed_and_empty_folders_pruned(self, qapp, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3")) root = tmp_path / "dev" (root / "Media" / "Old Band" / "Old Record").mkdir(parents=True) (root / "Media/Old Band/Old Record/gone.mp3").write_bytes(b"z") plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), root, copy_art=False) _run(AndTunesSyncWorker(plan)) assert not (root / "Media" / "Old Band").exists() assert (root / "Media/Band/Record/Song.mp3").exists() def test_the_devices_own_folders_survive_a_sync(self, qapp, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3")) root = tmp_path / "dev" (root / layout.BUTTONS_DIR).mkdir(parents=True) (root / layout.BUTTONS_DIR / "playlists.png").write_bytes(b"png") (root / layout.PLAYS_DIR).mkdir(parents=True) # empty on purpose plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), root, copy_art=False) _run(AndTunesSyncWorker(plan)) assert (root / layout.BUTTONS_DIR / "playlists.png").exists() assert (root / layout.PLAYS_DIR).is_dir() def test_a_stale_entry_pointing_outside_the_root_is_refused( self, qapp, tmp_path): """Belt and braces: the guard holds even if planning were wrong.""" root = tmp_path / "dev" / "andTunes" root.mkdir(parents=True) neighbour = tmp_path / "dev" / "a fresh master" neighbour.mkdir() (neighbour / "song.mp3").write_bytes(b"precious") plan = plan_andtunes_sync([], {}, root, copy_art=False) plan.stale = ["../a fresh master/song.mp3"] _run(AndTunesSyncWorker(plan)) assert (neighbour / "song.mp3").read_bytes() == b"precious" def test_a_source_that_vanished_is_skipped_not_fatal(self, qapp, tmp_path): here = _track(1, "Here", "Band", "Record", _audio(tmp_path, "a.mp3")) gone = _track(2, "Gone", "Band", "Record", _audio(tmp_path, "b.mp3")) root = tmp_path / "dev" plan = plan_andtunes_sync([_playlist("A", "P1", [1, 2])], _library(here, gone), root, copy_art=False) os.unlink(gone.location) # Syncthing moved it under us failures = [] worker = AndTunesSyncWorker(plan) worker.failed.connect(failures.append) _run(worker) assert not failures payload = json.loads((root / layout.MANIFEST_NAME).read_text()) assert [t["id"] for t in payload["tracks"]] == [1] assert "Gone" not in (root / "Playlists/A.m3u").read_text() def test_device_side_error_reports_failure(self, qapp, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3")) blocked = tmp_path / "dev" blocked.write_text("not a directory") # mkdir will raise plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), blocked, copy_art=False) failures = [] worker = AndTunesSyncWorker(plan) worker.failed.connect(failures.append) _run(worker) assert failures and "still connected" in failures[0] class TestCancel: def test_cancel_before_the_run_copies_nothing_and_still_indexes( self, qapp, tmp_path): track = _track(1, "Song", "Band", "Record", _audio(tmp_path, "a.mp3")) root = tmp_path / "dev" plan = plan_andtunes_sync([_playlist("A", "P1", [1])], _library(track), root, copy_art=False) worker = AndTunesSyncWorker(plan) stopped = [] worker.cancelled.connect(stopped.append) worker.cancel() _run(worker) assert stopped and stopped[0]["copied"] == 0 assert not (root / "Media").exists() def test_cancel_during_the_second_file_keeps_the_first(self, qapp, tmp_path): a = _track(1, "One", "Band", "Record", _audio(tmp_path, "a.mp3"), track_number=1) b = _track(2, "Two", "Band", "Record", _audio(tmp_path, "b.mp3"), track_number=2) root = tmp_path / "dev" plan = plan_andtunes_sync([_playlist("A", "P1", [1, 2])], _library(a, b), root, copy_art=False) # Cancel while the second file is in flight; the first must survive. _run(_cancel_on_emit(AndTunesSyncWorker(plan), 2)) assert (root / "Media/Band/Record/01 One.mp3").exists() assert not (root / "Media/Band/Record/02 Two.mp3").exists() def test_a_cancelled_sync_leaves_a_coherent_index(self, qapp, tmp_path): """The app must never open a manifest naming files that aren't there.""" a = _track(1, "One", "Band", "Record", _audio(tmp_path, "a.mp3"), track_number=1) b = _track(2, "Two", "Band", "Record", _audio(tmp_path, "b.mp3"), track_number=2) root = tmp_path / "dev" plan = plan_andtunes_sync([_playlist("A", "P1", [1, 2])], _library(a, b), root, copy_art=False) _run(_cancel_on_emit(AndTunesSyncWorker(plan), 2)) payload = json.loads((root / layout.MANIFEST_NAME).read_text()) for row in payload["tracks"]: assert (root / row["path"]).exists() assert payload["playlists"][0]["tracks"] == [1] assert "02 Two.mp3" not in (root / "Playlists/A.m3u").read_text() def test_resyncing_after_a_cancel_completes_the_library(self, qapp, tmp_path): a = _track(1, "One", "Band", "Record", _audio(tmp_path, "a.mp3"), track_number=1) b = _track(2, "Two", "Band", "Record", _audio(tmp_path, "b.mp3"), track_number=2) root = tmp_path / "dev" lib, pls = _library(a, b), [_playlist("A", "P1", [1, 2])] _run(_cancel_on_emit(AndTunesSyncWorker( plan_andtunes_sync(pls, lib, root, copy_art=False)), 2)) _run(AndTunesSyncWorker( plan_andtunes_sync(pls, lib, root, copy_art=False))) payload = json.loads((root / layout.MANIFEST_NAME).read_text()) assert [t["id"] for t in payload["tracks"]] == [1, 2] assert (root / "Media/Band/Record/02 Two.mp3").exists() class TestAlbumArt: def test_art_is_extracted_scaled_and_shared_across_an_album( self, qapp, tmp_path, mp3_file, jpeg_bytes, monkeypatch): monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache")) tagging.write_artwork(mp3_file, jpeg_bytes) second = tmp_path / "local" / "b.mp3" second.parent.mkdir(parents=True, exist_ok=True) second.write_bytes(mp3_file.read_bytes()) a = _track(1, "One", "Band", "Record", mp3_file, track_number=1) b = _track(2, "Two", "Band", "Record", second, track_number=2) root = tmp_path / "dev" plan = plan_andtunes_sync([_playlist("A", "P1", [1, 2])], _library(a, b), root) summary = [] worker = AndTunesSyncWorker(plan) worker.finished.connect(summary.append) _run(worker) rel = art_relpath(album_key(a)) assert (root / rel).exists() assert summary[0]["albums"] == 1 # one cover, not two payload = json.loads((root / layout.MANIFEST_NAME).read_text()) assert {t["art"] for t in payload["tracks"]} == {rel} def test_a_track_without_art_gets_no_art_key(self, qapp, tmp_path, mp3_file, monkeypatch): monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache")) track = _track(1, "Song", "Band", "Record", mp3_file) root = tmp_path / "dev" _run(AndTunesSyncWorker(plan_andtunes_sync( [_playlist("A", "P1", [1])], _library(track), root))) payload = json.loads((root / layout.MANIFEST_NAME).read_text()) assert "art" not in payload["tracks"][0] assert not (root / art_relpath(album_key(track))).exists() def test_render_never_upscales_a_small_cover(self, qapp, jpeg_bytes): from PyQt6.QtGui import QImage out = art.render_jpeg(jpeg_bytes) image = QImage() image.loadFromData(out) assert image.width() == 8 def test_the_cache_is_reused_until_the_file_changes( self, qapp, tmp_path, mp3_file, jpeg_bytes, monkeypatch): monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache")) tagging.write_artwork(mp3_file, jpeg_bytes) track = _track(1, "Song", "Band", "Record", mp3_file) assert art.album_art_jpeg(track) calls = [] real = tagging.read_embedded_artwork monkeypatch.setattr(art.tagging, "read_embedded_artwork", lambda p: calls.append(p) or real(p)) assert art.album_art_jpeg(track) assert calls == [] # served from cache os.utime(mp3_file, (9e9, 9e9)) # the file changed under us assert art.album_art_jpeg(track) assert calls # so the cover was re-read class TestSyncablePlaylists: def test_folders_and_system_playlists_are_not_offered(self): from lintunes.gui.device_sync_dialog import syncable_playlists class _Lib: playlists = { "P1": Playlist(name="Mix", persistent_id="P1"), "F1": Playlist(name="Folder", persistent_id="F1", playlist_type=PlaylistType.FOLDER), "S1": Playlist(name="Music", persistent_id="S1", is_system=True), "M1": Playlist(name="Recent", persistent_id="M1", playlist_type=PlaylistType.SMART), } assert [p.persistent_id for p in syncable_playlists(_Lib())] == ["P1", "M1"]