From f3d5ef6528dd59205b9f0b25215c1cb1965f70fa Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Sun, 17 May 2026 14:18:17 -0700 Subject: [PATCH] Test missing-track existing file imports Add service-level coverage for the Enhanced Library I Have This flow: copying an existing source file, writing the target album DB row, preserving source audio, inheriting album identity tags, and migrating older track tables that lack disc_number. --- core/library/missing_track_import.py | 7 + tests/library/test_missing_track_import.py | 260 +++++++++++++++++++++ 2 files changed, 267 insertions(+) create mode 100644 tests/library/test_missing_track_import.py diff --git a/core/library/missing_track_import.py b/core/library/missing_track_import.py index 65bf6357..7b322be4 100644 --- a/core/library/missing_track_import.py +++ b/core/library/missing_track_import.py @@ -140,6 +140,7 @@ def import_existing_track_for_album_slot(album_id: str, payload: dict, deps: Mis if not final_path or not os.path.exists(final_path): raise MissingTrackImportError("Post-processing did not produce a final file", 500) + ensure_tracks_disc_number_column(database) copy_album_identity_from_target_sibling( database, album_id, @@ -383,6 +384,12 @@ def _ensure_disc_number_column(cursor, conn) -> None: conn.commit() +def ensure_tracks_disc_number_column(database) -> None: + with database._get_connection() as conn: + cursor = conn.cursor() + _ensure_disc_number_column(cursor, conn) + + def _read_file_stats(final_path: str, source_track: dict) -> tuple[Optional[int], int]: file_size = None bitrate = source_track.get("bitrate") or 0 diff --git a/tests/library/test_missing_track_import.py b/tests/library/test_missing_track_import.py new file mode 100644 index 00000000..077810be --- /dev/null +++ b/tests/library/test_missing_track_import.py @@ -0,0 +1,260 @@ +"""Tests for the Enhanced Library "I Have This" import service.""" + +from __future__ import annotations + +import os +import shutil +import sqlite3 +from dataclasses import dataclass + +import pytest + +from core.library import missing_track_import as mti + + +class _ConnCtx: + def __init__(self, conn): + self.conn = conn + + def __enter__(self): + return self.conn + + def __exit__(self, exc_type, exc, tb): + return False + + +class _FakeDB: + def __init__(self, conn): + self.conn = conn + + def _get_connection(self): + return _ConnCtx(self.conn) + + +@dataclass +class _FakeConfig: + download_path: str + active_server: str = "navidrome" + + def get(self, key, default=None): + if key == "soulseek.download_path": + return self.download_path + return default + + def get_active_media_server(self): + return self.active_server + + +def _make_db(*, include_disc_number: bool = True) -> tuple[_FakeDB, sqlite3.Connection]: + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + cur = conn.cursor() + cur.execute( + """ + CREATE TABLE artists ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL + ) + """ + ) + cur.execute( + """ + CREATE TABLE albums ( + id TEXT PRIMARY KEY, + artist_id TEXT NOT NULL, + title TEXT NOT NULL, + year INTEGER, + track_count INTEGER, + server_source TEXT, + deezer_id TEXT, + thumb_url TEXT + ) + """ + ) + disc_col = ", disc_number INTEGER DEFAULT 1" if include_disc_number else "" + cur.execute( + f""" + CREATE TABLE tracks ( + id TEXT PRIMARY KEY, + album_id TEXT NOT NULL, + artist_id TEXT NOT NULL, + title TEXT NOT NULL, + track_number INTEGER{disc_col}, + duration INTEGER, + file_path TEXT, + bitrate INTEGER, + file_size INTEGER, + server_source TEXT, + deezer_id TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + cur.execute("INSERT INTO artists (id, name) VALUES ('artist-1', 'Kendrick Lamar')") + cur.execute( + """ + INSERT INTO albums (id, artist_id, title, year, track_count, server_source, deezer_id) + VALUES ('album-basic', 'artist-1', 'DAMN.', 2017, 14, 'navidrome', '302127') + """ + ) + cur.execute( + """ + INSERT INTO albums (id, artist_id, title, year, track_count, server_source, deezer_id) + VALUES ('album-deluxe', 'artist-1', 'DAMN. COLLECTORS EDITION', 2017, 14, 'navidrome', '999999') + """ + ) + conn.commit() + return _FakeDB(conn), conn + + +def _insert_track(conn, *, track_id, album_id, title, track_number, file_path, disc_number=1): + columns = [row[1] for row in conn.execute("PRAGMA table_info(tracks)").fetchall()] + if "disc_number" in columns: + conn.execute( + """ + INSERT INTO tracks (id, album_id, artist_id, title, track_number, disc_number, duration, file_path, bitrate, file_size, server_source) + VALUES (?, ?, 'artist-1', ?, ?, ?, 177000, ?, 900, 1234, 'navidrome') + """, + (track_id, album_id, title, track_number, disc_number, str(file_path)), + ) + else: + conn.execute( + """ + INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path, bitrate, file_size, server_source) + VALUES (?, ?, 'artist-1', ?, ?, 177000, ?, 900, 1234, 'navidrome') + """, + (track_id, album_id, title, track_number, str(file_path)), + ) + conn.commit() + + +def _deps(tmp_path, db, *, post_process_fn=None, sync_calls=None): + sync_calls = sync_calls if sync_calls is not None else [] + + def _default_post_process(_key, context, staged_path): + final_dir = tmp_path / "Library" / "Kendrick Lamar - 2017 DAMN" + final_dir.mkdir(parents=True, exist_ok=True) + final_path = final_dir / "08 - HUMBLE [FLAC 16bit].flac" + shutil.copy2(staged_path, final_path) + context["_final_processed_path"] = str(final_path) + + return mti.MissingTrackImportDeps( + database=db, + config_manager=_FakeConfig(str(tmp_path / "downloads")), + post_process_fn=post_process_fn or _default_post_process, + resolve_library_file_path_fn=lambda path: str(path) if path and os.path.exists(path) else None, + docker_resolve_path_fn=lambda path: path, + sync_tracks_to_server_fn=lambda rows, server: sync_calls.append((rows, server)), + service_id_columns={"deezer": {"track": "deezer_id"}}, + ) + + +def _payload(): + return { + "source_track_id": "deluxe-humble", + "album_source_id": "302127", + "total_discs": 1, + "expected_track": { + "title": "HUMBLE.", + "track_number": 8, + "disc_number": 1, + "duration": 177000, + "source": "deezer", + "track_id": "350171311", + "deezer_id": "350171311", + "artists": ["Kendrick Lamar"], + }, + } + + +def test_import_existing_track_copies_file_and_writes_target_album_row(tmp_path, monkeypatch): + db, conn = _make_db(include_disc_number=True) + source_file = tmp_path / "deluxe" / "08 - HUMBLE.flac" + source_file.parent.mkdir() + source_file.write_bytes(b"source audio") + sibling_file = tmp_path / "basic" / "01 - BLOOD.flac" + sibling_file.parent.mkdir() + sibling_file.write_bytes(b"sibling audio") + _insert_track(conn, track_id="basic-blood", album_id="album-basic", title="BLOOD.", track_number=1, file_path=sibling_file) + _insert_track(conn, track_id="deluxe-humble", album_id="album-deluxe", title="HUMBLE.", track_number=8, file_path=source_file) + + inherited = [] + monkeypatch.setattr(mti, "read_album_identity_tags", lambda path: {"musicbrainz_albumid": "target-release"} if path == str(sibling_file) else {}) + monkeypatch.setattr(mti, "write_album_identity_tags", lambda path, tags: inherited.append((path, tags)) or True) + + sync_calls = [] + result = mti.import_existing_track_for_album_slot("album-basic", _payload(), _deps(tmp_path, db, sync_calls=sync_calls)) + + assert source_file.read_bytes() == b"source audio" + assert os.path.exists(result["final_path"]) + assert inherited == [(result["final_path"], {"musicbrainz_albumid": "target-release"})] + + row = conn.execute("SELECT * FROM tracks WHERE album_id = 'album-basic' AND track_number = 8").fetchone() + assert row is not None + assert row["title"] == "HUMBLE." + assert row["disc_number"] == 1 + assert row["file_path"] == result["final_path"] + assert row["deezer_id"] == "350171311" + assert sync_calls and sync_calls[0][1] == "navidrome" + + +def test_import_adds_disc_number_column_for_older_track_tables(tmp_path, monkeypatch): + db, conn = _make_db(include_disc_number=False) + source_file = tmp_path / "deluxe" / "08 - HUMBLE.flac" + source_file.parent.mkdir() + source_file.write_bytes(b"source audio") + sibling_file = tmp_path / "basic" / "01 - BLOOD.flac" + sibling_file.parent.mkdir() + sibling_file.write_bytes(b"sibling audio") + _insert_track(conn, track_id="basic-blood", album_id="album-basic", title="BLOOD.", track_number=1, file_path=sibling_file) + _insert_track(conn, track_id="deluxe-humble", album_id="album-deluxe", title="HUMBLE.", track_number=8, file_path=source_file) + + write_calls = [] + monkeypatch.setattr(mti, "read_album_identity_tags", lambda path: {"musicbrainz_albumid": "target-release"} if path == str(sibling_file) else {}) + monkeypatch.setattr(mti, "write_album_identity_tags", lambda path, tags: write_calls.append((path, tags)) or True) + + result = mti.import_existing_track_for_album_slot("album-basic", _payload(), _deps(tmp_path, db)) + + columns = [row[1] for row in conn.execute("PRAGMA table_info(tracks)").fetchall()] + assert "disc_number" in columns + row = conn.execute("SELECT title, disc_number, file_path FROM tracks WHERE album_id = 'album-basic' AND track_number = 8").fetchone() + assert row["title"] == "HUMBLE." + assert row["disc_number"] == 1 + assert row["file_path"] == result["final_path"] + assert write_calls, "album identity inheritance should still run after old DB migration" + + +def test_copy_album_identity_uses_target_sibling_and_leaves_track_tags_to_imported_file(tmp_path, monkeypatch): + db, conn = _make_db(include_disc_number=True) + sibling_file = tmp_path / "basic" / "01 - BLOOD.flac" + sibling_file.parent.mkdir() + sibling_file.write_bytes(b"sibling") + final_file = tmp_path / "basic" / "08 - HUMBLE.flac" + final_file.write_bytes(b"imported") + _insert_track(conn, track_id="basic-blood", album_id="album-basic", title="BLOOD.", track_number=1, file_path=sibling_file) + + monkeypatch.setattr(mti, "read_album_identity_tags", lambda path: {"musicbrainz_albumid": "target-release", "barcode": "target-barcode"}) + writes = [] + monkeypatch.setattr(mti, "write_album_identity_tags", lambda path, tags: writes.append((path, tags)) or True) + + copied = mti.copy_album_identity_from_target_sibling( + db, + "album-basic", + str(final_file), + 1, + 8, + lambda path: str(path) if os.path.exists(path) else None, + ) + + assert copied is True + assert writes == [(str(final_file), {"musicbrainz_albumid": "target-release", "barcode": "target-barcode"})] + + +def test_import_rejects_missing_expected_track_context(tmp_path): + db, _conn = _make_db(include_disc_number=True) + with pytest.raises(mti.MissingTrackImportError) as exc: + mti.import_existing_track_for_album_slot("album-basic", {"source_track_id": "x", "expected_track": {}}, _deps(tmp_path, db)) + + assert exc.value.status_code == 400 + assert "expected_track" in str(exc.value)