from unittest.mock import patch import pytest from django.contrib.auth import get_user_model from django.urls import reverse from music.models import Album, Artist, Track from scrobbles.models import FavoriteMedia, Scrobble from scrobbles.tasks import enrich_media_from_webhook @pytest.mark.django_db @patch("music.models.resolve_track") def test_find_or_create_trust_webhook_data_skips_musicbrainz(mock_resolve): track = Track.find_or_create( title="TNT", artist_name="AC/DC", album_name="TNT", run_time_seconds=210, mbid="rec-123", artist_mbid="art-1", album_mbid="alb-1", trust_webhook_data=True, ) assert track.musicbrainz_id == "rec-123" assert track.base_run_time_seconds == 210 assert "musicbrainz-provider" in track.tags.names() assert "musicbrainz-enriched" not in track.tags.names() mock_resolve.assert_not_called() assert Artist.objects.filter(name="AC/DC", musicbrainz_id="art-1").exists() assert Album.objects.filter(name="TNT", musicbrainz_id="alb-1").exists() @pytest.mark.django_db @patch("music.models.resolve_track") def test_find_or_create_reuses_existing_track_by_mbid(mock_resolve): existing = Track.objects.create(title="TNT", musicbrainz_id="rec-123") track = Track.find_or_create( title="TNT", artist_name="AC/DC", mbid="rec-123", trust_webhook_data=True, ) assert track.id == existing.id mock_resolve.assert_not_called() @pytest.mark.django_db @patch("music.models.resolve_track") def test_fix_metadata_enriches_and_tags(mock_resolve): track = Track.objects.create(title="TNT") mock_resolve.return_value = ( {"recording_mbid": "rec-123", "length_ms": 210000}, "exact", ) assert track.fix_metadata() is True track.refresh_from_db() assert track.musicbrainz_id == "rec-123" assert track.base_run_time_seconds == 210 assert "musicbrainz-exact" in track.tags.names() assert "musicbrainz-enriched" in track.tags.names() @pytest.mark.django_db @patch("music.models.resolve_track", return_value=(None, "")) def test_fix_metadata_notfound_returns_false(mock_resolve): track = Track.objects.create(title="Unknown") assert track.fix_metadata() is False assert "musicbrainz-notfound" in track.tags.names() @pytest.mark.django_db def test_track_merge_repoints_scrobbles_and_favorites(): user = get_user_model().objects.create(email="merge@example.com") duplicate = Track.objects.create(title="TNT", musicbrainz_id="rec-123") canonical = Track.objects.create(title="TNT", musicbrainz_id="rec-123") duplicate.tags.add("cool") scrobble = Scrobble.objects.create( track=duplicate, media_type="Track", user=user, in_progress=False ) fav = FavoriteMedia.objects.create(user=user, track=duplicate, media_type="Track") repointed = Track.merge(duplicate, canonical) assert repointed == 1 scrobble.refresh_from_db() assert scrobble.track == canonical fav.refresh_from_db() assert fav.track == canonical assert "cool" in canonical.tags.names() assert not Track.objects.filter(id=duplicate.id).exists() @pytest.mark.django_db def test_track_merge_skips_in_progress_conflict(): user = get_user_model().objects.create(email="conflict@example.com") duplicate = Track.objects.create(title="TNT", musicbrainz_id="rec-123") canonical = Track.objects.create(title="TNT", musicbrainz_id="rec-123") Scrobble.objects.create( track=canonical, media_type="Track", user=user, in_progress=True ) scrobble = Scrobble.objects.create( track=duplicate, media_type="Track", user=user, in_progress=True ) repointed = Track.merge(duplicate, canonical) assert repointed == 0 scrobble.refresh_from_db() assert scrobble.track == duplicate assert Track.objects.filter(id=duplicate.id).exists() @pytest.mark.django_db @patch("scrobbles.tasks.enrich_media_from_webhook") def test_mopidy_webhook_creates_track_from_post_data_and_dispatches( mock_task, client, mopidy_track, valid_auth_token ): url = reverse("scrobbles:mopidy-webhook") headers = {"Authorization": f"Token {valid_auth_token}"} response = client.post( url, mopidy_track.request_data, content_type="application/json", headers=headers, ) assert response.status_code == 200 scrobble = Scrobble.objects.get(id=response.data["scrobble_id"]) assert scrobble.track.title == "Same in the End" assert scrobble.track.musicbrainz_id == mopidy_track.musicbrainz_track_id assert "webhook-metadata-only" in scrobble.tags.names() mock_task.delay.assert_called_once_with(scrobble.id) @pytest.mark.django_db @patch("scrobbles.tasks.enrich_media_from_webhook") def test_jellyfin_webhook_creates_track_from_post_data_and_dispatches( mock_task, client, jellyfin_track, valid_auth_token ): url = reverse("scrobbles:jellyfin-webhook") headers = {"Authorization": f"Token {valid_auth_token}"} response = client.post( url, jellyfin_track.request_json, content_type="application/json", headers=headers, ) assert response.status_code == 200 scrobble = Scrobble.objects.get(id=response.data["scrobble_id"]) assert scrobble.track.title == "Emotion" assert scrobble.track.musicbrainz_id == jellyfin_track.musicbrainz_track_id assert "webhook-metadata-only" in scrobble.tags.names() mock_task.delay.assert_called_once_with(scrobble.id) @pytest.mark.django_db @patch("music.models.resolve_track") def test_enrich_task_merges_onto_canonical_track(mock_resolve): user = get_user_model().objects.create(email="task@example.com") artist = Artist.objects.create(name="AC-DC") canonical = Track.objects.create( title="TNT", musicbrainz_id="rec-123", base_run_time_seconds=210 ) canonical.artists.add(artist) canonical.tags.add("musicbrainz-enriched") raw = Track.objects.create(title="TNT", artist_fk=artist) raw.artists.add(artist) scrobble = Scrobble.objects.create(track=raw, media_type="Track", user=user) mock_resolve.return_value = ( {"recording_mbid": "rec-123", "length_ms": 210000}, "exact", ) enrich_media_from_webhook(scrobble.id) scrobble.refresh_from_db() assert scrobble.track == canonical assert "webhook-enriched" in scrobble.tags.names() assert not Track.objects.filter(id=raw.id).exists() @pytest.mark.django_db @patch("music.models.resolve_track", return_value=(None, "")) def test_enrich_task_tags_failure(mock_resolve): user = get_user_model().objects.create(email="fail@example.com") track = Track.objects.create(title="Unknown Song") scrobble = Scrobble.objects.create(track=track, media_type="Track", user=user) enrich_media_from_webhook(scrobble.id) scrobble.refresh_from_db() assert "enrichment-failed" in scrobble.tags.names() assert "musicbrainz-notfound" in track.tags.names()