clean up hacky thing

rose/__init__.py

@@ -16,7 +16,6 @@
     genre_exists,
     get_collage,
     get_collage_releases,
-    get_path_of_track_in_playlist,
     get_playlist,
     get_playlist_tracks,
     get_release,
@@ -32,6 +31,9 @@
     make_release_logtext,
     make_track_logtext,
     maybe_invalidate_cache_database,
+    release_within_collage,
+    track_within_playlist,
+    track_within_release,
     update_cache,
     update_cache_evict_nonexistent_collages,
     update_cache_evict_nonexistent_playlists,
@@ -153,6 +155,7 @@
     "dump_track",
     "get_track",
     "get_tracks_of_release",
+    "track_within_release",
     # Artists
     "Artist",
     "ArtistMapping",
@@ -182,6 +185,7 @@
     "get_collage_releases",
     "list_collages",
     "remove_release_from_collage",
+    "release_within_collage",
     "rename_collage",
     # Playlists
     "Playlist",
@@ -195,7 +199,7 @@
     "dump_all_playlists",
     "dump_playlist",
     "edit_playlist_in_editor",  # TODO: Move editor part to CLI, make this file-submissions.
-    "get_path_of_track_in_playlist",  # TODO: Redesign.
+    "track_within_playlist",
     "remove_track_from_playlist",
     "rename_playlist",
     "set_playlist_cover_art",

rose/cache.py

@@ -2031,51 +2031,57 @@ def get_tracks_of_releases(
     return rval
 
 
-def get_path_of_track_in_release(
+def track_within_release(
     c: Config,
     track_id: str,
     release_id: str,
-) -> Path | None:
+) -> bool | None:
     with connect(c) as conn:
         cursor = conn.execute(
             """
-            SELECT source_path
+            SELECT 1
             FROM tracks
             WHERE id = ? AND release_id = ?
             """,
-            (
-                track_id,
-                release_id,
-            ),
+            (track_id, release_id),
         )
-        row = cursor.fetchone()
-        if row:
-            return Path(row["source_path"])
-        return None
+        return bool(cursor.fetchone())
 
 
-def get_path_of_track_in_playlist(
+def track_within_playlist(
     c: Config,
     track_id: str,
     playlist_name: str,
-) -> Path | None:
+) -> bool:
     with connect(c) as conn:
         cursor = conn.execute(
             """
-            SELECT t.source_path
+            SELECT 1
             FROM tracks t
             JOIN playlists_tracks pt ON pt.track_id = t.id AND pt.playlist_name = ?
             WHERE t.id = ?
             """,
-            (
-                playlist_name,
-                track_id,
-            ),
+            (playlist_name, track_id),
         )
-        row = cursor.fetchone()
-        if row:
-            return Path(row["source_path"])
-        return None
+        return bool(cursor.fetchone())
+
+
+def release_within_collage(
+    c: Config,
+    release_id: str,
+    collage_name: str,
+) -> bool:
+    with connect(c) as conn:
+        cursor = conn.execute(
+            """
+            SELECT 1
+            FROM releases t
+            JOIN collages_releases pt ON pt.release_id = t.id AND pt.collage_name = ?
+            WHERE t.id = ?
+            """,
+            (collage_name, release_id),
+        )
+        return bool(cursor.fetchone())
 
 
 def get_track_logtext(c: Config, track_id: str) -> str | None:

rose/cache_test.py

@@ -26,8 +26,6 @@
     genre_exists,
     get_collage,
     get_collage_releases,
-    get_path_of_track_in_playlist,
-    get_path_of_track_in_release,
     get_playlist,
     get_playlist_tracks,
     get_release,
@@ -47,6 +45,9 @@
     list_tracks,
     lock,
     maybe_invalidate_cache_database,
+    release_within_collage,
+    track_within_playlist,
+    track_within_release,
     update_cache,
     update_cache_evict_nonexistent_releases,
     update_cache_for_releases,
@@ -1541,25 +1542,27 @@ def test_get_track(config: Config) -> None:
 
 
 @pytest.mark.usefixtures("seeded_cache")
-def test_get_path_of_track_in_release(config: Config) -> None:
-    assert (
-        get_path_of_track_in_release(config, "t1", "r1")
-        == config.music_source_dir / "r1" / "01.m4a"
-    )
-    assert get_path_of_track_in_release(config, "t3", "r1") is None
-    assert get_path_of_track_in_release(config, "lalala", "r1") is None
-    assert get_path_of_track_in_release(config, "t1", "lalala") is None
+def test_track_within_release(config: Config) -> None:
+    assert track_within_release(config, "t1", "r1")
+    assert not track_within_release(config, "t3", "r1")
+    assert not track_within_release(config, "lalala", "r1")
+    assert not track_within_release(config, "t1", "lalala")
 
 
 @pytest.mark.usefixtures("seeded_cache")
-def test_get_path_of_track_in_playlist(config: Config) -> None:
-    assert (
-        get_path_of_track_in_playlist(config, "t1", "Lala Lisa")
-        == config.music_source_dir / "r1" / "01.m4a"
-    )
-    assert get_path_of_track_in_playlist(config, "t2", "Lala Lisa") is None
-    assert get_path_of_track_in_playlist(config, "lalala", "Lala Lisa") is None
-    assert get_path_of_track_in_playlist(config, "t1", "lalala") is None
+def test_track_within_playlist(config: Config) -> None:
+    assert track_within_playlist(config, "t1", "Lala Lisa")
+    assert not track_within_playlist(config, "t2", "Lala Lisa")
+    assert not track_within_playlist(config, "lalala", "Lala Lisa")
+    assert not track_within_playlist(config, "t1", "lalala")
+
+
+@pytest.mark.usefixtures("seeded_cache")
+def test_release_within_collage(config: Config) -> None:
+    assert release_within_collage(config, "r1", "Rose Gold")
+    assert not release_within_collage(config, "r1", "Ruby Red")
+    assert not release_within_collage(config, "lalala", "Rose Gold")
+    assert not release_within_collage(config, "r1", "lalala")
 
 
 @pytest.mark.usefixtures("seeded_cache")

rose_vfs/virtualfs.py

@@ -78,7 +78,6 @@
     evaluate_track_template,
     genre_exists,
     get_collage,
-    get_path_of_track_in_playlist,
     get_playlist,
     get_release,
     get_track,
@@ -102,7 +101,13 @@
     set_release_cover_art,
     update_cache_for_releases,
 )
-from rose.cache import get_collage_releases, get_playlist_tracks, list_releases_delete_this
+from rose.cache import (
+    get_collage_releases,
+    get_playlist_tracks,
+    list_releases_delete_this,
+    track_within_playlist,
+    track_within_release,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -915,12 +920,14 @@ def _getattr_release(self, p: VirtualPath) -> dict[str, Any]:
         if p.file == f".rose.{release.id}.toml":
             return self.stat("file")
         track_id = self._get_track_id(p)
-        tracks = get_tracks_of_release(self.config, release)
-        for t in tracks:
-            if t.id == track_id:
-                return self.stat("file", t.source_path)
-        logger.debug("LOGICAL: Resolved track_id not found in the given tracklist")
-        raise llfuse.FUSEError(errno.ENOENT)
+        if not track_within_release(self.config, track_id, release.id):
+            logger.debug("LOGICAL: Resolved track_id not found in the given release")
+            raise llfuse.FUSEError(errno.ENOENT)
+        if track := get_track(self.config, track_id):
+            return self.stat("file", track.source_path)
+        raise RoseError(
+            "Impossible: Resolved track_id after track_within_release check does not exist"
+        )
 
     def getattr(self, p: VirtualPath) -> dict[str, Any]:
         logger.debug(f"LOGICAL: Received getattr for {p=}")
@@ -934,16 +941,20 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
                 if playlist.cover_path and f"cover{playlist.cover_path.suffix}" == p.file:
                     return self.stat("file", playlist.cover_path)
                 track_id = self._get_track_id(p)
-                if source_path := get_path_of_track_in_playlist(self.config, track_id, p.playlist):
-                    return self.stat("file", source_path)
-                raise llfuse.FUSEError(errno.ENOENT)
+                if not track_within_playlist(self.config, track_id, p.playlist):
+                    raise llfuse.FUSEError(errno.ENOENT)
+                if track := get_track(self.config, track_id):
+                    return self.stat("file", track.source_path)
+                raise RoseError(
+                    "Impossible: Resolved track_id after track_within_playlist check does not exist"
+                )
             return self.stat("dir")
 
         # 6. Collages
         if p.collage:
             if not get_collage(self.config, p.collage):
                 raise llfuse.FUSEError(errno.ENOENT)
-            if p.release:
+            if p.release:  # TODO: Validate existence of release in collage.
                 return self._getattr_release(p)
             return self.stat("dir")
 
@@ -952,7 +963,7 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
             la = self.sanitizer.unsanitize(p.label, p.label_parent)
             if not label_exists(self.config, la) or not self.can_show.label(la):
                 raise llfuse.FUSEError(errno.ENOENT)
-            if p.release:
+            if p.release:  # TODO: Validate existence of release in label.
                 return self._getattr_release(p)
             return self.stat("dir")
 
@@ -961,7 +972,7 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
             d = self.sanitizer.unsanitize(p.descriptor, p.descriptor_parent)
             if not descriptor_exists(self.config, d) or not self.can_show.descriptor(d):
                 raise llfuse.FUSEError(errno.ENOENT)
-            if p.release:
+            if p.release:  # TODO: Validate existence of release in descriptor.
                 return self._getattr_release(p)
             return self.stat("dir")
 
@@ -970,7 +981,7 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
             g = self.sanitizer.unsanitize(p.genre, p.genre_parent)
             if not genre_exists(self.config, g) or not self.can_show.genre(g):
                 raise llfuse.FUSEError(errno.ENOENT)
-            if p.release:
+            if p.release:  # TODO: Validate existence of release in genre.
                 return self._getattr_release(p)
             return self.stat("dir")
 
@@ -979,7 +990,7 @@ def getattr(self, p: VirtualPath) -> dict[str, Any]:
             a = self.sanitizer.unsanitize(p.artist, p.artist_parent)
             if not artist_exists(self.config, a) or not self.can_show.artist(a):
                 raise llfuse.FUSEError(errno.ENOENT)
-            if p.release:
+            if p.release:  # TODO: Validate existence of release in artist.
                 return self._getattr_release(p)
             return self.stat("dir")