danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Broaden import and metadata test coverage

Browse source 
- Cover search_result fallback normalization and ambiguous album detection.
- Add staging metadata, multi-disc path, and MusicBrainz enrichment cases.
- Move the single-track context test next to the imports code it exercises.
This commit is contained in:
Antti Kettunen 2026-04-26 12:13:46 +03:00
parent 4f236baa6d
commit 9315e74bea
No known key found for this signature in database
GPG key ID: C6B2A3D250359BD7
6 changed files with 367 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
core/imports/context.py
View file

@ -51,13 +51,14 @@ def normalize_import_context(context: Optional[Dict[str, Any]]) -> Dict[str, Any
track_info = _as_dict(context.get("track_info"))
original_search = _as_dict(context.get("original_search_result"))
search_result = _as_dict(context.get("search_result"))
normalized_search = original_search or search_result
if source:
context["source"] = source
context["artist"] = artist
context["album"] = album
context["track_info"] = track_info
context["original_search_result"] = original_search or search_result
context["original_search_result"] = normalized_search
context.pop("_source", None)
context.pop("spotify_artist", None)
context.pop("spotify_album", None)
@ -67,11 +68,11 @@ def normalize_import_context(context: Optional[Dict[str, Any]]) -> Dict[str, Any
("clean_album", "spotify_clean_album"),
("clean_artist", "spotify_clean_artist"),
):
if clean_key not in original_search or original_search.get(clean_key) in (None, ""):
legacy_value = original_search.get(legacy_key)
if clean_key not in normalized_search or normalized_search.get(clean_key) in (None, ""):
legacy_value = normalized_search.get(legacy_key)
if legacy_value not in (None, ""):
original_search[clean_key] = legacy_value
original_search.pop(legacy_key, None)
normalized_search[clean_key] = legacy_value
normalized_search.pop(legacy_key, None)
has_clean = bool(context.get("has_clean_metadata", context.get("has_clean_spotify_data", False)))
has_full = bool(context.get("has_full_metadata", context.get("has_full_spotify_metadata", False)))

93
tests/imports/test_import_context.py
View file

@ -2,6 +2,7 @@ import pytest
from core.imports.context import (
build_import_album_info,
detect_album_info_web,
get_import_clean_album,
get_import_clean_artist,
get_import_clean_title,
@ -79,6 +80,33 @@ def test_normalize_import_context_promotes_legacy_source_alias():
assert get_import_source(normalized) == "spotify"
def test_normalize_import_context_promotes_search_result_when_original_search_missing():
context = {
"source": "spotify",
"track_info": {"name": "Song One", "id": "track-1"},
"search_result": {
"title": "Song One",
"album": "Album One",
"artist": "Artist One",
"spotify_clean_title": "Song One",
"spotify_clean_album": "Album One",
"spotify_clean_artist": "Artist One",
},
}
normalized = normalize_import_context(context)
assert normalized["original_search_result"]["clean_title"] == "Song One"
assert normalized["original_search_result"]["clean_album"] == "Album One"
assert normalized["original_search_result"]["clean_artist"] == "Artist One"
assert "spotify_clean_title" not in normalized["original_search_result"]
assert "spotify_clean_album" not in normalized["original_search_result"]
assert "spotify_clean_artist" not in normalized["original_search_result"]
assert get_import_clean_title(normalized) == "Song One"
assert get_import_clean_album(normalized) == "Album One"
assert get_import_clean_artist(normalized) == "Artist One"
def test_neutral_import_context_helpers_work_without_legacy_aliases():
context = {
"source": "deezer",
@ -216,3 +244,68 @@ def test_build_import_album_info_uses_normalized_album_context():
assert album_info["clean_track_name"] == "Song One"
assert album_info["album_image_url"] == "https://img.example/album.jpg"
assert album_info["source"] == "deezer"
@pytest.mark.parametrize(
"album_name,track_name,artist_name",
[
("Song One", "Song One", "Artist One"),
("Artist One", "Different Song", "Artist One"),
],
)
def test_detect_album_info_web_returns_none_for_ambiguous_album_names(album_name, track_name, artist_name):
context = normalize_import_context(
{
"source": "deezer",
"artist": {"name": artist_name},
"album": {"name": album_name, "total_tracks": 12, "album_type": "album"},
"track_info": {"name": track_name, "track_number": 4, "disc_number": 1},
"original_search_result": {
"title": track_name,
"clean_title": track_name,
"clean_album": album_name,
"clean_artist": artist_name,
},
}
)
assert detect_album_info_web(context) is None
def test_detect_album_info_web_forces_album_when_track_and_artist_differ():
context = normalize_import_context(
{
"source": "deezer",
"artist": {"name": "Artist One"},
"album": {
"name": "Album One",
"image_url": "https://img.example/album.jpg",
"release_date": "2024-05-01",
"total_tracks": 1,
"album_type": "album",
},
"track_info": {
"name": "Song One",
"track_number": 4,
"disc_number": 2,
"duration_ms": 240000,
"artists": [{"name": "Artist One"}],
},
"original_search_result": {
"title": "Song One",
"album": "Album One",
"clean_title": "Song One",
"clean_album": "Album One",
"clean_artist": "Artist One",
},
}
)
album_info = detect_album_info_web(context)
assert album_info is not None
assert album_info["is_album"] is True
assert album_info["confidence"] == 0.5
assert album_info["album_name"] == "Album One"
assert album_info["track_number"] == 4
assert album_info["disc_number"] == 2

34
tests/imports/test_import_file_ops.py
View file

@ -90,3 +90,37 @@ def test_read_staging_file_metadata_falls_back_to_filename_track_number(monkeypa
assert metadata["title"] == "07 - Song Two"
assert metadata["track_number"] == 7
assert metadata["disc_number"] == 1
def test_read_staging_file_metadata_uses_filename_fallbacks_when_tags_are_invalid(monkeypatch, tmp_path):
file_path = tmp_path / "02 - Song Three.flac"
file_path.write_text("fake")
class DummyTags:
def __init__(self):
self.values = {
"title": [""],
"artist": "Artist One",
"albumartist": "",
"album": ["Album One"],
"tracknumber": ["not-a-number"],
"discnumber": ["bad/disc"],
}
def get(self, key, default=None):
return self.values.get(key, default)
fake_mutagen = types.ModuleType("mutagen")
fake_mutagen.File = lambda path, easy=True: DummyTags()
monkeypatch.setitem(sys.modules, "mutagen", fake_mutagen)
metadata = read_staging_file_metadata(str(file_path), file_path.name)
assert metadata == {
"title": "02 - Song Three",
"artist": "Artist One",
"albumartist": "Artist One",
"album": "Album One",
"track_number": 2,
"disc_number": 1,
}

69
tests/imports/test_import_paths.py
View file

@ -159,3 +159,72 @@ def test_build_final_path_for_track_uses_template_and_disc_folder(monkeypatch, t
tmp_path / "Transfer" / "Artist One" / "Artist One - Album One" / "Disc 1" / "04 - Song One.flac"
)
assert (tmp_path / "Transfer" / "Artist One" / "Artist One - Album One" / "Disc 1").is_dir()
def test_build_final_path_for_track_uses_track_disc_number_without_provider_lookup(monkeypatch, tmp_path):
config = _Config(
{
"soulseek.transfer_path": str(tmp_path / "Transfer"),
"file_organization.enabled": True,
"file_organization.templates": {
"album_path": "$albumartist/$albumartist - $album/$track - $title",
"single_path": "$artist/$artist - $title",
},
"file_organization.collab_artist_mode": "first",
"file_organization.disc_label": "Disc",
}
)
monkeypatch.setattr(import_paths, "_get_config_manager", lambda: config)
calls = []
monkeypatch.setattr(
import_paths,
"_get_album_tracks_for_source",
lambda source, album_id: calls.append((source, album_id)) or None,
)
context = {
"artist": {"name": "Artist One"},
"album": {
"name": "Album One",
"id": "album-1",
"release_date": "2026-01-01",
"total_tracks": 12,
"album_type": "album",
"artists": [{"name": "Artist One"}],
},
"track_info": {
"name": "Song Two",
"id": "track-2",
"track_number": 4,
"disc_number": 2,
"artists": [{"name": "Artist One"}],
},
"original_search_result": {
"title": "Song Two",
"clean_title": "Song Two",
"clean_album": "Album One",
"clean_artist": "Artist One",
"artists": [{"name": "Artist One"}],
},
"source": "deezer",
"is_album_download": False,
}
final_path, created = import_paths.build_final_path_for_track(
context,
{"name": "Artist One"},
{
"is_album": True,
"album_name": "Album One",
"track_number": 4,
"disc_number": 2,
},
".flac",
)
assert created is True
assert calls == []
assert final_path == str(
tmp_path / "Transfer" / "Artist One" / "Artist One - Album One" / "Disc 2" / "04 - Song Two.flac"
)

23
tests/metadata/test_metadata_service_single_import_context.py → tests/imports/test_import_resolution_single_track_context.py
View file

@ -192,3 +192,26 @@ def test_get_single_track_import_context_uses_explicit_override_source(monkeypat
]
assert deezer_client.calls == []
assert spotify_client.calls == []
def test_get_single_track_import_context_returns_fallback_payload_when_no_source_matches(monkeypatch):
deezer_client = FakeClient()
spotify_client = FakeClient()
itunes_client = FakeClient()
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(
metadata_service,
"get_client_for_source",
lambda source: {"deezer": deezer_client, "spotify": spotify_client, "itunes": itunes_client}.get(source),
)
result = resolution.get_single_track_import_context("Missing Song", "Missing Artist")
assert result["success"] is False
assert result["source"] is None
assert result["context"]["artist"]["name"] == "Missing Artist"
assert result["context"]["track_info"]["name"] == "Missing Song"
assert result["context"]["original_search_result"]["clean_title"] == "Missing Song"
assert result["context"]["original_search_result"]["clean_artist"] == "Missing Artist"

142
tests/metadata/test_metadata_enrichment.py
View file

@ -1,4 +1,5 @@
import types
import sqlite3
from core.metadata import enrichment as me
from core.metadata import artwork as ma
@ -73,6 +74,16 @@ class _FakeResponse:
return False
class _FileDB:
def __init__(self, db_path):
self.db_path = db_path
def _get_connection(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _fake_symbols(audio):
def _tag_factory(kind):
return lambda **kwargs: _FakeTag(kind, **kwargs)
@ -208,6 +219,137 @@ def test_embed_source_ids_uses_current_source_ids_and_legacy_fallback(monkeypatc
assert "ITUNES_ALBUM_ID" in legacy_descs
def test_embed_source_ids_skips_disabled_source_specific_tags(monkeypatch):
audio = _FakeAudio()
symbols = _fake_symbols(audio)
monkeypatch.setattr(
ms,
"get_config_manager",
lambda: _Config({"deezer.embed_tags": False}),
)
monkeypatch.setattr(ms, "get_mutagen_symbols", lambda: symbols)
monkeypatch.setattr(ms, "get_database", lambda: None)
metadata = {
"source": "deezer",
"source_track_id": "dz-track",
"source_artist_id": "dz-artist",
"source_album_id": "dz-album",
"title": "Song One",
"artist": "Artist One",
"album_artist": "Artist One",
"album": "Album One",
}
me.embed_source_ids(audio, metadata, context={"track_info": {}, "original_search_result": {}})
assert audio.tags.added == []
def test_embed_source_ids_writes_musicbrainz_release_year_and_updates_album_year(tmp_path, monkeypatch):
db_path = tmp_path / "music.db"
conn = sqlite3.connect(db_path)
conn.execute("CREATE TABLE artists (id TEXT PRIMARY KEY, name TEXT)")
conn.execute("CREATE TABLE albums (id TEXT PRIMARY KEY, artist_id TEXT, title TEXT, year INTEGER)")
conn.execute("INSERT INTO artists (id, name) VALUES (?, ?)", ("artist-1", "Artist One"))
conn.execute(
"INSERT INTO albums (id, artist_id, title, year) VALUES (?, ?, ?, ?)",
("album-1", "artist-1", "Album One", None),
)
conn.commit()
conn.close()
class _FakeMBClient:
def get_recording(self, mbid, includes=None):
return {
"isrcs": ["ISRC-123"],
"genres": [{"name": "Post Rock", "count": 10}],
}
def get_release(self, mbid, includes=None):
return {
"release-group": {
"id": "rg-1",
"primary-type": "album",
"first-release-date": "2021-09-17",
},
"artist-credit": [{"artist": {"id": "artist-mb-1"}}],
"status": "Official",
"country": "US",
"barcode": "1234567890",
"media": [{"format": "CD", "tracks": [{"position": 1, "id": "reltrack-1", "recording": {"id": "rec-1"}}]}],
"label-info": [{"catalog-number": "CAT-1"}],
"text-representation": {"script": "Latn"},
"asin": "ASIN1",
}
class _FakeMBService:
def __init__(self):
self.mb_client = _FakeMBClient()
def match_recording(self, title, artist):
return {"mbid": "rec-mbid"}
def match_artist(self, artist):
return {"mbid": "artist-mbid"}
def match_release(self, album, artist):
return {"mbid": "release-mbid"}
audio = _FakeAudio()
symbols = _fake_symbols(audio)
runtime = types.SimpleNamespace(mb_worker=types.SimpleNamespace(mb_service=_FakeMBService()))
monkeypatch.setattr(
ms,
"get_config_manager",
lambda: _Config(
{
"metadata_enhancement.enabled": True,
"metadata_enhancement.embed_album_art": False,
"metadata_enhancement.tags.write_multi_artist": False,
"musicbrainz.embed_tags": True,
}
),
)
monkeypatch.setattr(ms, "get_mutagen_symbols", lambda: symbols)
monkeypatch.setattr(ms, "get_database", lambda: _FileDB(str(db_path)))
metadata = {
"source": "musicbrainz",
"title": "Song One",
"artist": "Artist One",
"album_artist": "Artist One",
"album": "Album One",
"track_number": 1,
"total_tracks": 12,
"disc_number": 1,
}
me.embed_source_ids(audio, metadata, context={"track_info": {}, "original_search_result": {}}, runtime=runtime)
assert metadata["musicbrainz_release_id"] == "release-mbid"
assert metadata["date"] == "2021"
assert any(frame.kind == "TDRC" for frame in audio.tags.added)
assert any(frame.kind == "TSRC" for frame in audio.tags.added)
check = sqlite3.connect(db_path)
check.row_factory = sqlite3.Row
row = check.execute(
"""
SELECT albums.year
FROM albums
JOIN artists ON artists.id = albums.artist_id
WHERE albums.title = ? AND artists.name = ?
""",
("Album One", "Artist One"),
).fetchone()
check.close()
assert row["year"] == 2021
def test_enhance_file_metadata_forwards_runtime_to_source_embedding(monkeypatch):
audio = _FakeAudio()
symbols = _fake_symbols(audio)