test the cache updater

README.md

@@ -1,4 +1,4 @@
-# rose
+# Rosé
 
 _In Progress_
 

conftest.py

@@ -5,9 +5,10 @@
 
 import _pytest.pathlib
 import pytest
+import yoyo
 from click.testing import CliRunner
 
-from rose.foundation.conf import Config
+from rose.foundation.conf import MIGRATIONS_PATH, Config
 
 logger = logging.getLogger(__name__)
 
@@ -20,12 +21,20 @@ def isolated_dir() -> Iterator[Path]:
 
 @pytest.fixture()
 def config(isolated_dir: Path) -> Config:
-    (isolated_dir / "cache").mkdir()
+    cache_dir = isolated_dir / "cache"
+    cache_dir.mkdir()
+    cache_database_path = cache_dir / "cache.sqlite3"
+
+    db_backend = yoyo.get_backend(f"sqlite:///{cache_database_path}")
+    db_migrations = yoyo.read_migrations(str(MIGRATIONS_PATH))
+    with db_backend.lock():
+        db_backend.apply_migrations(db_backend.to_apply(db_migrations))
+
     return Config(
         music_source_dir=isolated_dir / "source",
         fuse_mount_dir=isolated_dir / "mount",
-        cache_dir=isolated_dir / "cache",
-        cache_database_path=isolated_dir / "cache" / "cache.sqlite3",
+        cache_dir=cache_dir,
+        cache_database_path=cache_database_path,
     )
 
 

migrations/20231009_01_qlEHa-bootstrap.rollback.sql

@@ -9,6 +9,7 @@ DROP TABLE tracks_artists;
 DROP TABLE releases_artists;
 DROP TABLE artist_role_enum;
 DROP TABLE tracks;
+DROP TABLE releases_labels;
 DROP TABLE releases_genres;
 DROP TABLE releases;
 DROP TABLE release_type_enum;

migrations/20231009_01_qlEHa-bootstrap.sql

@@ -33,6 +33,13 @@ CREATE TABLE releases_genres (
 );
 CREATE INDEX releases_genres_genre ON releases_genres(genre);
 
+CREATE TABLE releases_labels (
+    release_id TEXT,
+    label TEXT,
+    PRIMARY KEY (release_id, label)
+);
+CREATE INDEX releases_labels_label ON releases_labels(label);
+
 CREATE TABLE tracks (
     id TEXT PRIMARY KEY,
     source_path TEXT NOT NULL UNIQUE,

rose/cache/process.py → rose/cache/update.py

@@ -31,6 +31,8 @@
     "unknown",
 ]
 
+ID_REGEX = re.compile(r"\{id=([^}]+)\}")
+
 
 @dataclass
 class CachedRelease:
@@ -71,9 +73,9 @@ def update_cache_for_all_releases(c: Config) -> None:
         conn.execute(
             f"""
             DELETE FROM releases
-            WHERE source_path NOT IN {",".join(["?"] * len(dirs))}
+            WHERE source_path NOT IN ({",".join(["?"] * len(dirs))})
             """,
-            dirs,
+            [str(d) for d in dirs],
         )
 
 
@@ -107,8 +109,9 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
                     source_path=release_dir.resolve(),
                     title=tags.album or "Unknown Release",
                     release_type=(
-                        tags.release_type
-                        if tags.release_type in SUPPORTED_RELEASE_TYPES
+                        tags.release_type.lower()
+                        if tags.release_type
+                        and tags.release_type.lower() in SUPPORTED_RELEASE_TYPES
                         else "unknown"
                     ),
                     release_year=tags.year,
@@ -148,6 +151,14 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
                         """,
                         (release.id, genre),
                     )
+                for label in tags.label:
+                    conn.execute(
+                        """
+                        INSERT INTO releases_labels (release_id, label) VALUES (?, ?)
+                        ON CONFLICT (release_id, label) DO NOTHING
+                        """,
+                        (release.id, label),
+                    )
                 for role, names in asdict(tags.album_artists).items():
                     for name in names:
                         conn.execute(
@@ -161,14 +172,18 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
 
             # Now process the track. Release is guaranteed to exist here.
             filepath = Path(f.path)
+            # Get the mtime before we may possibly rename the file.
+            source_mtime = int(f.stat().st_mtime)
+
             track_id = _parse_uuid_from_path(filepath)
             if not track_id:
                 track_id = str(uuid6.uuid7())
                 filepath = _rename_with_uuid(filepath, track_id)
+
             track = CachedTrack(
                 id=track_id,
                 source_path=filepath,
-                source_mtime=int(f.stat().st_mtime),
+                source_mtime=source_mtime,
                 title=tags.title or "Unknown Title",
                 release_id=release.id,
                 trackno=tags.track_number or "1",
@@ -223,13 +238,10 @@ def update_cache_for_release(c: Config, release_dir: Path) -> Path:
 
 
 def _parse_uuid_from_path(path: Path) -> str | None:
-    if m := re.search(r"\{id=([^\]]+)\}$", path.stem):
+    if m := ID_REGEX.search(path.stem):
         return m[1]
     return None
 
 
 def _rename_with_uuid(src: Path, uuid: str) -> Path:
-    new_stem = src.stem + f" {{id={uuid}}}"
-    dst = src.with_stem(new_stem)
-    src.rename(dst)
-    return dst
+    return src.rename(src.with_stem(src.stem + f" {{id={uuid}}}"))

rose/cache/update_test.py

@@ -0,0 +1,155 @@
+import shutil
+from pathlib import Path
+
+from rose.cache.database import connect
+from rose.cache.update import ID_REGEX, update_cache_for_all_releases, update_cache_for_release
+from rose.foundation.conf import Config
+
+TESTDATA = Path(__file__).resolve().parent / "testdata"
+TEST_RELEASE_1 = TESTDATA / "Test Release 1"
+TEST_RELEASE_2 = TESTDATA / "Test Release 2 {id=ilovecarly}"
+
+
+def test_update_cache_for_release(config: Config) -> None:
+    release_dir = config.music_source_dir / TEST_RELEASE_1.name
+    shutil.copytree(TEST_RELEASE_1, release_dir)
+    updated_release_dir = update_cache_for_release(config, release_dir)
+
+    # Check that the release directory was given a UUID.
+    m = ID_REGEX.search(updated_release_dir.name)
+    assert m is not None
+    release_id = m[1]
+
+    # Assert that the release metadata was read correctly.
+    with connect(config) as conn:
+        cursor = conn.execute(
+            """
+            SELECT id, source_path, title, release_type, release_year, new
+            FROM releases WHERE id = ?
+            """,
+            (release_id,),
+        )
+        row = cursor.fetchone()
+        assert row["source_path"] == str(updated_release_dir)
+        assert row["title"] == "A Cool Album"
+        assert row["release_type"] == "album"
+        assert row["release_year"] == 1990
+        assert row["new"]
+
+        cursor = conn.execute(
+            "SELECT genre FROM releases_genres WHERE release_id = ?",
+            (release_id,),
+        )
+        genres = {r["genre"] for r in cursor.fetchall()}
+        assert genres == {"Electronic", "House"}
+
+        cursor = conn.execute(
+            "SELECT label FROM releases_labels WHERE release_id = ?",
+            (release_id,),
+        )
+        labels = {r["label"] for r in cursor.fetchall()}
+        assert labels == {"A Cool Label"}
+
+        cursor = conn.execute(
+            "SELECT artist, role FROM releases_artists WHERE release_id = ?",
+            (release_id,),
+        )
+        artists = {(r["artist"], r["role"]) for r in cursor.fetchall()}
+        assert artists == {
+            ("Artist A", "main"),
+            ("Artist B", "main"),
+        }
+
+        for f in updated_release_dir.iterdir():
+            if f.suffix != ".m4a":
+                continue
+
+            # Check that the track file was given a UUID.
+            m = ID_REGEX.search(f.name)
+            assert m is not None
+            track_id = m[1]
+
+            # Assert that the track metadata was read correctly.
+            cursor = conn.execute(
+                """
+                SELECT
+                    id, source_path, title, release_id, track_number, disc_number, duration_seconds
+                FROM tracks WHERE id = ?
+                """,
+                (track_id,),
+            )
+            row = cursor.fetchone()
+            assert row["source_path"] == str(f)
+            assert row["title"] == "Title"
+            assert row["release_id"] == release_id
+            assert row["track_number"] != ""
+            assert row["disc_number"] == "1"
+            assert row["duration_seconds"] == 2
+
+            cursor = conn.execute(
+                "SELECT artist, role FROM tracks_artists WHERE track_id = ?",
+                (track_id,),
+            )
+            artists = {(r["artist"], r["role"]) for r in cursor.fetchall()}
+            assert artists == {
+                ("Artist GH", "main"),
+                ("Artist HI", "main"),
+                ("Artist C", "guest"),
+                ("Artist A", "guest"),
+                ("Artist AB", "remixer"),
+                ("Artist BC", "remixer"),
+                ("Artist CD", "producer"),
+                ("Artist DE", "producer"),
+                ("Artist EF", "composer"),
+                ("Artist FG", "composer"),
+                ("Artist IJ", "djmixer"),
+                ("Artist JK", "djmixer"),
+            }
+
+
+def test_update_cache_with_existing_id(config: Config) -> None:
+    """Test that IDs in filenames are read and preserved."""
+    release_dir = config.music_source_dir / TEST_RELEASE_2.name
+    shutil.copytree(TEST_RELEASE_2, release_dir)
+    updated_release_dir = update_cache_for_release(config, release_dir)
+    assert release_dir == updated_release_dir
+
+    with connect(config) as conn:
+        m = ID_REGEX.search(release_dir.name)
+        assert m is not None
+        release_id = m[1]
+        cursor = conn.execute("SELECT EXISTS(SELECT * FROM releases WHERE id = ?)", (release_id,))
+        assert cursor.fetchone()[0]
+
+        for f in release_dir.iterdir():
+            if f.suffix != ".m4a":
+                continue
+
+            # Check that the track file was given a UUID.
+            m = ID_REGEX.search(f.name)
+            assert m is not None
+            track_id = m[1]
+            cursor = conn.execute("SELECT EXISTS(SELECT * FROM tracks WHERE id = ?)", (track_id,))
+            assert cursor.fetchone()[0]
+
+
+def test_update_cache_for_all_releases(config: Config) -> None:
+    shutil.copytree(TEST_RELEASE_1, config.music_source_dir / TEST_RELEASE_1.name)
+    shutil.copytree(TEST_RELEASE_2, config.music_source_dir / TEST_RELEASE_2.name)
+
+    # Test that we prune deleted releases too.
+    with connect(config) as conn:
+        conn.execute(
+            """
+            INSERT INTO releases (id, source_path, title, release_type)
+            VALUES ('aaaaaa', '/nonexistent', 'aa', 'unknown')
+            """
+        )
+
+    update_cache_for_all_releases(config)
+
+    with connect(config) as conn:
+        cursor = conn.execute("SELECT COUNT(*) FROM releases")
+        assert cursor.fetchone()[0] == 2
+        cursor = conn.execute("SELECT COUNT(*) FROM tracks")
+        assert cursor.fetchone()[0] == 4