v0.17.0: the desktop already knows what the songs are
The first half of andTunes, a music player for the Rabbit R1. Auxio re-reads Android's MediaStore on every launch, which is the "your songs will show up here" hang — so the answer isn't a faster scanner, it's not scanning at all. LinTunes already knows the artist, album and year of every file it just copied, so it writes them into Music/andTunes/library.json and the app reads one file instead of indexing a filesystem. New lintunes/andtunes/: layout.py (a de-duplicated Media/<Artist>/<Album>/ tree, so a song in three playlists is stored once), manifest.py (pure, Qt-free), art.py (one 480 px cover per album via QImage, cached and invalidated by the audio file's mtime — which is exactly what embedding new art moves), and sync.py, a third sibling of plan_export/ExportWorker. Three rules the code depends on: planning never renders art (a mutagen open per album, and planning runs on the GUI thread); the index is written last and after a cancel lists only tracks whose files actually landed; and every delete goes through layout.assert_inside, which is why the old Music/<Playlist>/ folders are structurally unreachable rather than merely un-referenced. Connections gains "Sync to Rabbit" and "Rabbit Sync Settings…" — the ticked set is the device's contents, so unticking is how a playlist comes off, which nothing could do before. The selection rides preferences.json so both machines agree. The old per-playlist sync stays, renamed "(Auxio)": removing it now would leave the device full of files and nothing able to open them until the app exists. The app itself is rounds 48-50; its board is andtunes/TASKS.md, including the measured screen facts (480x640 px at density override 160 — one dp is half its usual physical size on a 2.88" panel). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015wrTys5U1fLb4pBD2LKWzV
This commit is contained in:
@@ -0,0 +1,537 @@
|
||||
"""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/<Playlist>/`` 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"]
|
||||
Reference in New Issue
Block a user