"""Round 16: live multi-machine sync — conflict merge summaries + backups, and the LibraryManager reloading/reconciling external (Syncthing) changes without losing local unsaved edits. """ import json from lintunes.storage import json_storage, conflict_resolver from lintunes.library_manager import LibraryManager from lintunes.models import Library, Track, Playlist, PlaylistType # --------------------------------------------------------------------------- # # helpers # --------------------------------------------------------------------------- # def _save(tmp_path, tracks=(), playlists=()): lib = Library() for t in tracks: lib.tracks[t.track_id] = t for p in playlists: lib.playlists[p.persistent_id] = p json_storage.save_library(lib, tmp_path) return lib def _write_library_json(tmp_path, tracks_dict): (tmp_path / "library.json").write_text( json.dumps(tracks_dict, indent=2), encoding="utf-8") def _read_library_json(tmp_path): return json.loads((tmp_path / "library.json").read_text()) def _conflict_name(original: str) -> str: # Syncthing inserts before the extension: library.json -> library.sync-...json stem, _, ext = original.rpartition(".") return f"{stem}.sync-conflict-20240101-120000-ABCDEFG.{ext}" # --------------------------------------------------------------------------- # # conflict_resolver: merges, summaries, backups, restore # --------------------------------------------------------------------------- # class TestConflictResolver: def test_library_playcount_max_and_summary(self, tmp_path): _save(tmp_path, [Track(track_id=1, name="A", play_count=5)]) # Other machine's copy has a higher play count. conflict = {"1": {"track_id": 1, "name": "A", "play_count": 9}} (tmp_path / _conflict_name("library.json")).write_text(json.dumps(conflict)) summaries = conflict_resolver.resolve_conflicts(tmp_path) assert _read_library_json(tmp_path)["1"]["play_count"] == 9 assert len(summaries) == 1 assert summaries[0].kind == "library" assert any("9" in line for line in summaries[0].lines) def test_backs_up_both_sides_and_removes_conflict(self, tmp_path): _save(tmp_path, [Track(track_id=1, name="A", play_count=5)]) conflict_path = tmp_path / _conflict_name("library.json") conflict_path.write_text(json.dumps({"1": {"track_id": 1, "play_count": 9}})) summaries = conflict_resolver.resolve_conflicts(tmp_path) backup_dir = tmp_path / ".resolved" assert (backup_dir).exists() # both the pre-merge original and the incoming copy were kept assert list(backup_dir.rglob("original/library.json")) assert list(backup_dir.rglob("incoming/library.json")) assert not conflict_path.exists() # conflict file cleared from the data dir assert summaries[0].backup_dir def test_playlist_union_keeps_all_tracks(self, tmp_path): pl = Playlist(name="Road", persistent_id="AAAA1111", playlist_type=PlaylistType.REGULAR, track_ids=[1, 2]) _save(tmp_path, playlists=[pl]) pfile = tmp_path / "playlists" / "AAAA1111.json" base = json.loads(pfile.read_text()) other = dict(base) other["track_ids"] = [2, 3] # other machine added 3 (tmp_path / "playlists" / _conflict_name("AAAA1111.json")).write_text( json.dumps(other)) conflict_resolver.resolve_conflicts(tmp_path) merged = set(json.loads(pfile.read_text())["track_ids"]) assert merged == {1, 2, 3} # nothing dropped def test_restore_backup_reverts(self, tmp_path): _save(tmp_path, [Track(track_id=1, name="A", play_count=5)]) (tmp_path / _conflict_name("library.json")).write_text( json.dumps({"1": {"track_id": 1, "play_count": 9}})) summaries = conflict_resolver.resolve_conflicts(tmp_path) assert _read_library_json(tmp_path)["1"]["play_count"] == 9 conflict_resolver.restore_backup(summaries[0].backup_dir, tmp_path) assert _read_library_json(tmp_path)["1"]["play_count"] == 5 # pre-merge # --------------------------------------------------------------------------- # # LibraryManager: self-write suppression + reconciling reload # --------------------------------------------------------------------------- # class TestManagerSync: def _mgr(self, tmp_path, tracks=(), playlists=()): lib = _save(tmp_path, tracks, playlists) return LibraryManager(lib, tmp_path) def test_own_write_not_flagged_external(self, qapp, tmp_path): mgr = self._mgr(tmp_path, [Track(track_id=1, name="A", rating=0)]) assert mgr._has_external_changes() is False mgr.update_track_fields(1, {"rating": 80}) mgr.flush() assert mgr._has_external_changes() is False # our own save def test_external_write_flagged(self, qapp, tmp_path): mgr = self._mgr(tmp_path, [Track(track_id=1, name="A")]) data = _read_library_json(tmp_path) data["1"]["name"] = "A much longer name (changes size + mtime)" _write_library_json(tmp_path, data) assert mgr._has_external_changes() is True def test_reload_adopts_external_and_keeps_local_edit(self, qapp, tmp_path): mgr = self._mgr(tmp_path, [ Track(track_id=1, name="One"), Track(track_id=2, name="Two")]) # Local unsaved edit (bumps date_modified to now → should win for t1). mgr.update_track_fields(1, {"rating": 100}) # Other machine edited t2 with a clearly-newer timestamp. data = _read_library_json(tmp_path) data["2"]["rating"] = 80 data["2"]["date_modified"] = "2999-01-01T00:00:00" _write_library_json(tmp_path, data) mgr.reload_from_disk() assert mgr.library.tracks[1].rating == 100 # local edit preserved assert mgr.library.tracks[2].rating == 80 # external change adopted def test_check_for_external_merges_conflict_and_emits(self, qapp, tmp_path): mgr = self._mgr(tmp_path, [Track(track_id=1, name="A", play_count=2)]) seen = [] mgr.conflict_resolved.connect(lambda s: seen.append(s)) (tmp_path / _conflict_name("library.json")).write_text( json.dumps({"1": {"track_id": 1, "play_count": 7}})) mgr.check_for_external_changes() assert mgr.library.tracks[1].play_count == 7 # merged into memory assert len(seen) == 1 and seen[0][0].kind == "library" def test_reload_picks_up_new_playlist(self, qapp, tmp_path): mgr = self._mgr(tmp_path, [Track(track_id=1, name="A")]) # Other machine created a playlist file. newpl = Playlist(name="Fresh", persistent_id="BBBB2222", playlist_type=PlaylistType.REGULAR, track_ids=[1]) json_storage.save_playlist(newpl, tmp_path) mgr.check_for_external_changes() assert "BBBB2222" in mgr.library.playlists assert mgr.library.playlists["BBBB2222"].name == "Fresh"