"""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