222 lines
7.6 KiB
Python
222 lines
7.6 KiB
Python
from datetime import datetime, timedelta
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import pytz
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils import timezone
|
|
from music.models import Track
|
|
from scrobbles.importers.lastfm import LastFM, dispatch_historical_imports
|
|
from scrobbles.models import LastFmImport, Scrobble
|
|
|
|
User = get_user_model()
|
|
|
|
UTC = pytz.utc
|
|
|
|
|
|
@pytest.fixture
|
|
def lfm_user():
|
|
user = User.objects.create(username="lfmuser")
|
|
profile = user.profile
|
|
profile.lastfm_username = "lfmuser"
|
|
profile.lastfm_password = "secret"
|
|
profile.save()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_lastfm_network():
|
|
with patch("scrobbles.importers.lastfm.pylast.LastFMNetwork") as network, patch(
|
|
"scrobbles.importers.lastfm.pylast.md5", side_effect=lambda x: x
|
|
):
|
|
network.return_value.get_user.return_value = MagicMock()
|
|
yield network
|
|
|
|
|
|
def make_scrobble_dict(timestamp, title="Emotion", artist="Carly Rae Jepsen"):
|
|
return {
|
|
"artist": artist,
|
|
"album": "Emotion",
|
|
"title": title,
|
|
"mbid": None,
|
|
"run_time_seconds": 60,
|
|
"timestamp": timestamp,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_track_find_or_create():
|
|
with patch("scrobbles.importers.lastfm.Track.find_or_create") as mock_find:
|
|
|
|
def _create(**kwargs):
|
|
title = kwargs.get("title", "Emotion")
|
|
track = Track.objects.filter(title=title).first()
|
|
if not track:
|
|
track = Track.objects.create(
|
|
title=title,
|
|
base_run_time_seconds=kwargs.get("run_time_seconds"),
|
|
)
|
|
return track
|
|
|
|
mock_find.side_effect = _create
|
|
yield mock_find
|
|
|
|
|
|
class TestImportFromLastfm:
|
|
@pytest.mark.django_db
|
|
def test_import_creates_scrobble(
|
|
self, lfm_user, mock_lastfm_network, mock_track_find_or_create
|
|
):
|
|
timestamp = datetime(2023, 2, 15, 12, 0, 0, tzinfo=UTC)
|
|
lastfm = LastFM(lfm_user)
|
|
lastfm.get_last_scrobbles = MagicMock(
|
|
return_value=[make_scrobble_dict(timestamp)]
|
|
)
|
|
|
|
created = lastfm.import_from_lastfm()
|
|
|
|
assert len(created) == 1
|
|
scrobble = created[0]
|
|
assert scrobble.user == lfm_user
|
|
assert scrobble.timestamp == timestamp
|
|
assert scrobble.source == "Last.fm"
|
|
assert scrobble.media_type == "Track"
|
|
|
|
@pytest.mark.django_db
|
|
def test_import_skips_existing_scrobble_by_timestamp(
|
|
self, lfm_user, mock_lastfm_network, mock_track_find_or_create
|
|
):
|
|
timestamp = datetime(2023, 2, 15, 12, 0, 0, tzinfo=UTC)
|
|
track = Track.objects.create(title="Emotion", base_run_time_seconds=60)
|
|
existing = Scrobble.objects.create(
|
|
user=lfm_user,
|
|
timestamp=timestamp,
|
|
stop_timestamp=timestamp + timedelta(seconds=60),
|
|
source="Mopidy",
|
|
track=track,
|
|
media_type="Track",
|
|
)
|
|
# The existing scrobble was created long ago, not at import time.
|
|
Scrobble.objects.filter(pk=existing.pk).update(
|
|
created=timestamp - timedelta(days=365)
|
|
)
|
|
|
|
lastfm = LastFM(lfm_user)
|
|
lastfm.get_last_scrobbles = MagicMock(
|
|
return_value=[make_scrobble_dict(timestamp)]
|
|
)
|
|
|
|
created = lastfm.import_from_lastfm()
|
|
|
|
assert created == []
|
|
assert Scrobble.objects.filter(track=track).count() == 1
|
|
|
|
@pytest.mark.django_db
|
|
def test_import_creates_scrobble_when_only_created_close(
|
|
self, lfm_user, mock_lastfm_network, mock_track_find_or_create
|
|
):
|
|
timestamp = datetime(2023, 2, 15, 12, 0, 0, tzinfo=UTC)
|
|
track = Track.objects.create(title="Emotion", base_run_time_seconds=60)
|
|
# An existing scrobble for the same track but at a *different* play
|
|
# time (its row was created recently) must not block the new one.
|
|
Scrobble.objects.create(
|
|
user=lfm_user,
|
|
timestamp=timestamp - timedelta(hours=5),
|
|
stop_timestamp=timestamp - timedelta(hours=4, minutes=59),
|
|
source="Mopidy",
|
|
track=track,
|
|
media_type="Track",
|
|
)
|
|
|
|
lastfm = LastFM(lfm_user)
|
|
lastfm.get_last_scrobbles = MagicMock(
|
|
return_value=[make_scrobble_dict(timestamp)]
|
|
)
|
|
|
|
created = lastfm.import_from_lastfm()
|
|
|
|
assert len(created) == 1
|
|
assert Scrobble.objects.filter(track=track).count() == 2
|
|
|
|
|
|
class TestDispatchHistoricalImports:
|
|
def _seed_import(self, user, scrobble, processed_finished):
|
|
lfm_import = LastFmImport.objects.create(user=user)
|
|
lfm_import.record_log([scrobble])
|
|
lfm_import.mark_started()
|
|
LastFmImport.objects.filter(pk=lfm_import.pk).update(
|
|
processed_finished=processed_finished
|
|
)
|
|
return lfm_import
|
|
|
|
@pytest.mark.django_db
|
|
def test_resumes_before_earliest_scrobble_not_earliest_processed(
|
|
self, lfm_user, mock_lastfm_network
|
|
):
|
|
# Import A finished first but covers a LATER month; import B finished
|
|
# second but holds the EARLIEST scrobbles. This is the exact scenario
|
|
# from production (import 1 = April finished 19:33, import 5 = February).
|
|
feb_scrobble = Scrobble.objects.create(
|
|
user=lfm_user,
|
|
timestamp=datetime(2023, 2, 15, 12, 0, 0, tzinfo=UTC),
|
|
source="Last.fm",
|
|
media_type="Track",
|
|
)
|
|
apr_scrobble = Scrobble.objects.create(
|
|
user=lfm_user,
|
|
timestamp=datetime(2023, 4, 15, 12, 0, 0, tzinfo=UTC),
|
|
source="Last.fm",
|
|
media_type="Track",
|
|
)
|
|
now = timezone.now()
|
|
self._seed_import(lfm_user, apr_scrobble, now - timedelta(hours=2))
|
|
self._seed_import(lfm_user, feb_scrobble, now - timedelta(hours=1))
|
|
|
|
mock_lastfm = MagicMock()
|
|
mock_lastfm.get_earliest_scrobble_timestamp.return_value = datetime(
|
|
2022, 6, 1, tzinfo=UTC
|
|
)
|
|
|
|
def fake_get_last_scrobbles(time_from=None, time_to=None, check=False):
|
|
if time_from and time_from.month == 1 and time_from.year == 2023:
|
|
return True
|
|
return False
|
|
|
|
mock_lastfm.get_last_scrobbles = fake_get_last_scrobbles
|
|
|
|
with patch(
|
|
"scrobbles.importers.lastfm.LastFM", return_value=mock_lastfm
|
|
), patch("scrobbles.tasks.process_lastfm_import.delay") as mock_delay:
|
|
dispatched = dispatch_historical_imports(lfm_user.id)
|
|
|
|
assert dispatched == 1
|
|
mock_delay.assert_called_once()
|
|
args, kwargs = mock_delay.call_args
|
|
assert kwargs["time_from"] == datetime(2023, 1, 1, tzinfo=UTC)
|
|
assert kwargs["time_to"] == datetime(
|
|
2023, 1, 31, 23, 59, 59, 999999, tzinfo=UTC
|
|
)
|
|
|
|
# Only one new import (January 2023); February and April are untouched.
|
|
new_imports = LastFmImport.objects.filter(
|
|
user=lfm_user, processed_finished__isnull=True
|
|
)
|
|
assert new_imports.count() == 1
|
|
|
|
@pytest.mark.django_db
|
|
def test_no_existing_imports_starts_from_now(self, lfm_user, mock_lastfm_network):
|
|
mock_lastfm = MagicMock()
|
|
mock_lastfm.get_earliest_scrobble_timestamp.return_value = datetime(
|
|
2022, 6, 1, tzinfo=UTC
|
|
)
|
|
mock_lastfm.get_last_scrobbles = MagicMock(return_value=False)
|
|
|
|
with patch(
|
|
"scrobbles.importers.lastfm.LastFM", return_value=mock_lastfm
|
|
), patch("scrobbles.tasks.process_lastfm_import.delay") as mock_delay:
|
|
dispatched = dispatch_historical_imports(lfm_user.id)
|
|
|
|
assert dispatched == 0
|
|
mock_delay.assert_not_called()
|
|
assert not LastFmImport.objects.filter(user=lfm_user).exists()
|