"""Round 51: counts come back from the Rabbit, and only playable files go. What's pinned: * **the format gate** — andTunes sync goes through `web_support.conversion_for` like the web-mix export: FairPlay is refused and reported, ALAC and AIFF become FLAC (lossless, never a lossy re-encode), everything else is a byte copy. A conversion is cached so the next sync is a size diff again, and one file ffmpeg can't read costs that song, not the sync; * **play counts** — the app's journal is the Round 38 per-machine shape, so bringing it back is `merge_totals` (the per-track max) into the data dir, and `PlayJournal` folds it in like any other machine's. A copy that's behind can never pull a count down, and an idle sync writes nothing; * **the Auxio sync is retired** from the menu. """ import json import shutil import subprocess import pytest from lintunes.andtunes import plays from lintunes.andtunes.sync import ( AndTunesSyncWorker, flac_cache, plan_andtunes_sync, ) from lintunes.export import web_support from lintunes.models import Library, Track from lintunes.models.playlist import Playlist, PlaylistType from lintunes.storage.play_journal import PlayJournal, merge_totals needs_ffmpeg = pytest.mark.skipif(not web_support.ffmpeg_available(), reason="ffmpeg not installed") @pytest.fixture(autouse=True) def _private_cache(tmp_path, monkeypatch): monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache")) def _file(tmp_path, name, data=b"x" * 64): path = tmp_path / "local" / name path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) return path def _aiff(tmp_path, name="Reel.aiff"): path = tmp_path / "local" / name path.parent.mkdir(parents=True, exist_ok=True) subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-f", "lavfi", "-i", "sine=frequency=440:duration=0.5", "-codec:a", "pcm_s16be", str(path)], check=True) return path def _track(tid, path, kind="", name=None): return Track(track_id=tid, name=name or f"Song {tid}", artist="Band", album="Record", location=str(path), total_time=60_000, kind=kind) def _playlist(*tids): return Playlist(name="Mix", persistent_id="AAAA0001", playlist_type=PlaylistType.REGULAR, track_ids=list(tids)) def _root(tmp_path): return tmp_path / "device" / "Music" / "andTunes" def _run(plan): worker = AndTunesSyncWorker(plan) out = {} worker.finished.connect(lambda s: out.setdefault("finished", s)) worker.failed.connect(lambda m: out.setdefault("failed", m)) worker._run() assert "failed" not in out, out.get("failed") return out["finished"] def _manifest(root): return json.loads((root / "library.json").read_text(encoding="utf-8")) # ---- the format gate ---- def test_gate_refuses_fairplay_and_converts_lossless(tmp_path): tracks = { 1: _track(1, _file(tmp_path, "a.mp3"), "MPEG audio file"), 2: _track(2, _file(tmp_path, "b.m4a"), "Apple Lossless audio file"), 3: _track(3, _file(tmp_path, "c.aiff"), "AIFF audio file"), 4: _track(4, _file(tmp_path, "d.m4p"), "Protected AAC audio file"), 5: _track(5, _file(tmp_path, "e.m4a"), "AAC audio file"), } plan = plan_andtunes_sync([_playlist(1, 2, 3, 4, 5)], tracks, _root(tmp_path)) by_id = {item.track_id: item for item in plan.items} assert set(by_id) == {1, 2, 3, 5} assert by_id[1].convert_to is None and by_id[1].rel.endswith(".mp3") assert by_id[5].convert_to is None and by_id[5].rel.endswith(".m4a") for tid in (2, 3): assert by_id[tid].convert_to == "flac" assert by_id[tid].rel.endswith(".flac") assert plan.refused == [("Band - Song 4", "it's copy-protected AAC")] assert plan.playlists[0].track_ids == [1, 2, 3, 5] assert plan.needs_ffmpeg and len(plan.conversions) == 2 def test_a_suffixless_file_still_gets_a_flac_name(tmp_path): tracks = {1: _track(1, _file(tmp_path, "noext"), "")} plan = plan_andtunes_sync([_playlist(1)], tracks, _root(tmp_path)) assert plan.items[0].rel.endswith("Song 1.flac") @needs_ffmpeg def test_conversion_lands_as_flac_and_is_cached(tmp_path): src = _aiff(tmp_path) tracks = {1: _track(1, src, "AIFF audio file")} root = _root(tmp_path) plan = plan_andtunes_sync([_playlist(1)], tracks, root, copy_art=False) summary = _run(plan) assert summary["tracks"] == 1 and summary["unconverted"] == 0 entry = _manifest(root)["tracks"][0] assert entry["path"].endswith(".flac") landed = root / entry["path"] assert landed.read_bytes()[:4] == b"fLaC" cache = flac_cache(tracks[1]) assert cache.read_bytes() == landed.read_bytes() # Next sync: a plain size diff against the cached FLAC — nothing to do. again = plan_andtunes_sync([_playlist(1)], tracks, root, copy_art=False) assert again.items[0].convert_to is None assert again.items[0].src == cache assert again.kept == 1 and not again.copies and not again.needs_ffmpeg @needs_ffmpeg def test_a_retag_makes_the_cached_flac_stale(tmp_path): src = _aiff(tmp_path) tracks = {1: _track(1, src, "AIFF audio file")} root = _root(tmp_path) _run(plan_andtunes_sync([_playlist(1)], tracks, root, copy_art=False)) cache = flac_cache(tracks[1]) stamp = cache.stat().st_mtime import os os.utime(src, (stamp + 10, stamp + 10)) # the source got rewritten plan = plan_andtunes_sync([_playlist(1)], tracks, root, copy_art=False) assert plan.items[0].convert_to == "flac" assert plan.items[0].device_size == (root / plan.items[0].rel).stat().st_size # Re-converting the same audio gives the same FLAC, so it isn't re-copied. summary = _run(plan) assert summary["tracks"] == 1 def test_a_file_ffmpeg_cant_read_costs_that_song_not_the_sync(tmp_path, monkeypatch): def broken(src, dest, cancel=None): raise OSError("ffmpeg failed on it") monkeypatch.setattr(web_support, "convert_to_flac", broken) tracks = {1: _track(1, _file(tmp_path, "a.mp3"), "MPEG audio file"), 2: _track(2, _file(tmp_path, "b.aiff"), "AIFF audio file")} root = _root(tmp_path) summary = _run(plan_andtunes_sync([_playlist(1, 2)], tracks, root, copy_art=False)) assert summary["unconverted"] == 1 and summary["tracks"] == 1 manifest = _manifest(root) assert [t["id"] for t in manifest["tracks"]] == [1] assert manifest["playlists"][0]["tracks"] == [1] assert not flac_cache(tracks[2]).exists() def test_without_ffmpeg_conversions_can_be_left_out(tmp_path): tracks = {1: _track(1, _file(tmp_path, "a.mp3"), "MPEG audio file"), 2: _track(2, _file(tmp_path, "b.aiff"), "AIFF audio file")} root = _root(tmp_path) plan = plan_andtunes_sync([_playlist(1, 2)], tracks, root, copy_art=False) plan.drop_conversions("it needs ffmpeg to convert") assert not plan.needs_ffmpeg assert ("Band - Song 2", "it needs ffmpeg to convert") in plan.refused assert plan.bytes_to_copy == 64 summary = _run(plan) assert summary["refused"] == 1 assert [t["id"] for t in _manifest(root)["tracks"]] == [1] # ---- play counts coming back ---- def _device_journal(tmp_path, entries, name="andtunes-1a2b3c4d.json"): folder = _root(tmp_path) / "plays" folder.mkdir(parents=True, exist_ok=True) (folder / name).write_text(json.dumps(entries), encoding="utf-8") return folder / name def test_merge_totals_takes_the_higher_total_and_newest_stamp(): mine = {"1": {"plays": 5, "last_played": "2026-09-01T10:00:00"}, "2": {"plays": 1, "last_played": "2026-09-02T10:00:00"}} incoming = {"1": {"plays": 3, "last_played": "2026-09-05T10:00:00"}, "2": {"plays": 1, "last_played": "2026-09-02T10:00:00"}, "3": {"plays": 2, "last_played": "2026-09-03T10:00:00"}} merged, changed = merge_totals(mine, incoming) assert merged["1"] == {"plays": 5, "last_played": "2026-09-05T10:00:00"} assert merged["2"] == mine["2"] assert merged["3"] == incoming["3"] assert changed == 2 def test_bring_back_writes_only_when_something_moved(tmp_path): data = tmp_path / "data" _device_journal(tmp_path, {"1": {"plays": 3, "last_played": "2026-09-11T10:00:00"}}) assert plays.bring_back(_root(tmp_path), data) == 1 dest = data / "plays" / "andtunes-1a2b3c4d.json" assert json.loads(dest.read_text())["1"]["plays"] == 3 # A device copy that's behind (reinstalled app, restored backup) can't # pull the count down, and changes nothing on disk. before = dest.stat().st_mtime_ns _device_journal(tmp_path, {"1": {"plays": 2, "last_played": "2026-09-10T10:00:00"}}) assert plays.bring_back(_root(tmp_path), data) == 0 assert dest.stat().st_mtime_ns == before assert json.loads(dest.read_text())["1"]["plays"] == 3 _device_journal(tmp_path, {"1": {"plays": 7, "last_played": "2026-09-12T10:00:00"}}) assert plays.bring_back(_root(tmp_path), data) == 1 assert json.loads(dest.read_text())["1"] == { "plays": 7, "last_played": "2026-09-12T10:00:00"} def test_bring_back_ignores_what_isnt_an_andtunes_journal(tmp_path): data = tmp_path / "data" _device_journal(tmp_path, {"1": {"plays": 1}}, name="someone-else.json") bad = _device_journal(tmp_path, {}, name="andtunes-half.json") bad.write_text('{"1": {"pla', encoding="utf-8") # mid-write _device_journal(tmp_path, {"tracks": {"1": 2}, "9": {"plays": 1}}, name="andtunes-odd.json") assert plays.bring_back(_root(tmp_path), data) == 1 assert sorted(p.name for p in (data / "plays").iterdir()) == ["andtunes-odd.json"] assert json.loads((data / "plays" / "andtunes-odd.json").read_text()) == { "9": {"plays": 1}} def test_the_rabbit_is_one_more_machine_to_play_journal(tmp_path): data = tmp_path / "data" _device_journal(tmp_path, {"1": {"plays": 3, "last_played": "2026-09-11T10:00:00"}}) plays.bring_back(_root(tmp_path), data) library = Library() library.tracks[1] = Track(track_id=1, name="Song", play_count=10) PlayJournal(machine="thisbox").load(data, library) assert library.tracks[1].play_count == 13 assert library.tracks[1].play_date_utc == "2026-09-11T10:00:00" def test_sync_brings_the_counts_back(tmp_path): data = tmp_path / "data" _device_journal(tmp_path, {"1": {"plays": 4, "last_played": "2026-09-11T10:00:00"}}) tracks = {1: _track(1, _file(tmp_path, "a.mp3"), "MPEG audio file")} root = _root(tmp_path) plan = plan_andtunes_sync([_playlist(1)], tracks, root, copy_art=False, data_dir=data) summary = _run(plan) assert summary["plays"] == 1 # The device keeps its own journal: plays/ is the app's, never pruned. assert (root / "plays" / "andtunes-1a2b3c4d.json").exists() assert not any(rel.startswith("plays") for rel in plan.stale) # Nothing new on the device → nothing to write the second time. assert _run(plan_andtunes_sync([_playlist(1)], tracks, root, copy_art=False, data_dir=data))["plays"] == 0 def test_without_a_data_dir_nothing_comes_back(tmp_path): _device_journal(tmp_path, {"1": {"plays": 4}}) tracks = {1: _track(1, _file(tmp_path, "a.mp3"), "MPEG audio file")} summary = _run(plan_andtunes_sync([_playlist(1)], tracks, _root(tmp_path), copy_art=False)) assert summary["plays"] == 0 # ---- the old sync ---- def test_the_auxio_sync_is_off_the_menu(): from lintunes.gui import main_window assert not hasattr(main_window.MainWindow, "_sync_playlist_to_device") source = shutil.os.path.join(shutil.os.path.dirname(main_window.__file__), "main_window.py") with open(source, encoding="utf-8") as f: assert "Sync Playlist to Rabbit (Auxio)\", \"\"" not in f.read()