"""Round 14: smart playlists — criteria model, evaluation, iTunes import, manager recompute/guards, and conflict resolution. The iTunes blobs in tests/smart_blobs.json are real "Smart Info"/"Smart Criteria" data captured from the test library (gitignored), base64-encoded so the parser is exercised against genuine iTunes 12 output. """ import base64 import json import os from datetime import datetime from pathlib import Path from lintunes.smart import ( SmartCriteria, SmartGroup, SmartRule, SmartLimit, evaluate, parse_itunes_smart, ) from lintunes.library_manager import LibraryManager from lintunes.models import Library, Track, Playlist, PlaylistType _BLOBS = json.loads((Path(__file__).parent / "smart_blobs.json").read_text()) def _blob(name): rec = _BLOBS[name] return base64.b64decode(rec["info_b64"]), base64.b64decode(rec["crit_b64"]) # --------------------------------------------------------------------------- # # iTunes parser / converter (real captured blobs) # --------------------------------------------------------------------------- # class TestParseItunes: def test_sun_string_rules(self): c = parse_itunes_smart(*_blob("sun")) assert not c.unsupported and not c.has_nested assert c.root.conjunction == "all" rules = c.root.children assert all(isinstance(r, SmartRule) for r in rules) assert (rules[0].field, rules[0].operator, rules[0].value) == ("name", "contains", "sun") assert (rules[1].field, rules[1].operator, rules[1].value) == ("name", "not_contains", "sunday") def test_unrated_numeric_rules(self): c = parse_itunes_smart(*_blob("unrated")) assert not c.unsupported rating, plays = c.root.children assert rating.field == "rating" and rating.operator == "in_range" assert rating.value == 0 and rating.value2 == 0 assert plays.field == "play_count" and plays.operator == "greater" and plays.value == 1 def test_top_relative_date_and_limit(self): c = parse_itunes_smart(*_blob("top 300 of the decade!")) assert not c.unsupported rule = c.root.children[0] assert rule.field == "date_added" and rule.operator == "in_last" assert rule.value == 120 and rule.unit == "months" assert c.limit.enabled and c.limit.count == 300 and c.limit.by == "items" def test_mediakind_is_unsupported(self): # "convert these" matches on MediaKind (Podcast), which LinTunes can't # represent -> falls back to the imported snapshot. c = parse_itunes_smart(*_blob("convert these")) assert c.unsupported is True def test_junk_blob_is_unsupported_no_raise(self): c = parse_itunes_smart(b"not a real blob", b"x" * 200) assert c.unsupported is True def test_missing_blobs_unsupported(self): assert parse_itunes_smart(None, None).unsupported is True # --------------------------------------------------------------------------- # # Schema round-trip # --------------------------------------------------------------------------- # def test_criteria_dict_roundtrip_with_nesting(): crit = SmartCriteria( root=SmartGroup("all", [ SmartRule("genre", "contains", "Rock"), SmartGroup("any", [ SmartRule("rating", "in_range", 80, 100), SmartRule("play_count", "greater", 10), ]), ]), limit=SmartLimit(True, 25, "items", "random"), live_update=False, has_nested=True) back = SmartCriteria.from_dict(json.loads(json.dumps(crit.to_dict()))) assert back.root.conjunction == "all" assert isinstance(back.root.children[1], SmartGroup) assert back.root.children[1].children[0].value2 == 100 assert back.limit.selection == "random" and back.live_update is False assert back.has_nested is True # --------------------------------------------------------------------------- # # Evaluator # --------------------------------------------------------------------------- # def _tracks(*specs): out = [] for i, kw in enumerate(specs, start=1): kw.setdefault("track_id", i) out.append(Track(**kw)) return out class TestEvaluate: def test_string_contains_casefold(self): ts = _tracks({"name": "Sunshine"}, {"name": "Moon"}, {"name": "SUN city"}) c = SmartCriteria(root=SmartGroup("all", [SmartRule("name", "contains", "sun")])) assert evaluate(c, ts) == [1, 3] def test_string_is_not(self): ts = _tracks({"genre": "Rock"}, {"genre": "Jazz"}) c = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is_not", "rock")])) assert evaluate(c, ts) == [2] def test_numeric_in_range_and_greater(self): ts = _tracks({"play_count": 1}, {"play_count": 5}, {"play_count": 10}) c = SmartCriteria(root=SmartGroup("all", [SmartRule("play_count", "in_range", 2, 9)])) assert evaluate(c, ts) == [2] c2 = SmartCriteria(root=SmartGroup("all", [SmartRule("play_count", "greater", 4)])) assert evaluate(c2, ts) == [2, 3] def test_rating_stars(self): ts = _tracks({"rating": 60}, {"rating": 80}, {"rating": 100}) c = SmartCriteria(root=SmartGroup("all", [SmartRule("rating", "in_range", 80, 100)])) assert evaluate(c, ts) == [2, 3] def test_date_in_last(self): now = datetime(2026, 6, 26) ts = _tracks( {"date_added": "2026-06-01T00:00:00"}, # within {"date_added": "2025-01-01T00:00:00"}, # outside {"date_added": None}, # never -> excluded ) c = SmartCriteria(root=SmartGroup("all", [SmartRule("date_added", "in_last", 3, unit="months")])) assert evaluate(c, ts, now=now) == [1] def test_date_before_treats_null_as_distant_past(self): # A never-played track (play_date_utc=None) must match "last played # before X" — iTunes treats null dates as the distant past. ts = _tracks( {"play_date_utc": "2020-01-01T00:00:00"}, # before -> match {"play_date_utc": "2025-01-01T00:00:00"}, # after -> no {"play_date_utc": None}, # never played -> match ) c = SmartCriteria(root=SmartGroup("all", [ SmartRule("last_played", "before", "2023-01-01T00:00:00")])) assert evaluate(c, ts) == [1, 3] # "after" must NOT match a null date. c2 = SmartCriteria(root=SmartGroup("all", [ SmartRule("last_played", "after", "2023-01-01T00:00:00")])) assert evaluate(c2, ts) == [2] def test_bool_compilation(self): ts = _tracks({"compilation": True}, {"compilation": False}) c = SmartCriteria(root=SmartGroup("all", [SmartRule("compilation", "is", True)])) assert evaluate(c, ts) == [1] def test_match_any_vs_all(self): ts = _tracks({"genre": "Rock", "year": 1990}, {"genre": "Jazz", "year": 2020}) rules = [SmartRule("genre", "is", "rock"), SmartRule("year", "greater", 2000)] assert evaluate(SmartCriteria(root=SmartGroup("any", rules)), ts) == [1, 2] assert evaluate(SmartCriteria(root=SmartGroup("all", rules)), ts) == [] def test_nested_group(self): ts = _tracks( {"genre": "Rock", "rating": 100}, {"genre": "Rock", "rating": 20}, {"genre": "Jazz", "rating": 100}, ) crit = SmartCriteria(root=SmartGroup("all", [ SmartRule("genre", "is", "rock"), SmartGroup("any", [SmartRule("rating", "greater", 79)]), ])) assert evaluate(crit, ts) == [1] def test_empty_rules_matches_all(self): ts = _tracks({"name": "a"}, {"name": "b"}) assert evaluate(SmartCriteria(root=SmartGroup("all", [])), ts) == [1, 2] def test_limit_items_by_most_played(self): ts = _tracks({"play_count": 1}, {"play_count": 9}, {"play_count": 5}) c = SmartCriteria(root=SmartGroup("all", []), limit=SmartLimit(True, 2, "items", "most_often_played")) # Highest two are tracks 2 (9) and 3 (5); returned sorted by id. assert evaluate(c, ts) == [2, 3] def test_limit_by_minutes_budget(self): ts = _tracks({"total_time": 120000}, {"total_time": 120000}, {"total_time": 120000}) c = SmartCriteria(root=SmartGroup("all", []), limit=SmartLimit(True, 3, "minutes", "least_recently_added")) # Budget 3 minutes = 180000 ms -> only one 2-minute track fits. assert len(evaluate(c, ts)) == 1 # --------------------------------------------------------------------------- # # LibraryManager # --------------------------------------------------------------------------- # def _manager(tmp_path, tracks=()): library = Library() for t in tracks: library.tracks[t.track_id] = t return LibraryManager(library, tmp_path) class TestManagerSmart: def test_create_computes_membership(self, qapp, tmp_path): mgr = _manager(tmp_path, _tracks( {"genre": "Rock"}, {"genre": "Jazz"}, {"genre": "Rock"})) crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) pl = mgr.create_smart_playlist("rockers", crit) assert pl.track_ids == [1, 3] assert pl.is_smart def test_edit_field_updates_membership_and_emits(self, qapp, tmp_path): mgr = _manager(tmp_path, _tracks({"genre": "Jazz"}, {"genre": "Rock"})) crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) pl = mgr.create_smart_playlist("rockers", crit) assert pl.track_ids == [2] emitted = [] mgr.playlist_content_changed.connect(emitted.append) mgr.update_track_fields(1, {"genre": "Rock"}) # track 1 becomes Rock mgr._flush_recompute() assert pl.track_ids == [1, 2] assert pl.persistent_id in emitted def test_record_play_updates_most_played(self, qapp, tmp_path): mgr = _manager(tmp_path, _tracks({"play_count": 0}, {"play_count": 0})) crit = SmartCriteria(root=SmartGroup("all", [SmartRule("play_count", "greater", 0)])) pl = mgr.create_smart_playlist("played", crit) assert pl.track_ids == [] mgr.record_play(2) mgr._flush_recompute() assert pl.track_ids == [2] def test_noop_recompute_not_dirty(self, qapp, tmp_path): mgr = _manager(tmp_path, _tracks({"genre": "Rock"})) crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) pl = mgr.create_smart_playlist("rockers", crit) mgr.flush() assert pl.persistent_id not in mgr._dirty_playlists assert mgr.recompute_smart_playlist(pl.persistent_id) is False assert pl.persistent_id not in mgr._dirty_playlists def test_manual_edits_blocked(self, qapp, tmp_path): mgr = _manager(tmp_path, _tracks({"genre": "Rock"}, {"genre": "Jazz"})) crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) pl = mgr.create_smart_playlist("rockers", crit) assert pl.track_ids == [1] mgr.add_tracks_to_playlist(pl.persistent_id, [2]) mgr.remove_tracks_from_playlist(pl.persistent_id, [0]) mgr.move_tracks_in_playlist(pl.persistent_id, [0], 0) assert pl.track_ids == [1] # unchanged def test_unsupported_keeps_snapshot(self, qapp, tmp_path): mgr = _manager(tmp_path, _tracks({"genre": "Rock"}, {"genre": "Jazz"})) pl = Playlist(name="frozen", persistent_id="AB12CD34", playlist_type=PlaylistType.SMART, smart_criteria=SmartCriteria(unsupported=True), track_ids=[2]) mgr.library.playlists[pl.persistent_id] = pl assert mgr.recompute_smart_playlist(pl.persistent_id, force=True) is False assert pl.track_ids == [2] # snapshot preserved def test_edit_criteria_undo_redo(self, qapp, tmp_path): mgr = _manager(tmp_path, _tracks({"genre": "Rock"}, {"genre": "Jazz"})) crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) pl = mgr.create_smart_playlist("p", crit) assert pl.track_ids == [1] new = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "jazz")])) mgr.set_smart_criteria(pl.persistent_id, new) assert pl.track_ids == [2] mgr.undo_stack.undo() assert pl.smart_criteria.root.children[0].value == "rock" assert pl.track_ids == [1] mgr.undo_stack.redo() assert pl.track_ids == [2] def test_recompute_all_respects_live_update(self, qapp, tmp_path): mgr = _manager(tmp_path, _tracks({"genre": "Rock"})) live = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")]), live_update=True) frozen = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")]), live_update=False) a = Playlist(name="a", persistent_id="AAAA0001", playlist_type=PlaylistType.SMART, smart_criteria=live, track_ids=[]) b = Playlist(name="b", persistent_id="BBBB0002", playlist_type=PlaylistType.SMART, smart_criteria=frozen, track_ids=[]) mgr.library.playlists["AAAA0001"] = a mgr.library.playlists["BBBB0002"] = b changed = mgr.recompute_all_smart() # force=False assert "AAAA0001" in changed and "BBBB0002" not in changed assert a.track_ids == [1] and b.track_ids == [] # --------------------------------------------------------------------------- # # Conflict resolution # --------------------------------------------------------------------------- # def test_conflict_resolver_smart_newest_wins(tmp_path): from lintunes.storage.conflict_resolver import resolve_conflicts pdir = tmp_path / "playlists" pdir.mkdir() def crit_dict(genre): return SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", genre)])).to_dict() original = {"name": "p", "persistent_id": "PID1", "playlist_type": "smart", "track_ids": [1, 2], "smart_criteria": crit_dict("rock"), "settings": {}} conflict = {"name": "p", "persistent_id": "PID1", "playlist_type": "smart", "track_ids": [3, 4], "smart_criteria": crit_dict("jazz"), "settings": {}} orig_path = pdir / "PID1.json" conf_path = pdir / "PID1.sync-conflict-20250101-120000-ABCDEFG.json" orig_path.write_text(json.dumps(original)) conf_path.write_text(json.dumps(conflict)) os.utime(orig_path, (1000, 1000)) # original older os.utime(conf_path, (2000, 2000)) # conflict newer resolve_conflicts(tmp_path) merged = json.loads(orig_path.read_text()) # Newest criteria wins; track_ids are NOT unioned (no [1,2,3,4]). assert merged["smart_criteria"]["root"]["children"][0]["value"] == "jazz" assert merged["track_ids"] == [3, 4]