trav
a0a1aea57e
Quitting mid-transfer now asks (Keep Syncing / Quit Anyway) from closeEvent, which every quit path hits — including the self-update restart, which previously bypassed closeEvent via a bare quit() and now routes through close(). A running sync holds its own GNOME inhibitor (suspend + logout, "Syncing a playlist to a device") so the machine won't sleep or shutdown/restart under a copy. Along the way: the Inhibit D-Bus call marshaled Python ints as signed against GNOME's (susu) signature, so every call was rejected and the playback sleep inhibitor had silently never worked — _uint fixes both holders (verified live: cookies taken, IsInhibited flips, releases clean). Also investigated trav's mid-copy Syncthing question: the transfer works from a click-time snapshot and never reads live library state, so remote changes can't corrupt it — no state freeze needed. Hardened the one real gap: a source file relocated under the queue (remote metadata edit moving files) is now skipped, dropped from the m3u, and reported, instead of aborting the whole sync.
226 lines
9.2 KiB
Python
226 lines
9.2 KiB
Python
"""Round 26: Device menu — sync a playlist to the Rabbit R1.
|
|
|
|
The Rabbit connects over MTP/gvfs (a FUSE path), so all sync logic lives in
|
|
device_sync.py as plain file I/O testable against tmp_path stand-ins: a fake
|
|
gvfs tree for detection, a fake device Music dir for planning and copying.
|
|
The diff is name+size (MTP mtimes are unreliable), collision suffixes are
|
|
order-independent, and only files inside the playlist's own folder are ever
|
|
deleted.
|
|
"""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from lintunes import device_sync
|
|
from lintunes.device_sync import (
|
|
DeviceSyncWorker, build_m3u, find_rabbit, plan_sync, sanitize_name,
|
|
track_filename,
|
|
)
|
|
from lintunes.models import Track
|
|
|
|
|
|
def _track(tid, name, artist, path, total_time=180_000):
|
|
return Track(track_id=tid, name=name, artist=artist,
|
|
location=str(path), total_time=total_time)
|
|
|
|
|
|
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
|
|
|
|
|
|
class TestNames:
|
|
def test_sanitize_strips_forbidden_chars(self):
|
|
assert sanitize_name('Road / Trip: "mix"?') == "Road Trip mix"
|
|
|
|
def test_sanitize_trims_dots_and_spaces(self):
|
|
assert sanitize_name(" ..hits.. ") == "hits"
|
|
|
|
def test_sanitize_empty_falls_back(self):
|
|
assert sanitize_name("???") == "Untitled"
|
|
|
|
def test_track_filename_artist_and_title(self, tmp_path):
|
|
src = _audio(tmp_path, "whatever.mp3")
|
|
track = _track(1, "Song", "Band", src)
|
|
assert track_filename(track) == "Band - Song.mp3"
|
|
|
|
def test_track_filename_no_artist(self, tmp_path):
|
|
src = _audio(tmp_path, "whatever.m4a")
|
|
assert track_filename(_track(1, "Song", "", src)) == "Song.m4a"
|
|
|
|
def test_track_filename_falls_back_to_stem(self, tmp_path):
|
|
src = _audio(tmp_path, "raw_rip.mp3")
|
|
assert track_filename(_track(1, "", "", src)) == "raw_rip.mp3"
|
|
|
|
|
|
class TestPlan:
|
|
def test_fresh_sync_copies_everything(self, tmp_path):
|
|
a = _audio(tmp_path, "a.mp3", b"a" * 50)
|
|
b = _audio(tmp_path, "b.mp3", b"b" * 70)
|
|
tracks = [_track(1, "A", "X", a), _track(2, "B", "Y", b)]
|
|
plan = plan_sync("Mix", tracks, tmp_path / "Music")
|
|
|
|
assert plan.dest_dir == tmp_path / "Music" / "Mix"
|
|
assert plan.m3u_name == "Mix.m3u"
|
|
assert [i.dest_name for i in plan.copies] == ["X - A.mp3", "Y - B.mp3"]
|
|
assert plan.bytes_to_copy == 120
|
|
assert plan.kept == 0 and not plan.stale and plan.skipped == 0
|
|
|
|
def test_incremental_keeps_matching_and_deletes_stale(self, tmp_path):
|
|
a = _audio(tmp_path, "a.mp3", b"a" * 50)
|
|
b = _audio(tmp_path, "b.mp3", b"b" * 70)
|
|
dest = tmp_path / "Music" / "Mix"
|
|
dest.mkdir(parents=True)
|
|
(dest / "X - A.mp3").write_bytes(b"a" * 50) # same size -> kept
|
|
(dest / "Y - B.mp3").write_bytes(b"old") # size mismatch -> recopy
|
|
(dest / "Gone - Track.mp3").write_bytes(b"z" * 30) # stale
|
|
(dest / "Old Name.m3u").write_bytes(b"#EXTM3U\n") # stale old m3u
|
|
(dest / "subdir").mkdir() # never touched
|
|
|
|
tracks = [_track(1, "A", "X", a), _track(2, "B", "Y", b)]
|
|
plan = plan_sync("Mix", tracks, tmp_path / "Music")
|
|
|
|
assert plan.kept == 1
|
|
assert [i.dest_name for i in plan.copies] == ["Y - B.mp3"]
|
|
assert sorted(plan.stale) == ["Gone - Track.mp3", "Old Name.m3u"]
|
|
assert plan.bytes_freed == 30 + len(b"#EXTM3U\n")
|
|
|
|
def test_collision_suffixes_are_order_independent(self, tmp_path):
|
|
a = _audio(tmp_path, "a.mp3")
|
|
b = _audio(tmp_path, "b.mp3")
|
|
t1, t2 = _track(1, "Same", "Band", a), _track(2, "Same", "Band", b)
|
|
|
|
names = {i.track_id: i.dest_name
|
|
for i in plan_sync("Mix", [t1, t2], tmp_path / "Music").copies}
|
|
reversed_names = {i.track_id: i.dest_name
|
|
for i in plan_sync("Mix", [t2, t1],
|
|
tmp_path / "Music").copies}
|
|
assert names == reversed_names
|
|
assert names[1] == "Band - Same [1].mp3"
|
|
assert names[2] == "Band - Same [2].mp3"
|
|
|
|
def test_missing_locations_are_skipped(self, tmp_path):
|
|
real = _audio(tmp_path, "real.mp3")
|
|
tracks = [
|
|
_track(1, "Real", "X", real),
|
|
_track(2, "No file", "X", tmp_path / "nope.mp3"),
|
|
Track(track_id=3, name="No location", location=""),
|
|
]
|
|
plan = plan_sync("Mix", tracks, tmp_path / "Music")
|
|
assert plan.skipped == 2
|
|
assert len(plan.copies) == 1 and len(plan.entries) == 1
|
|
|
|
def test_duplicate_track_ids_copy_once_listed_twice(self, tmp_path):
|
|
a = _audio(tmp_path, "a.mp3")
|
|
track = _track(1, "A", "X", a)
|
|
plan = plan_sync("Mix", [track, track], tmp_path / "Music")
|
|
assert len(plan.copies) == 1
|
|
assert [e[0] for e in plan.entries] == ["X - A.mp3", "X - A.mp3"]
|
|
|
|
def test_folder_and_m3u_use_sanitized_name(self, tmp_path):
|
|
plan = plan_sync("Road / Trip?", [], tmp_path / "Music")
|
|
assert plan.dest_dir.name == "Road Trip"
|
|
assert plan.m3u_name == "Road Trip.m3u"
|
|
|
|
|
|
class TestM3u:
|
|
def test_build_m3u(self):
|
|
text = build_m3u([("X - A.mp3", 181, "X - A"), ("Y - B.mp3", 45, "Y - B")])
|
|
assert text == ("#EXTM3U\n"
|
|
"#EXTINF:181,X - A\nX - A.mp3\n"
|
|
"#EXTINF:45,Y - B\nY - B.mp3\n")
|
|
|
|
def test_entry_seconds_round_from_ms(self, tmp_path):
|
|
a = _audio(tmp_path, "a.mp3")
|
|
plan = plan_sync("Mix", [_track(1, "A", "X", a, total_time=181_499)],
|
|
tmp_path / "Music")
|
|
assert plan.entries[0][1] == 181
|
|
|
|
|
|
class TestWorker:
|
|
def _synced(self, qapp, plan):
|
|
worker = DeviceSyncWorker(plan)
|
|
results = {}
|
|
worker.finished.connect(lambda s: results.update(s))
|
|
worker.failed.connect(lambda m: results.update(error=m))
|
|
worker._run() # synchronously on the test thread; signals fire directly
|
|
return results
|
|
|
|
def test_end_to_end(self, qapp, tmp_path):
|
|
a = _audio(tmp_path, "a.mp3", os.urandom(3 * 1024) )
|
|
b = _audio(tmp_path, "b.mp3", b"b" * 70)
|
|
dest = tmp_path / "Music" / "Mix"
|
|
dest.mkdir(parents=True)
|
|
(dest / "Stale.mp3").write_bytes(b"old")
|
|
tracks = [_track(1, "A", "X", a), _track(2, "B", "Y", b)]
|
|
plan = plan_sync("Mix", tracks, tmp_path / "Music")
|
|
|
|
summary = self._synced(qapp, plan)
|
|
|
|
assert summary["copied"] == 2 and summary["removed"] == 1
|
|
assert (dest / "X - A.mp3").read_bytes() == a.read_bytes()
|
|
assert (dest / "Y - B.mp3").read_bytes() == b.read_bytes()
|
|
assert not (dest / "Stale.mp3").exists()
|
|
assert (dest / "Mix.m3u").read_text(encoding="utf-8") == build_m3u(
|
|
plan.entries)
|
|
|
|
def test_progress_reports_kib(self, qapp, tmp_path):
|
|
a = _audio(tmp_path, "a.mp3", b"a" * (2 * device_sync.CHUNK))
|
|
plan = plan_sync("Mix", [_track(1, "A", "X", a)], tmp_path / "Music")
|
|
worker = DeviceSyncWorker(plan)
|
|
seen = []
|
|
worker.progress.connect(lambda d, t, l: seen.append((d, t, l)))
|
|
worker._run()
|
|
assert seen[-1][0] == seen[-1][1] == 2 * device_sync.CHUNK // 1024
|
|
assert "1/1" in seen[-1][2]
|
|
|
|
def test_device_side_error_reports_failure(self, qapp, tmp_path):
|
|
# A file squatting on the dest-dir path stands in for a device-side
|
|
# write error (unplug mid-sync). Vanished *sources* are tolerated
|
|
# (round 27); device errors still abort.
|
|
a = _audio(tmp_path, "a.mp3")
|
|
plan = plan_sync("Mix", [_track(1, "A", "X", a)], tmp_path / "Music")
|
|
(tmp_path / "Music").mkdir()
|
|
(tmp_path / "Music" / "Mix").write_bytes(b"not a dir")
|
|
result = self._synced(qapp, plan)
|
|
assert "Sync failed" in result["error"]
|
|
|
|
def test_empty_playlist_clears_folder(self, qapp, tmp_path):
|
|
dest = tmp_path / "Music" / "Mix"
|
|
dest.mkdir(parents=True)
|
|
(dest / "Old - Song.mp3").write_bytes(b"x")
|
|
plan = plan_sync("Mix", [], tmp_path / "Music")
|
|
summary = self._synced(qapp, plan)
|
|
assert summary["removed"] == 1
|
|
assert sorted(p.name for p in dest.iterdir()) == ["Mix.m3u"]
|
|
assert (dest / "Mix.m3u").read_text() == "#EXTM3U\n"
|
|
|
|
|
|
class TestFindRabbit:
|
|
def _gvfs(self, tmp_path, host):
|
|
storage = tmp_path / "gvfs" / host / "Internal shared storage"
|
|
(storage / "Music").mkdir(parents=True)
|
|
return tmp_path / "gvfs"
|
|
|
|
def test_finds_rabbit_mount(self, tmp_path):
|
|
gvfs = self._gvfs(tmp_path, "mtp:host=unknown_Rabbit_R1_919109A4")
|
|
device = find_rabbit(gvfs)
|
|
assert device is not None
|
|
assert device.name == "Rabbit R1"
|
|
assert device.music_dir == (gvfs / "mtp:host=unknown_Rabbit_R1_919109A4"
|
|
/ "Internal shared storage" / "Music")
|
|
|
|
def test_match_is_case_insensitive(self, tmp_path):
|
|
gvfs = self._gvfs(tmp_path, "mtp:host=RABBIT_r1_abc")
|
|
assert find_rabbit(gvfs) is not None
|
|
|
|
def test_ignores_other_mtp_devices(self, tmp_path):
|
|
gvfs = self._gvfs(tmp_path, "mtp:host=Google_Pixel_8_xyz")
|
|
assert find_rabbit(gvfs) is None
|
|
|
|
def test_no_gvfs_dir_is_none(self, tmp_path):
|
|
assert find_rabbit(tmp_path / "does-not-exist") is None
|