vrobbler/tests/scrobbles_tests/test_webhook_enrichment.py

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_find_or_create_looks_up_by_mbid_before_title(mock_resolve):
    artist = Artist.objects.create(name="AC/DC")
    wrong = Track.objects.create(title="TNT", musicbrainz_id="rec-999")
    wrong.artists.add(artist)
    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
    assert track.id != wrong.id


@pytest.mark.django_db
@patch("music.models.resolve_track")
def test_find_or_create_title_fallback_skips_track_with_conflicting_mbid(
    mock_resolve,
):
    wrong = Track.objects.create(title="TNT", musicbrainz_id="rec-999")

    track = Track.find_or_create(
        title="TNT",
        artist_name="AC/DC",
        mbid="rec-123",
        trust_webhook_data=True,
    )

    assert track.id != wrong.id
    assert track.musicbrainz_id == "rec-123"
    wrong.refresh_from_db()
    assert wrong.musicbrainz_id == "rec-999"


@pytest.mark.django_db
@patch("music.models.resolve_track", return_value=(None, ""))
def test_find_or_create_falls_back_to_title_and_artist(mock_resolve):
    artist = Artist.objects.create(name="AC/DC")
    wrong = Track.objects.create(title="TNT")
    existing = Track.objects.create(title="TNT")
    existing.artists.add(artist)

    track = Track.find_or_create(
        title="TNT",
        artist_name="AC/DC",
    )

    assert track.id == existing.id
    assert track.id != wrong.id


@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()