Senko-san
c72d19599a
Implements the §6.2 enrichment pipeline: embedded tags → Chromaprint
fingerprint → AcoustID lookup. Well-tagged files get correct
artist/album/title offline; the rest are identified via AcoustID
(which also yields a MusicBrainz recording id in one call).
- domain: AudioTags/Fingerprint/RecordingMatch value objects; ports
AudioTagReader, AudioFingerprinter, AcoustIdClient; TrackRepository
.apply_enrichment (gap-fill, never erases) + AlbumRepository.get_or_create
- infrastructure/metadata: MutagenTagReader, FpcalcFingerprinter,
AcoustIdHttpClient (rich meta=recordings+releasegroups, throttled)
- application: MetadataEnrichmentService — tags preferred, AcoustID fills
gaps; resolves artist/album; status enriched/failed; skips manual;
every external step wrapped (graceful degradation)
- workers: enrich_task registered; enqueue_enrich is best-effort and
deferred so the caller's txn commits before the worker reads the row
- wiring: upload enqueues after add; import returns imported_ids and
enqueues post-commit (mid-scan would race the worker); manual
POST /tracks/{id}/metadata/enrich endpoint
- deps: add mutagen (fpcalc/ffmpeg already in the image)
Tests: metadata service orchestration, AcoustID parser, tag helpers.
125 passed; mypy strict + ruff clean.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
"""Unit tests for the AcoustID response parser — pure, no network."""
|
|
|
|
from app.infrastructure.metadata.acoustid import _parse_best_match
|
|
|
|
|
|
def _payload_with_results(results: list[object]) -> dict[str, object]:
|
|
return {"status": "ok", "results": results}
|
|
|
|
|
|
def test_parses_full_recording() -> None:
|
|
payload = _payload_with_results(
|
|
[
|
|
{
|
|
"id": "acoustid-1",
|
|
"score": 0.97,
|
|
"recordings": [
|
|
{
|
|
"id": "mb-rec-1",
|
|
"title": "One More Time",
|
|
"artists": [{"id": "a1", "name": "Daft Punk"}],
|
|
"releasegroups": [{"id": "rg1", "title": "Discovery"}],
|
|
}
|
|
],
|
|
}
|
|
]
|
|
)
|
|
|
|
match = _parse_best_match(payload)
|
|
|
|
assert match is not None
|
|
assert match.acoustid == "acoustid-1"
|
|
assert match.recording_mbid == "mb-rec-1"
|
|
assert match.title == "One More Time"
|
|
assert match.artist == "Daft Punk"
|
|
assert match.album == "Discovery"
|
|
assert match.score == 0.97
|
|
|
|
|
|
def test_picks_highest_score() -> None:
|
|
payload = _payload_with_results(
|
|
[
|
|
{"id": "low", "score": 0.40, "recordings": [{"id": "r-low", "title": "Low"}]},
|
|
{"id": "high", "score": 0.92, "recordings": [{"id": "r-high", "title": "High"}]},
|
|
]
|
|
)
|
|
|
|
match = _parse_best_match(payload)
|
|
|
|
assert match is not None
|
|
assert match.acoustid == "high"
|
|
assert match.title == "High"
|
|
|
|
|
|
def test_result_without_recordings_still_returns_id() -> None:
|
|
payload = _payload_with_results([{"id": "acoustid-only", "score": 0.5}])
|
|
|
|
match = _parse_best_match(payload)
|
|
|
|
assert match is not None
|
|
assert match.acoustid == "acoustid-only"
|
|
assert match.recording_mbid is None
|
|
assert match.title is None
|
|
|
|
|
|
def test_error_status_returns_none() -> None:
|
|
assert _parse_best_match({"status": "error", "error": {"message": "bad"}}) is None
|
|
|
|
|
|
def test_empty_results_returns_none() -> None:
|
|
assert _parse_best_match(_payload_with_results([])) is None
|
|
|
|
|
|
def test_non_dict_payload_returns_none() -> None:
|
|
assert _parse_best_match("nonsense") is None
|
|
assert _parse_best_match(None) is None
|