141 lines
4.8 KiB
Python
141 lines
4.8 KiB
Python
import datetime
|
|
|
|
from django.core.paginator import Paginator
|
|
from django.utils import timezone
|
|
|
|
from funkwhale_api.federation import serializers
|
|
from funkwhale_api.federation import tasks
|
|
|
|
|
|
def test_scan_library_does_nothing_if_federation_disabled(mocker, factories):
|
|
library = factories['federation.Library'](federation_enabled=False)
|
|
tasks.scan_library(library_id=library.pk)
|
|
|
|
assert library.tracks.count() == 0
|
|
|
|
|
|
def test_scan_library_page_does_nothing_if_federation_disabled(
|
|
mocker, factories):
|
|
library = factories['federation.Library'](federation_enabled=False)
|
|
tasks.scan_library_page(library_id=library.pk, page_url=None)
|
|
|
|
assert library.tracks.count() == 0
|
|
|
|
|
|
def test_scan_library_fetches_page_and_calls_scan_page(
|
|
mocker, factories, r_mock):
|
|
now = timezone.now()
|
|
library = factories['federation.Library'](federation_enabled=True)
|
|
collection_conf = {
|
|
'actor': library.actor,
|
|
'id': library.url,
|
|
'page_size': 10,
|
|
'items': range(10),
|
|
}
|
|
collection = serializers.PaginatedCollectionSerializer(collection_conf)
|
|
scan_page = mocker.patch(
|
|
'funkwhale_api.federation.tasks.scan_library_page.delay')
|
|
r_mock.get(collection_conf['id'], json=collection.data)
|
|
tasks.scan_library(library_id=library.pk)
|
|
|
|
scan_page.assert_called_once_with(
|
|
library_id=library.id,
|
|
page_url=collection.data['first'],
|
|
until=None,
|
|
)
|
|
library.refresh_from_db()
|
|
assert library.fetched_date > now
|
|
|
|
|
|
def test_scan_page_fetches_page_and_creates_tracks(
|
|
mocker, factories, r_mock):
|
|
library = factories['federation.Library'](federation_enabled=True)
|
|
tfs = factories['music.TrackFile'].create_batch(size=5)
|
|
page_conf = {
|
|
'actor': library.actor,
|
|
'id': library.url,
|
|
'page': Paginator(tfs, 5).page(1),
|
|
'item_serializer': serializers.AudioSerializer,
|
|
}
|
|
page = serializers.CollectionPageSerializer(page_conf)
|
|
r_mock.get(page.data['id'], json=page.data)
|
|
|
|
tasks.scan_library_page(library_id=library.pk, page_url=page.data['id'])
|
|
|
|
lts = list(library.tracks.all().order_by('-published_date'))
|
|
assert len(lts) == 5
|
|
|
|
|
|
def test_scan_page_trigger_next_page_scan_skip_if_same(
|
|
mocker, factories, r_mock):
|
|
patched_scan = mocker.patch(
|
|
'funkwhale_api.federation.tasks.scan_library_page.delay'
|
|
)
|
|
library = factories['federation.Library'](federation_enabled=True)
|
|
tfs = factories['music.TrackFile'].create_batch(size=1)
|
|
page_conf = {
|
|
'actor': library.actor,
|
|
'id': library.url,
|
|
'page': Paginator(tfs, 3).page(1),
|
|
'item_serializer': serializers.AudioSerializer,
|
|
}
|
|
page = serializers.CollectionPageSerializer(page_conf)
|
|
data = page.data
|
|
data['next'] = data['id']
|
|
r_mock.get(page.data['id'], json=data)
|
|
|
|
tasks.scan_library_page(library_id=library.pk, page_url=data['id'])
|
|
patched_scan.assert_not_called()
|
|
|
|
|
|
def test_scan_page_stops_once_until_is_reached(
|
|
mocker, factories, r_mock):
|
|
library = factories['federation.Library'](federation_enabled=True)
|
|
tfs = list(reversed(factories['music.TrackFile'].create_batch(size=5)))
|
|
page_conf = {
|
|
'actor': library.actor,
|
|
'id': library.url,
|
|
'page': Paginator(tfs, 3).page(1),
|
|
'item_serializer': serializers.AudioSerializer,
|
|
}
|
|
page = serializers.CollectionPageSerializer(page_conf)
|
|
r_mock.get(page.data['id'], json=page.data)
|
|
|
|
tasks.scan_library_page(
|
|
library_id=library.pk,
|
|
page_url=page.data['id'],
|
|
until=tfs[1].creation_date)
|
|
|
|
lts = list(library.tracks.all().order_by('-published_date'))
|
|
assert len(lts) == 2
|
|
for i, tf in enumerate(tfs[:1]):
|
|
assert tf.creation_date == lts[i].published_date
|
|
|
|
|
|
def test_clean_federation_music_cache_if_no_listen(preferences, factories):
|
|
preferences['federation__music_cache_duration'] = 60
|
|
lt1 = factories['federation.LibraryTrack'](with_audio_file=True)
|
|
lt2 = factories['federation.LibraryTrack'](with_audio_file=True)
|
|
lt3 = factories['federation.LibraryTrack'](with_audio_file=True)
|
|
tf1 = factories['music.TrackFile'](library_track=lt1)
|
|
tf2 = factories['music.TrackFile'](library_track=lt2)
|
|
tf3 = factories['music.TrackFile'](library_track=lt3)
|
|
|
|
# we listen to the first one, and the second one (but weeks ago)
|
|
listening1 = factories['history.Listening'](
|
|
track=tf1.track,
|
|
creation_date=timezone.now())
|
|
listening2 = factories['history.Listening'](
|
|
track=tf2.track,
|
|
creation_date=timezone.now() - datetime.timedelta(minutes=61))
|
|
|
|
tasks.clean_music_cache()
|
|
|
|
lt1.refresh_from_db()
|
|
lt2.refresh_from_db()
|
|
lt3.refresh_from_db()
|
|
|
|
assert bool(lt1.audio_file) is True
|
|
assert bool(lt2.audio_file) is False
|
|
assert bool(lt3.audio_file) is False
|