Verified Commit 18d8baae authored by Eliot Berriot's avatar Eliot Berriot
Browse files

API Views/serializers/tests for activity (#141)

parent 1f2e14b2
from rest_framework import routers
from django.conf.urls import include, url
from funkwhale_api.activity import views as activity_views
from funkwhale_api.instance import views as instance_views
from funkwhale_api.music import views
from funkwhale_api.playlists import views as playlists_views
......@@ -10,6 +11,7 @@ from dynamic_preferences.users.viewsets import UserPreferencesViewSet
router = routers.SimpleRouter()
router.register(r'settings', GlobalPreferencesViewSet, base_name='settings')
router.register(r'activity', activity_views.ActivityViewSet, 'activity')
router.register(r'tags', views.TagViewSet, 'tags')
router.register(r'tracks', views.TrackViewSet, 'tracks')
router.register(r'trackfiles', views.TrackFileViewSet, 'trackfiles')
......
from rest_framework import serializers
from funkwhale_api.activity import record
class ModelSerializer(serializers.ModelSerializer):
id = serializers.CharField(source='get_activity_url')
......@@ -8,3 +10,15 @@ class ModelSerializer(serializers.ModelSerializer):
def get_url(self, obj):
return self.get_id(obj)
class AutoSerializer(serializers.Serializer):
"""
A serializer that will automatically use registered activity serializers
to serialize an henerogeneous list of objects (favorites, listenings, etc.)
"""
def to_representation(self, instance):
serializer = record.registry[instance._meta.label]['serializer'](
instance
)
return serializer.data
from django.db import models
from funkwhale_api.common import fields
from funkwhale_api.favorites.models import TrackFavorite
from funkwhale_api.history.models import Listening
def combined_recent(limit, **kwargs):
datetime_field = kwargs.pop('datetime_field', 'creation_date')
source_querysets = {
qs.model._meta.label: qs for qs in kwargs.pop('querysets')
}
querysets = {
k: qs.annotate(
__type=models.Value(
qs.model._meta.label, output_field=models.CharField()
)
).values('pk', datetime_field, '__type')
for k, qs in source_querysets.items()
}
_qs_list = list(querysets.values())
union_qs = _qs_list[0].union(*_qs_list[1:])
records = []
for row in union_qs.order_by('-{}'.format(datetime_field))[:limit]:
records.append({
'type': row['__type'],
'when': row[datetime_field],
'pk': row['pk']
})
# Now we bulk-load each object type in turn
to_load = {}
for record in records:
to_load.setdefault(record['type'], []).append(record['pk'])
fetched = {}
for key, pks in to_load.items():
for item in source_querysets[key].filter(pk__in=pks):
fetched[(key, item.pk)] = item
# Annotate 'records' with loaded objects
for record in records:
record['object'] = fetched[(record['type'], record['pk'])]
return records
def get_activity(user, limit=20):
query = fields.privacy_level_query(
user, lookup_field='user__privacy_level')
querysets = [
Listening.objects.filter(query).select_related(
'track',
'user',
'track__artist',
'track__album__artist',
),
TrackFavorite.objects.filter(query).select_related(
'track',
'user',
'track__artist',
'track__album__artist',
),
]
records = combined_recent(limit=limit, querysets=querysets)
return [r['object'] for r in records]
from rest_framework import viewsets
from rest_framework.response import Response
from funkwhale_api.common.permissions import ConditionalAuthentication
from funkwhale_api.favorites.models import TrackFavorite
from . import serializers
from . import utils
class ActivityViewSet(viewsets.GenericViewSet):
serializer_class = serializers.AutoSerializer
permission_classes = [ConditionalAuthentication]
queryset = TrackFavorite.objects.none()
def list(self, request, *args, **kwargs):
activity = utils.get_activity(user=request.user)
serializer = self.serializer_class(activity, many=True)
return Response({'results': serializer.data}, status=200)
from funkwhale_api.activity import serializers
from funkwhale_api.favorites.serializers import TrackFavoriteActivitySerializer
from funkwhale_api.history.serializers import \
ListeningActivitySerializer
def test_autoserializer(factories):
favorite = factories['favorites.TrackFavorite']()
listening = factories['history.Listening']()
objects = [favorite, listening]
serializer = serializers.AutoSerializer(objects, many=True)
expected = [
TrackFavoriteActivitySerializer(favorite).data,
ListeningActivitySerializer(listening).data,
]
assert serializer.data == expected
from funkwhale_api.activity import utils
def test_get_activity(factories):
user = factories['users.User']()
listening = factories['history.Listening']()
favorite = factories['favorites.TrackFavorite']()
objects = list(utils.get_activity(user))
assert objects == [favorite, listening]
def test_get_activity_honors_privacy_level(factories, anonymous_user):
listening = factories['history.Listening'](user__privacy_level='me')
favorite1 = factories['favorites.TrackFavorite'](
user__privacy_level='everyone')
favorite2 = factories['favorites.TrackFavorite'](
user__privacy_level='instance')
objects = list(utils.get_activity(anonymous_user))
assert objects == [favorite1]
from django.urls import reverse
from funkwhale_api.activity import serializers
from funkwhale_api.activity import utils
def test_activity_view(factories, api_client, settings, anonymous_user):
settings.API_AUTHENTICATION_REQUIRED = False
favorite = factories['favorites.TrackFavorite'](
user__privacy_level='everyone')
listening = factories['history.Listening']()
url = reverse('api:v1:activity-list')
objects = utils.get_activity(anonymous_user)
serializer = serializers.AutoSerializer(objects, many=True)
response = api_client.get(url)
assert response.status_code == 200
assert response.data['results'] == serializer.data
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment