"""Pin the post-process context dict the auto-import worker hands to ``_post_process_matched_download``. Background ---------- Auto-import doesn't write to the SoulSync standalone DB itself — it routes every matched track through the same ``_post_process_matched_download`` callback the regular download flow uses. The pipeline downstream (``record_soulsync_library_entry``, ``record_library_history_download``, ``record_download_provenance``) reads: - ``context["source"]`` — picks the right source-id columns (``spotify_track_id`` / ``deezer_id`` / ``itunes_track_id`` / etc.) - ``context["_download_username"]`` — labels the row in library history + provenance ("Auto-Import" instead of falling back to the Soulseek default). - ``context["track_info"]["musicbrainz_recording_id"]`` and ``context["track_info"]["isrc"]`` — per-recording IDs that land on the dedicated ``musicbrainz_recording_id`` / ``isrc`` track columns. If the worker drops any of these, the soulsync library row gets written but with NULL on every source-id column, and library history mislabels every imported file as a Soulseek download. SoulSync standalone is meant to be a full server replacement so it must reach parity with what a Plex / Jellyfin / Navidrome scan would write. These tests pin that contract at the worker boundary. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Dict, List from unittest.mock import MagicMock import pytest @dataclass class _FakeCandidate: path: str name: str audio_files: List[str] = field(default_factory=list) disc_structure: Dict[int, List[str]] = field(default_factory=dict) folder_hash: str = "fake-hash" is_single: bool = False @pytest.fixture def worker_with_capture(tmp_path): """Worker whose ``process_callback`` captures the per-track context dict so the test can assert on its shape directly.""" from core.auto_import_worker import AutoImportWorker captured: List[Dict[str, Any]] = [] fake_db = MagicMock() fake_cfg = MagicMock() fake_cfg.get.side_effect = lambda key, default=None: default def _capture(_key, ctx, _path): captured.append(ctx) worker = AutoImportWorker( database=fake_db, staging_path=str(tmp_path), transfer_path=str(tmp_path / "transfer"), process_callback=_capture, config_manager=fake_cfg, automation_engine=None, ) worker._captured = captured return worker def _make_match_result(source: str, track_count: int = 1) -> Dict[str, Any]: return { "matches": [], # filled by tests "unmatched_files": [], "total_tracks": track_count, "matched_count": track_count, "confidence": 0.95, "album_data": { "id": "ALBUM-ID-FROM-SOURCE", "total_tracks": track_count, "album_type": "album", "release_date": "2024-01-01", "images": [{"url": "https://img.example/cover.jpg"}], "artists": [{"name": "A", "id": "ARTIST-ID-FROM-SOURCE"}], }, } def _make_identification(source: str = "deezer") -> Dict[str, Any]: return { "source": source, "artist_name": "A", "artist_id": "ARTIST-ID-FROM-SOURCE", "album_name": "Album", "album_id": "ALBUM-ID-FROM-SOURCE", "image_url": "https://img.example/cover.jpg", "release_date": "2024-01-01", "method": "tags", } # --------------------------------------------------------------------------- # context["source"] propagation # --------------------------------------------------------------------------- @pytest.mark.parametrize("source", ["spotify", "deezer", "itunes", "discogs"]) def test_context_carries_source(worker_with_capture, tmp_path, source): """Worker must propagate ``identification['source']`` onto the top-level context. Without it, ``record_soulsync_library_entry`` can't pick a source column and writes the row with NULL on every source-id field.""" f = tmp_path / "01.flac" f.write_bytes(b"audio") cand = _FakeCandidate(path=str(tmp_path), name="Album") ident = _make_identification(source=source) mr = _make_match_result(source, 1) mr["matches"] = [{ "track": {"id": "t1", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}]}, "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) ctx = worker_with_capture._captured[0] assert ctx["source"] == source, ( f"Expected context['source']={source!r}, got {ctx.get('source')!r}. " f"Without this, soulsync library writes the row with NULL on " f"{source}_track_id." ) # --------------------------------------------------------------------------- # Auto-import labels: history + provenance must NOT default to Soulseek # --------------------------------------------------------------------------- def test_context_marks_download_username_as_auto_import(worker_with_capture, tmp_path): """``_download_username='auto_import'`` is what triggers the "Auto-Import" / "auto_import" branch in side_effects.py source maps. Without it, every imported file is labelled as a Soulseek download in library history + provenance — false signal in the UI.""" f = tmp_path / "01.flac" f.write_bytes(b"audio") cand = _FakeCandidate(path=str(tmp_path), name="Album") ident = _make_identification("spotify") mr = _make_match_result("spotify", 1) mr["matches"] = [{ "track": {"id": "t1", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}]}, "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) ctx = worker_with_capture._captured[0] assert ctx.get("_download_username") == "auto_import" # --------------------------------------------------------------------------- # Per-recording IDs flow through to track_info # --------------------------------------------------------------------------- def test_context_propagates_isrc_and_mbid_when_present(worker_with_capture, tmp_path): """When the metadata-source response carries per-recording IDs (Picard-tagged libraries always have MBID, MusicBrainz-enriched Spotify carries ISRC), they must end up on context['track_info']['isrc'] / ['musicbrainz_recording_id'] so the soulsync library row writes them onto dedicated columns.""" f = tmp_path / "01.flac" f.write_bytes(b"audio") cand = _FakeCandidate(path=str(tmp_path), name="Album") ident = _make_identification("spotify") mr = _make_match_result("spotify", 1) mr["matches"] = [{ "track": { "id": "spotify-track-id", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}], "isrc": "USABC1234567", "musicbrainz_recording_id": "abcd1234-mbid-uuid-form", }, "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) ti = worker_with_capture._captured[0]["track_info"] assert ti["isrc"] == "USABC1234567" assert ti["musicbrainz_recording_id"] == "abcd1234-mbid-uuid-form" def test_context_per_recording_ids_default_empty_when_missing(worker_with_capture, tmp_path): """Missing IDs default to empty string, NOT to None — the side- effects layer normalises to None at write time. Empty string keeps the field present in the dict so downstream code that does `track_info.get("isrc")` doesn't have to special-case missing keys.""" f = tmp_path / "01.flac" f.write_bytes(b"audio") cand = _FakeCandidate(path=str(tmp_path), name="Album") ident = _make_identification("deezer") mr = _make_match_result("deezer", 1) mr["matches"] = [{ "track": {"id": "111", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}]}, # no isrc / mbid "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) ti = worker_with_capture._captured[0]["track_info"] assert ti.get("isrc") == "" assert ti.get("musicbrainz_recording_id") == "" # --------------------------------------------------------------------------- # Album back-reference on track_info # --------------------------------------------------------------------------- def test_track_info_includes_album_id_back_reference(worker_with_capture, tmp_path): """`get_import_source_ids` reads `track_info.album_id` as one of the fallback paths for resolving the album-source-id. Without the back reference, sources whose API response nests album under `track.album.id` fall through and the soulsync row writes NULL on the album-source-id column.""" f = tmp_path / "01.flac" f.write_bytes(b"audio") cand = _FakeCandidate(path=str(tmp_path), name="Album") ident = _make_identification("spotify") mr = _make_match_result("spotify", 1) mr["matches"] = [{ "track": {"id": "t1", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}]}, "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) ti = worker_with_capture._captured[0]["track_info"] assert ti.get("album_id") == "ALBUM-ID-FROM-SOURCE" # --------------------------------------------------------------------------- # Artist source-id propagation — identification dict → context → DB write # --------------------------------------------------------------------------- def test_context_artist_id_uses_identification_artist_id(worker_with_capture, tmp_path): """When `identification` carries `artist_id` (from the metadata source's search response), it must end up on `context['spotify_artist']['id']` so the standalone library write populates the `_artist_id` column on the artists row. Before this fix the worker put `identification['album_id']` into that field — a copy-paste bug that wrote the album ID into the artist's source-ID column. Honest pin: artist_id flows from identification through to context, no falsey fallback.""" f = tmp_path / "01.flac" f.write_bytes(b"audio") cand = _FakeCandidate(path=str(tmp_path), name="Album") ident = _make_identification("spotify") ident["artist_id"] = "SPOTIFY-ARTIST-ID-XYZ" ident["album_id"] = "SPOTIFY-ALBUM-ID-DIFFERENT" mr = _make_match_result("spotify", 1) mr["matches"] = [{ "track": {"id": "t1", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}]}, "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) ctx = worker_with_capture._captured[0] assert ctx["spotify_artist"]["id"] == "SPOTIFY-ARTIST-ID-XYZ", ( "spotify_artist['id'] should hold the artist's source ID, NOT " "the album_id (regression case for the prior copy-paste bug)." ) # Album artists list must also carry the artist source ID so # `get_import_source_ids` can resolve it via the album→artists # fallback path. assert ctx["spotify_album"]["artists"][0]["id"] == "SPOTIFY-ARTIST-ID-XYZ" def test_context_artist_id_is_empty_when_identification_missing_it(worker_with_capture, tmp_path): """When the identification dict doesn't surface artist_id (e.g. filename-only identification fallback), context falls back to empty string — NOT to album_id (the prior wrong fallback).""" f = tmp_path / "01.flac" f.write_bytes(b"audio") cand = _FakeCandidate(path=str(tmp_path), name="Album") ident = _make_identification("spotify") ident.pop("artist_id", None) # force no artist_id ident["album_id"] = "SOME-ALBUM-ID" mr = _make_match_result("spotify", 1) mr["matches"] = [{ "track": {"id": "t1", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}]}, "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) ctx = worker_with_capture._captured[0] assert ctx["spotify_artist"]["id"] == "", ( "spotify_artist['id'] must be empty (NULL on the artists row) " "when the identification dict has no artist_id. It must NEVER " "fall back to album_id — that was the bug this PR fixed." ) # --------------------------------------------------------------------------- # Genre aggregation — soulsync standalone parity with deep-scan behaviour # --------------------------------------------------------------------------- def test_context_aggregates_genres_from_track_tags(worker_with_capture, tmp_path, monkeypatch): """Worker reads GENRE tag from each matched file and surfaces a deduped list on `spotify_artist['genres']`. Mirrors what `soulsync_client._scan_transfer` does at deep-scan time so the standalone library write populates the artists row's genres column instead of leaving it empty (which is what plex/jellyfin/ navidrome scans would have provided).""" from core import auto_import_worker as worker_mod files = [] for i in range(1, 4): f = tmp_path / f"0{i}.flac" f.write_bytes(b"audio") files.append(f) # Stub `_read_file_tags` so we don't need real audio. Each file # carries a different (overlapping) genre set — deduped result # should preserve insertion order + original casing. fake_tags = { str(files[0]): {'genres': ['Hip-Hop', 'Rap'], 'isrc': '', 'mbid': '', 'duration_ms': 200000, 'title': 'A', 'artist': 'X', 'album': 'Album', 'track_number': 1, 'disc_number': 1, 'year': ''}, str(files[1]): {'genres': ['Rap', 'Trap'], 'isrc': '', 'mbid': '', 'duration_ms': 200000, 'title': 'B', 'artist': 'X', 'album': 'Album', 'track_number': 2, 'disc_number': 1, 'year': ''}, str(files[2]): {'genres': ['hip-hop'], 'isrc': '', 'mbid': '', # case-insensitive dup 'duration_ms': 200000, 'title': 'C', 'artist': 'X', 'album': 'Album', 'track_number': 3, 'disc_number': 1, 'year': ''}, } monkeypatch.setattr(worker_mod, '_read_file_tags', lambda path: fake_tags.get(str(path), {'genres': []})) cand = _FakeCandidate(path=str(tmp_path), name="Album", audio_files=[str(f) for f in files]) ident = _make_identification("spotify") mr = _make_match_result("spotify", 3) mr["matches"] = [ {"track": {"id": f"t{i}", "name": f"Track {i}", "track_number": i, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "X"}]}, "file": str(files[i - 1]), "confidence": 0.95} for i in range(1, 4) ] worker_with_capture._process_matches(cand, ident, mr) ctx = worker_with_capture._captured[0] genres = ctx["spotify_artist"]["genres"] # Insertion-order preserved: Hip-Hop (file 1), Rap (file 1), Trap (file 2). # 'hip-hop' from file 3 deduped against 'Hip-Hop' (case-insensitive). assert genres == ["Hip-Hop", "Rap", "Trap"], ( f"Expected deduped insertion-order genres, got {genres}" ) def test_context_genres_empty_when_no_tags(worker_with_capture, tmp_path, monkeypatch): """No GENRE tag on any file → empty list. Standalone library write handles empty list gracefully (genres column stays empty / NULL).""" from core import auto_import_worker as worker_mod f = tmp_path / "01.flac" f.write_bytes(b"audio") monkeypatch.setattr(worker_mod, '_read_file_tags', lambda path: {'genres': [], 'isrc': '', 'mbid': '', 'duration_ms': 200000, 'title': '', 'artist': '', 'album': '', 'track_number': 1, 'disc_number': 1, 'year': ''}) cand = _FakeCandidate(path=str(tmp_path), name="Album", audio_files=[str(f)]) ident = _make_identification("spotify") mr = _make_match_result("spotify", 1) mr["matches"] = [{ "track": {"id": "t1", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}]}, "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) assert worker_with_capture._captured[0]["spotify_artist"]["genres"] == [] # --------------------------------------------------------------------------- # Defensive ISRC/MBID type coercion # --------------------------------------------------------------------------- def test_context_isrc_mbid_coerced_to_string(worker_with_capture, tmp_path): """If a metadata source returns ISRC or MBID as int / non-string (no current source does, but defensive against future drift), the worker coerces to string before assignment so the side-effects layer's `.strip()` doesn't AttributeError.""" f = tmp_path / "01.flac" f.write_bytes(b"audio") cand = _FakeCandidate(path=str(tmp_path), name="Album", audio_files=[str(f)]) ident = _make_identification("deezer") mr = _make_match_result("deezer", 1) mr["matches"] = [{ "track": { "id": "111", "name": "Track", "track_number": 1, "disc_number": 1, "duration_ms": 200000, "artists": [{"name": "A"}], # Hostile types: ints / None — must not propagate # through to side_effects un-cast. "isrc": 12345678, "mbid": None, "musicbrainz_recording_id": 999, }, "file": str(f), "confidence": 0.95, }] worker_with_capture._process_matches(cand, ident, mr) ti = worker_with_capture._captured[0]["track_info"] assert isinstance(ti["isrc"], str) assert isinstance(ti["musicbrainz_recording_id"], str) # int 12345678 → "12345678", int 999 → "999" assert ti["isrc"] == "12345678" assert ti["musicbrainz_recording_id"] == "999" def test_search_metadata_source_extracts_artist_id_from_dict_artist(): """`_search_metadata_source` must extract the artist source ID from `best_result.artists[0]['id']` so identification carries it forward. Without this, the worker's context-shape contract above is satisfied syntactically but the DB always sees empty.""" from core.auto_import_worker import AutoImportWorker, FolderCandidate from unittest.mock import patch, MagicMock fake_album = MagicMock() fake_album.id = "ALBUM-ID" fake_album.name = "Test Album" fake_album.artists = [{"id": "ARTIST-SRC-ID", "name": "Test Artist"}] fake_album.image_url = "https://img.example/cover.jpg" fake_album.release_date = "2024-01-01" fake_album.total_tracks = 10 fake_client = MagicMock() fake_client.search_albums.return_value = [fake_album] candidate = FolderCandidate( path="/staging/album", name="Test Album", audio_files=[f"/staging/album/0{i}.flac" for i in range(1, 11)], ) worker = AutoImportWorker(database=MagicMock(), process_callback=lambda *a, **k: None) with patch("core.metadata_service.get_primary_source", return_value="spotify"), \ patch("core.metadata_service.get_source_priority", return_value=["spotify"]), \ patch("core.metadata_service.get_client_for_source", return_value=fake_client): result = worker._search_metadata_source( "Test Artist", "Test Album", "tags", candidate, ) assert result is not None assert result.get("artist_id") == "ARTIST-SRC-ID", ( "_search_metadata_source must extract artist_id from " "best_result.artists[0]['id'] so the rest of the pipeline " "can write it to the artists table." )