diff --git a/CLAUDE.md b/CLAUDE.md index 7f4111a..7e219ce 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -97,6 +97,14 @@ persistence) → GUI (Qt widgets that read the manager and connect to its signal so the desktop's media keys / now-playing popup control playback. Spacebar and arrow keys are handled locally via `MainWindow.eventFilter`. +- **`lintunes/device_sync.py`** — one-way playlist sync to the Rabbit R1 (Device + menu). The Rabbit mounts via **MTP/gvfs** (a FUSE path under + `/run/user//gvfs`), not mass storage — so plain file I/O, but never + copystat and never trust mtimes (diff by name+size). Sync owns exactly + `Music//` on the device (creates/overwrites/deletes there, + plus an Auxio-importable `.m3u`); it never deletes outside that folder and + only ever *reads* local library files. + ## Conventions & gotchas - **Tests are organized as `tests/test_roundN.py`** — each development round adds diff --git a/TASKS.md b/TASKS.md index d552718..fa74571 100644 --- a/TASKS.md +++ b/TASKS.md @@ -3,6 +3,40 @@ Legend: `[ ]` todo · `[~]` in progress · `[x]` done. When a round closes, move its finished items to `tasks-done.md`. + + +# tasks + +- [ ] ability to delete songs from library +- [ ] archive the done tasks in here to another file, this is crufty.... + + + + + + + + + + + + + +# old + +## Round 26 — Device menu: sync playlist to Rabbit R1 (v0.2.0) + +- [x] Device menu (left of Track) with "Sync Playlist to Rabbit"; grayed out on + the Library view, while syncing, or when no Rabbit is plugged in. +- [x] `device_sync.py`: MTP/gvfs detection (the Rabbit mounts via MTP, not mass + storage), name+size incremental diff, stale-file deletion scoped to the + playlist's own folder, Auxio-importable `.m3u`, chunked-copy worker thread. +- [x] Up-front free-space check with a needed-vs-available alert. +- [x] Right-justified sync progress bar in the status bar (version left, + totals middle, sync right). + +Tests in `tests/test_round26.py`. Feature round → minor bump **0.2.0**. + ## Round 25 — folder highlight while dragging a playlist (v0.1.6) Tests in `tests/test_round25.py`. Fix round → patch bump **0.1.6**. diff --git a/lintunes/__init__.py b/lintunes/__init__.py index a7ab430..672967b 100644 --- a/lintunes/__init__.py +++ b/lintunes/__init__.py @@ -1,3 +1,3 @@ """LinTunes — iTunes-style music library manager and player for Linux.""" -__version__ = "0.1.6" +__version__ = "0.2.0" diff --git a/lintunes/device_sync.py b/lintunes/device_sync.py new file mode 100644 index 0000000..f1cde9b --- /dev/null +++ b/lintunes/device_sync.py @@ -0,0 +1,272 @@ +"""One-way playlist sync to a USB-connected media player (Rabbit R1). + +The Rabbit mounts over MTP, which gvfs exposes as a FUSE directory under +/run/user//gvfs — so plain file I/O works, with two MTP caveats honored +throughout: never copy metadata (copystat raises EPERM on gvfs-MTP) and never +trust mtimes (compare files by name + size instead). + +Contract: sync is strictly lintunes → device. LinTunes owns exactly +``Music//`` on the device — files there are created, replaced, +and deleted freely to mirror the playlist, but nothing outside that folder is +ever touched (sibling playlist folders, Android's ``.thumbnails``), and local +library files are only ever read. A playlist renamed in lintunes syncs to a +fresh folder; the old one is left for manual cleanup. Alongside the audio +files an ``.m3u`` (UTF-8, filenames relative to its own folder) carries the +playlist order — the format Auxio on the Rabbit imports. +""" + +import os +import re +import threading +from dataclasses import dataclass, field +from pathlib import Path + +from PyQt6.QtCore import QObject, pyqtSignal + +CHUNK = 1024 * 1024 # copy unit; also the progress granularity +SPACE_MARGIN = 16 * 1024 * 1024 # headroom the free-space check insists on + +# Characters Android/FAT-style storage can't take in a filename. +_FORBIDDEN = re.compile(r'[\\/:*?"<>|\x00-\x1f]') + + +@dataclass +class Device: + name: str # human name for UI strings, e.g. "Rabbit R1" + root: Path # storage root ("Internal shared storage") — disk_usage target + music_dir: Path # root / "Music" + + +def find_rabbit(gvfs_root: Path | None = None) -> Device | None: + """The connected Rabbit R1's storage, or None when it isn't plugged in.""" + if gvfs_root is None: + gvfs_root = Path(f"/run/user/{os.getuid()}/gvfs") + try: + mounts = list(gvfs_root.iterdir()) + except OSError: + return None + for mount in mounts: + if not mount.name.startswith("mtp:host=") or "rabbit" not in mount.name.lower(): + continue + try: + # The single MTP storage volume ("Internal shared storage"). + storage = next((c for c in sorted(mount.iterdir()) if c.is_dir()), None) + except OSError: + continue + if storage is not None: + return Device("Rabbit R1", storage, storage / "Music") + return None + + +def find_device() -> Device | None: + """The first supported connected device (only the Rabbit for now).""" + return find_rabbit() + + +def sanitize_name(name: str) -> str: + """A playlist/track name reduced to a safe cross-filesystem filename.""" + cleaned = " ".join(_FORBIDDEN.sub(" ", name).split()) + cleaned = cleaned.strip(". ")[:150].strip(". ") + return cleaned or "Untitled" + + +def track_display(track) -> str: + """The 'Artist - Title' line shown in the m3u and progress text.""" + name = track.name or Path(track.location).stem or f"Track {track.track_id}" + return f"{track.artist} - {name}" if track.artist else name + + +def track_filename(track) -> str: + ext = Path(track.location).suffix + return sanitize_name(track_display(track)) + ext + + +def format_bytes(n: int) -> str: + for unit in ("B", "KB", "MB", "GB"): + if n < 1024 or unit == "GB": + return f"{n:.1f} {unit}" if unit != "B" else f"{n} B" + n /= 1024 + return f"{n:.1f} GB" + + +@dataclass +class SyncItem: + track_id: int + src: Path + dest_name: str + size: int + + +@dataclass +class SyncPlan: + device: Device + playlist_name: str + dest_dir: Path + m3u_name: str + copies: list = field(default_factory=list) # SyncItems to (re)copy + kept: int = 0 # already on device, same size + stale: list = field(default_factory=list) # filenames in dest_dir to delete + skipped: int = 0 # tracks with no usable local file + entries: list = field(default_factory=list) # (dest_name, secs, display) m3u order + bytes_to_copy: int = 0 + bytes_freed: int = 0 + + +def plan_sync(playlist_name: str, tracks: list, music_dir: Path, + device: Device | None = None) -> SyncPlan: + """Diff the playlist against what's already in its device folder. + + Pure planning — nothing is created, copied, or deleted here. Existing + device files matching by name + size are kept; everything else in the + folder (old m3us included) is marked stale. Subdirectories are never + touched. + """ + folder = sanitize_name(playlist_name) + plan = SyncPlan(device=device, playlist_name=playlist_name, + dest_dir=music_dir / folder, m3u_name=folder + ".m3u") + + # Resolve each unique track to a real local file. + items: dict[int, SyncItem] = {} + usable: list = [] + seen_skipped: set[int] = set() + for track in tracks: + if track.track_id in items: + usable.append(track) + continue + if track.track_id in seen_skipped: + continue + src = Path(track.location) if track.location else None + if src is None or not src.is_file(): + seen_skipped.add(track.track_id) + continue + items[track.track_id] = SyncItem( + track.track_id, src, track_filename(track), src.stat().st_size) + usable.append(track) + plan.skipped = len(seen_skipped) + + # Disambiguate name collisions with a stable [track_id] suffix. Every + # member of a colliding group gets the suffix (not just the "second" one) + # so names never depend on playlist order or membership history. + by_name: dict[str, list[SyncItem]] = {} + for item in items.values(): + by_name.setdefault(item.dest_name, []).append(item) + for group in by_name.values(): + if len(group) > 1: + for item in group: + stem, ext = os.path.splitext(item.dest_name) + item.dest_name = f"{stem} [{item.track_id}]{ext}" + + for track in usable: + item = items[track.track_id] + plan.entries.append( + (item.dest_name, round(track.total_time / 1000), track_display(track))) + + # Diff against the device folder by name + size (mtimes lie over MTP). + on_device: dict[str, int] = {} + try: + for entry in os.scandir(plan.dest_dir): + if entry.is_file(follow_symlinks=False): + on_device[entry.name] = entry.stat().st_size + except FileNotFoundError: + pass + + expected = {item.dest_name for item in items.values()} | {plan.m3u_name} + for item in items.values(): + if on_device.get(item.dest_name) == item.size: + plan.kept += 1 + else: + plan.copies.append(item) + plan.bytes_to_copy += item.size + for name, size in on_device.items(): + if name not in expected: + plan.stale.append(name) + plan.bytes_freed += size + return plan + + +def build_m3u(entries: list) -> str: + lines = ["#EXTM3U"] + for dest_name, secs, display in entries: + lines.append(f"#EXTINF:{secs},{display}") + lines.append(dest_name) + return "\n".join(lines) + "\n" + + +class DeviceSyncWorker(QObject): + """Runs a SyncPlan on a daemon thread, reporting through signals. + + Progress is emitted in KiB (a pyqtSignal(int) is a C int — byte counts + overflow past 2 GiB). An unplug mid-copy surfaces as `failed`; the next + sync self-heals because the partial file loses the name+size diff. + """ + + progress = pyqtSignal(int, int, str) # done_kib, total_kib, "12/240 Artist - Title" + finished = pyqtSignal(dict) + failed = pyqtSignal(str) + + def __init__(self, plan: SyncPlan, parent=None): + super().__init__(parent) + self._plan = plan + self._busy = False + + def busy(self) -> bool: + return self._busy + + def start(self): + if self._busy: + return + self._busy = True + threading.Thread(target=self._run_guarded, daemon=True).start() + + def _run_guarded(self): + try: + self._run() + finally: + self._busy = False + + def _run(self): + plan = self._plan + dest = None + try: + plan.dest_dir.mkdir(parents=True, exist_ok=True) + removed = 0 + for name in plan.stale: + try: + (plan.dest_dir / name).unlink() + removed += 1 + except OSError: + pass # one stubborn file shouldn't kill the sync + total_kib = max(plan.bytes_to_copy // 1024, 1) + done = 0 + for i, item in enumerate(plan.copies, start=1): + label = f"{i}/{len(plan.copies)} {Path(item.dest_name).stem}" + dest = plan.dest_dir / item.dest_name + # Manual chunked copy: byte-accurate progress, and no + # copystat (gvfs-MTP rejects it). + with open(item.src, "rb") as fsrc, open(dest, "wb") as fdst: + while True: + chunk = fsrc.read(CHUNK) + if not chunk: + break + fdst.write(chunk) + done += len(chunk) + self.progress.emit(done // 1024, total_kib, label) + dest = None + # Written last so an interrupted sync leaves the old m3u intact. + (plan.dest_dir / plan.m3u_name).write_text( + build_m3u(plan.entries), encoding="utf-8") + device_name = plan.device.name if plan.device else "device" + self.finished.emit({ + "playlist": plan.playlist_name, "device": device_name, + "copied": len(plan.copies), "kept": plan.kept, + "removed": removed, "skipped": plan.skipped, + }) + except OSError as e: + if dest is not None: + try: + dest.unlink() # drop the half-copied file if the mount survives + except OSError: + pass + device_name = plan.device.name if plan.device else "device" + self.failed.emit( + f"Sync failed: {e}. Is the {device_name} still connected?") diff --git a/lintunes/gui/main_window.py b/lintunes/gui/main_window.py index 4df36b7..7ffd653 100644 --- a/lintunes/gui/main_window.py +++ b/lintunes/gui/main_window.py @@ -1,14 +1,15 @@ +import shutil from pathlib import Path from PyQt6.QtWidgets import ( QMainWindow, QSplitter, QStackedWidget, QWidget, QVBoxLayout, QLineEdit, QPlainTextEdit, QTextEdit, QAbstractSpinBox, QComboBox, QApplication, - QLabel, QMessageBox, QFileDialog, + QLabel, QMessageBox, QFileDialog, QProgressBar, ) from PyQt6.QtCore import Qt, QEvent, QTimer from PyQt6.QtGui import QAction, QKeySequence -from lintunes import mpris, tagging, theme +from lintunes import device_sync, mpris, tagging, theme from lintunes.art_search import AlbumArtFetcher from lintunes.eventlog import log_control from lintunes.inhibit import SleepInhibitor @@ -101,6 +102,15 @@ class MainWindow(QMainWindow): lambda msg: self.statusBar().showMessage(msg, 8000)) self._updater.update_applied.connect(self._restart_for_update) self._updater.start_checking() + # Device-sync progress, right-justified: permanent widgets sit on the + # status bar's right and survive showMessage. No stretch (see above) + # and hidden while idle so it costs no space. + self._sync_progress = QProgressBar() + self._sync_progress.setFixedWidth(220) + self._sync_progress.setTextVisible(True) + self._sync_progress.hide() + self.statusBar().addPermanentWidget(self._sync_progress) + self._sync_worker = None # Wiring self._transport.play_clicked.connect(self.play_pause) @@ -226,6 +236,15 @@ class MainWindow(QMainWindow): self._add_action(view_menu, "Toggle Column Browser", "Ctrl+B", self._library_view.toggle_browser) + device_menu = bar.addMenu("&Device") + self._sync_action = self._add_action( + device_menu, "Sync Playlist to Rabbit", "", + self._sync_playlist_to_device) + self._sync_action.setEnabled(False) + # Re-checked every time the menu opens: cheap (one gvfs listdir), and + # always reflects plug/unplug and the current view. + device_menu.aboutToShow.connect(self._refresh_device_actions) + track_menu = bar.addMenu("&Track") self._add_action(track_menu, "Get Info", "Ctrl+I", self._info_for_current_view) self._add_action(track_menu, "Go to Current Song", "Ctrl+L", @@ -250,6 +269,72 @@ class MainWindow(QMainWindow): self._redo_action.setText( f"Redo {stack.redo_label()}" if can_redo else "Redo") + # ---- device sync ---- + + def _refresh_device_actions(self): + syncing = self._sync_worker is not None and self._sync_worker.busy() + on_playlist = (self._content.currentWidget() is self._playlist_view + and bool(self._playlist_view.playlist_id)) + self._sync_action.setEnabled( + not syncing and on_playlist + and device_sync.find_device() is not None) + + def _sync_playlist_to_device(self): + # Everything may have changed since the menu opened — re-verify. + if self._sync_worker is not None and self._sync_worker.busy(): + return + device = device_sync.find_device() + playlist = self._manager.library.playlists.get( + self._playlist_view.playlist_id) + if device is None or playlist is None: + return + tracks = [self._manager.library.tracks[tid] + for tid in playlist.track_ids + if tid in self._manager.library.tracks] + try: + plan = device_sync.plan_sync( + playlist.name, tracks, device.music_dir, device) + free = shutil.disk_usage(device.root).free + except OSError as e: + QMessageBox.warning(self, "Sync failed", + f"Couldn't read the {device.name}: {e}") + return + needed = plan.bytes_to_copy + device_sync.SPACE_MARGIN + if free + plan.bytes_freed < needed: + QMessageBox.warning( + self, "Not enough space", + f"Syncing “{playlist.name}” needs " + f"{device_sync.format_bytes(needed - plan.bytes_freed)}, but the " + f"{device.name} only has {device_sync.format_bytes(free)} free.") + return + self._sync_worker = device_sync.DeviceSyncWorker(plan, self) + self._sync_worker.progress.connect(self._on_sync_progress) + self._sync_worker.finished.connect(self._on_sync_finished) + self._sync_worker.failed.connect(self._on_sync_failed) + self._sync_progress.setRange(0, max(plan.bytes_to_copy // 1024, 1)) + self._sync_progress.setValue(0) + self._sync_progress.setFormat(f"Syncing “{playlist.name}”…") + self._sync_progress.show() + self._sync_worker.start() + + def _on_sync_progress(self, done_kib, total_kib, label): + self._sync_progress.setRange(0, total_kib) + self._sync_progress.setValue(done_kib) + self._sync_progress.setFormat(f"Syncing {label} · %p%") + + def _on_sync_finished(self, summary): + self._sync_progress.hide() + msg = (f"Synced “{summary['playlist']}” to the {summary['device']}: " + f"{summary['copied']} copied, {summary['kept']} up to date, " + f"{summary['removed']} removed") + if summary["skipped"]: + msg += f", {summary['skipped']} skipped (no local file)" + self.statusBar().showMessage(msg, 8000) + + def _on_sync_failed(self, message): + self._sync_progress.hide() + QMessageBox.warning(self, "Sync failed", message) + # ---- view switching ---- def _show_library(self): diff --git a/tests/test_round26.py b/tests/test_round26.py new file mode 100644 index 0000000..ed18e22 --- /dev/null +++ b/tests/test_round26.py @@ -0,0 +1,224 @@ +"""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_source_vanishing_reports_failure(self, qapp, tmp_path): + a = _audio(tmp_path, "a.mp3") + plan = plan_sync("Mix", [_track(1, "A", "X", a)], tmp_path / "Music") + a.unlink() # unplugged/deleted between plan and run + result = self._synced(qapp, plan) + assert "Sync failed" in result["error"] + # No half-copied file left behind at full-looking size. + leftover = plan.dest_dir / "X - A.mp3" + assert not leftover.exists() or leftover.stat().st_size != plan.copies[0].size + + 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