novaburst/funquail
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Attribute artist

Browse source
This commit is contained in:
Eliot Berriot 2019-04-11 10:17:10 +02:00
commit 4e44e4e4b6
31 changed files with 1741 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

2
api/tests/common/test_decorators.py
View file

@ -44,7 +44,7 @@ def test_mutations_route_create_success(factories, api_request, is_approved, moc
on_commit = mocker.patch("funkwhale_api.common.utils.on_commit")
user = factories["users.User"](permission_library=True)
actor = user.create_actor()
track = factories["music.Track"](title="foo")
track = factories["music.Track"](title="foo", local=True)
view = V.as_view({"post": "mutations"})
request = api_request.post(

2
api/tests/common/test_mutations.py
View file

@ -10,7 +10,7 @@ def mutations_registry():
return mutations.Registry()
def test_apply_mutation(mutations_registry):
def test_apply_mutation(mutations_registry, db):
class Obj:
pass

43
api/tests/common/test_utils.py
View file

@ -1,3 +1,5 @@
import pytest
from funkwhale_api.common import utils
@ -42,3 +44,44 @@ def test_update_prefix(factories):
old = n.fid
n.refresh_from_db()
assert n.fid == old.replace("http://", "https://")
@pytest.mark.parametrize(
"conf, mock_args, data, expected",
[
(
["field1", "field2"],
{"field1": "foo", "field2": "test"},
{"field1": "bar"},
{"field1": "bar"},
),
(
["field1", "field2"],
{"field1": "foo", "field2": "test"},
{"field1": "foo"},
{},
),
(
["field1", "field2"],
{"field1": "foo", "field2": "test"},
{"field1": "foo", "field2": "test"},
{},
),
(
["field1", "field2"],
{"field1": "foo", "field2": "test"},
{"field1": "bar", "field2": "test1"},
{"field1": "bar", "field2": "test1"},
),
(
[("field1", "Hello"), ("field2", "World")],
{"Hello": "foo", "World": "test"},
{"field1": "bar", "field2": "test1"},
{"Hello": "bar", "World": "test1"},
),
],
)
def test_get_updated_fields(conf, mock_args, data, expected, mocker):
obj = mocker.Mock(**mock_args)
assert utils.get_updated_fields(conf, data, obj) == expected

47
api/tests/federation/test_activity.py
View file

@ -436,6 +436,53 @@ def test_prepare_deliveries_and_inbox_items(factories):
assert inbox_item.type == "to"
def test_prepare_deliveries_and_inbox_items_instances_with_followers(factories):
domain1 = factories["federation.Domain"](with_service_actor=True)
domain2 = factories["federation.Domain"](with_service_actor=True)
library = factories["music.Library"](actor__local=True)
factories["federation.LibraryFollow"](
target=library, actor__local=True, approved=True
).actor
library_follower_remote = factories["federation.LibraryFollow"](
target=library, actor__domain=domain1, approved=True
).actor
followed_actor = factories["federation.Actor"](local=True)
factories["federation.Follow"](
target=followed_actor, actor__local=True, approved=True
).actor
actor_follower_remote = factories["federation.Follow"](
target=followed_actor, actor__domain=domain2, approved=True
).actor
recipients = [activity.PUBLIC_ADDRESS, {"type": "instances_with_followers"}]
inbox_items, deliveries, urls = activity.prepare_deliveries_and_inbox_items(
recipients, "to"
)
expected_deliveries = sorted(
[
models.Delivery(
inbox_url=library_follower_remote.domain.service_actor.inbox_url
),
models.Delivery(
inbox_url=actor_follower_remote.domain.service_actor.inbox_url
),
],
key=lambda v: v.inbox_url,
)
assert inbox_items == []
assert len(expected_deliveries) == len(deliveries)
for delivery, expected_delivery in zip(
sorted(deliveries, key=lambda v: v.inbox_url), expected_deliveries
):
assert delivery.inbox_url == expected_delivery.inbox_url
def test_should_rotate_actor_key(settings, cache, now):
actor_id = 42
settings.ACTOR_KEY_ROTATION_DELAY = 10

30
api/tests/federation/test_models.py
View file

@ -134,3 +134,33 @@ def test_actor_stats(factories):
actor = factories["federation.Actor"]()
assert actor.get_stats() == expected
def test_actor_can_manage_false(mocker, factories):
obj = mocker.Mock()
actor = factories["federation.Actor"]()
assert actor.can_manage(obj) is False
def test_actor_can_manage_attributed_to(mocker, factories):
actor = factories["federation.Actor"]()
obj = mocker.Mock(attributed_to_id=actor.pk)
assert actor.can_manage(obj) is True
def test_actor_can_manage_domain_not_service_actor(mocker, factories):
actor = factories["federation.Actor"]()
obj = mocker.Mock(fid="https://{}/hello".format(actor.domain_id))
assert actor.can_manage(obj) is False
def test_actor_can_manage_domain_service_actor(mocker, factories):
actor = factories["federation.Actor"]()
actor.domain.service_actor = actor
actor.domain.save()
obj = mocker.Mock(fid="https://{}/hello".format(actor.domain_id))
assert actor.can_manage(obj) is True

92
api/tests/federation/test_routes.py
View file

@ -1,6 +1,6 @@
import pytest
from funkwhale_api.federation import jsonld, routes, serializers
from funkwhale_api.federation import actors, contexts, jsonld, routes, serializers
@pytest.mark.parametrize(
@ -13,6 +13,9 @@ from funkwhale_api.federation import jsonld, routes, serializers
({"type": "Delete", "object.type": "Library"}, routes.inbox_delete_library),
({"type": "Delete", "object.type": "Audio"}, routes.inbox_delete_audio),
({"type": "Undo", "object.type": "Follow"}, routes.inbox_undo_follow),
({"type": "Update", "object.type": "Artist"}, routes.inbox_update_artist),
({"type": "Update", "object.type": "Album"}, routes.inbox_update_album),
({"type": "Update", "object.type": "Track"}, routes.inbox_update_track),
],
)
def test_inbox_routes(route, handler):
@ -34,6 +37,7 @@ def test_inbox_routes(route, handler):
({"type": "Delete", "object.type": "Library"}, routes.outbox_delete_library),
({"type": "Delete", "object.type": "Audio"}, routes.outbox_delete_audio),
({"type": "Undo", "object.type": "Follow"}, routes.outbox_undo_follow),
({"type": "Update", "object.type": "Track"}, routes.outbox_update_track),
],
)
def test_outbox_routes(route, handler):
@ -405,3 +409,89 @@ def test_outbox_delete_follow_library(factories):
assert activity["actor"] == follow.actor
assert activity["object"] == follow
assert activity["related_object"] == follow.target
def test_handle_library_entry_update_can_manage(factories, mocker):
update_library_entity = mocker.patch(
"funkwhale_api.music.tasks.update_library_entity"
)
activity = factories["federation.Activity"]()
obj = factories["music.Artist"]()
actor = factories["federation.Actor"]()
mocker.patch.object(actor, "can_manage", return_value=False)
data = serializers.ArtistSerializer(obj).data
data["name"] = "New name"
payload = {"type": "Update", "actor": actor, "object": data}
routes.inbox_update_artist(
payload, context={"actor": actor, "raise_exception": True, "activity": activity}
)
update_library_entity.assert_not_called()
def test_inbox_update_artist(factories, mocker):
update_library_entity = mocker.patch(
"funkwhale_api.music.tasks.update_library_entity"
)
activity = factories["federation.Activity"]()
obj = factories["music.Artist"](attributed=True)
actor = obj.attributed_to
data = serializers.ArtistSerializer(obj).data
data["name"] = "New name"
payload = {"type": "Update", "actor": actor, "object": data}
routes.inbox_update_artist(
payload, context={"actor": actor, "raise_exception": True, "activity": activity}
)
update_library_entity.assert_called_once_with(obj, {"name": "New name"})
def test_inbox_update_album(factories, mocker):
update_library_entity = mocker.patch(
"funkwhale_api.music.tasks.update_library_entity"
)
activity = factories["federation.Activity"]()
obj = factories["music.Album"](attributed=True)
actor = obj.attributed_to
data = serializers.AlbumSerializer(obj).data
data["name"] = "New title"
payload = {"type": "Update", "actor": actor, "object": data}
routes.inbox_update_album(
payload, context={"actor": actor, "raise_exception": True, "activity": activity}
)
update_library_entity.assert_called_once_with(obj, {"title": "New title"})
def test_inbox_update_track(factories, mocker):
update_library_entity = mocker.patch(
"funkwhale_api.music.tasks.update_library_entity"
)
activity = factories["federation.Activity"]()
obj = factories["music.Track"](attributed=True)
actor = obj.attributed_to
data = serializers.TrackSerializer(obj).data
data["name"] = "New title"
payload = {"type": "Update", "actor": actor, "object": data}
routes.inbox_update_track(
payload, context={"actor": actor, "raise_exception": True, "activity": activity}
)
update_library_entity.assert_called_once_with(obj, {"title": "New title"})
def test_outbox_update_track(factories):
track = factories["music.Track"]()
activity = list(routes.outbox_update_track({"track": track}))[0]
expected = serializers.ActivitySerializer(
{"type": "Update", "object": serializers.TrackSerializer(track).data}
).data
expected["to"] = [contexts.AS.Public, {"type": "instances_with_followers"}]
assert dict(activity["payload"]) == dict(expected)
assert activity["actor"] == actors.get_service_actor()

873
api/tests/federation/test_serializers.py
View file

@ -1,13 +1,23 @@
import io
import pytest
import uuid
from django.core.paginator import Paginator
from django.utils import timezone
from funkwhale_api.federation import keys
from funkwhale_api.federation import jsonld
from funkwhale_api.federation import models
from funkwhale_api.federation import serializers
from funkwhale_api.federation import utils
from funkwhale_api.music import licenses
def test_actor_serializer_from_ap(db):
private, public = keys.get_key_pair()
actor_url = "https://test.federation/actor"
payload = {
"@context": jsonld.get_default_context(),
"@context": jsonld.get_default_context_fw(),
"id": actor_url,
"type": "Person",
"outbox": "https://test.com/outbox",
@ -47,3 +57,864 @@ def test_actor_serializer_from_ap(db):
assert actor.private_key is None
assert actor.public_key == payload["publicKey"]["publicKeyPem"]
assert actor.domain_id == "test.federation"
def test_actor_serializer_only_mandatory_field_from_ap(db):
payload = {
"@context": jsonld.get_default_context(),
"id": "https://test.federation/user",
"type": "Person",
"following": "https://test.federation/user/following",
"followers": "https://test.federation/user/followers",
"inbox": "https://test.federation/user/inbox",
"outbox": "https://test.federation/user/outbox",
"preferredUsername": "user",
}
serializer = serializers.ActorSerializer(data=payload)
assert serializer.is_valid(raise_exception=True)
actor = serializer.build()
assert actor.fid == payload["id"]
assert actor.inbox_url == payload["inbox"]
assert actor.outbox_url == payload["outbox"]
assert actor.followers_url == payload["followers"]
assert actor.following_url == payload["following"]
assert actor.preferred_username == payload["preferredUsername"]
assert actor.domain.pk == "test.federation"
assert actor.type == "Person"
assert actor.manually_approves_followers is None
def test_actor_serializer_to_ap():
expected = {
"@context": jsonld.get_default_context(),
"id": "https://test.federation/user",
"type": "Person",
"following": "https://test.federation/user/following",
"followers": "https://test.federation/user/followers",
"inbox": "https://test.federation/user/inbox",
"outbox": "https://test.federation/user/outbox",
"preferredUsername": "user",
"name": "Real User",
"summary": "Hello world",
"manuallyApprovesFollowers": False,
"publicKey": {
"id": "https://test.federation/user#main-key",
"owner": "https://test.federation/user",
"publicKeyPem": "yolo",
},
"endpoints": {"sharedInbox": "https://test.federation/inbox"},
}
ac = models.Actor(
fid=expected["id"],
inbox_url=expected["inbox"],
outbox_url=expected["outbox"],
shared_inbox_url=expected["endpoints"]["sharedInbox"],
followers_url=expected["followers"],
following_url=expected["following"],
public_key=expected["publicKey"]["publicKeyPem"],
preferred_username=expected["preferredUsername"],
name=expected["name"],
domain=models.Domain(pk="test.federation"),
summary=expected["summary"],
type="Person",
manually_approves_followers=False,
)
serializer = serializers.ActorSerializer(ac)
assert serializer.data == expected
def test_webfinger_serializer():
expected = {
"subject": "acct:service@test.federation",
"links": [
{
"rel": "self",
"href": "https://test.federation/federation/instance/actor",
"type": "application/activity+json",
}
],
"aliases": ["https://test.federation/federation/instance/actor"],
}
actor = models.Actor(
fid=expected["links"][0]["href"],
preferred_username="service",
domain=models.Domain(pk="test.federation"),
)
serializer = serializers.ActorWebfingerSerializer(actor)
assert serializer.data == expected
def test_follow_serializer_to_ap(factories):
follow = factories["federation.Follow"](local=True)
serializer = serializers.FollowSerializer(follow)
expected = {
"@context": jsonld.get_default_context(),
"id": follow.get_federation_id(),
"type": "Follow",
"actor": follow.actor.fid,
"object": follow.target.fid,
}
assert serializer.data == expected
def test_follow_serializer_save(factories):
actor = factories["federation.Actor"]()
target = factories["federation.Actor"]()
data = {
"id": "https://test.follow",
"type": "Follow",
"actor": actor.fid,
"object": target.fid,
}
serializer = serializers.FollowSerializer(data=data)
assert serializer.is_valid(raise_exception=True)
follow = serializer.save()
assert follow.pk is not None
assert follow.actor == actor
assert follow.target == target
assert follow.approved is None
def test_follow_serializer_save_validates_on_context(factories):
actor = factories["federation.Actor"]()
target = factories["federation.Actor"]()
impostor = factories["federation.Actor"]()
data = {
"id": "https://test.follow",
"type": "Follow",
"actor": actor.fid,
"object": target.fid,
}
serializer = serializers.FollowSerializer(
data=data, context={"follow_actor": impostor, "follow_target": impostor}
)
assert serializer.is_valid() is False
assert "actor" in serializer.errors
assert "object" in serializer.errors
def test_accept_follow_serializer_representation(factories):
follow = factories["federation.Follow"](approved=None)
expected = {
"@context": jsonld.get_default_context(),
"id": follow.get_federation_id() + "/accept",
"type": "Accept",
"actor": follow.target.fid,
"object": serializers.FollowSerializer(follow).data,
}
serializer = serializers.AcceptFollowSerializer(follow)
assert serializer.data == expected
def test_accept_follow_serializer_save(factories):
follow = factories["federation.Follow"](approved=None)
data = {
"@context": jsonld.get_default_context_fw(),
"id": follow.get_federation_id() + "/accept",
"type": "Accept",
"actor": follow.target.fid,
"object": serializers.FollowSerializer(follow).data,
}
serializer = serializers.AcceptFollowSerializer(data=data)
assert serializer.is_valid(raise_exception=True)
serializer.save()
follow.refresh_from_db()
assert follow.approved is True
def test_accept_follow_serializer_validates_on_context(factories):
follow = factories["federation.Follow"](approved=None)
impostor = factories["federation.Actor"]()
data = {
"@context": jsonld.get_default_context_fw(),
"id": follow.get_federation_id() + "/accept",
"type": "Accept",
"actor": impostor.url,
"object": serializers.FollowSerializer(follow).data,
}
serializer = serializers.AcceptFollowSerializer(
data=data, context={"follow_actor": impostor, "follow_target": impostor}
)
assert serializer.is_valid() is False
assert "actor" in serializer.errors["object"]
assert "object" in serializer.errors["object"]
def test_undo_follow_serializer_representation(factories):
follow = factories["federation.Follow"](approved=True)
expected = {
"@context": jsonld.get_default_context(),
"id": follow.get_federation_id() + "/undo",
"type": "Undo",
"actor": follow.actor.fid,
"object": serializers.FollowSerializer(follow).data,
}
serializer = serializers.UndoFollowSerializer(follow)
assert serializer.data == expected
def test_undo_follow_serializer_save(factories):
follow = factories["federation.Follow"](approved=True)
data = {
"@context": jsonld.get_default_context_fw(),
"id": follow.get_federation_id() + "/undo",
"type": "Undo",
"actor": follow.actor.fid,
"object": serializers.FollowSerializer(follow).data,
}
serializer = serializers.UndoFollowSerializer(data=data)
assert serializer.is_valid(raise_exception=True)
serializer.save()
with pytest.raises(models.Follow.DoesNotExist):
follow.refresh_from_db()
def test_undo_follow_serializer_validates_on_context(factories):
follow = factories["federation.Follow"](approved=True)
impostor = factories["federation.Actor"]()
data = {
"@context": jsonld.get_default_context_fw(),
"id": follow.get_federation_id() + "/undo",
"type": "Undo",
"actor": impostor.url,
"object": serializers.FollowSerializer(follow).data,
}
serializer = serializers.UndoFollowSerializer(
data=data, context={"follow_actor": impostor, "follow_target": impostor}
)
assert serializer.is_valid() is False
assert "actor" in serializer.errors["object"]
assert "object" in serializer.errors["object"]
def test_paginated_collection_serializer(factories):
uploads = factories["music.Upload"].create_batch(size=5)
actor = factories["federation.Actor"](local=True)
conf = {
"id": "https://test.federation/test",
"items": uploads,
"item_serializer": serializers.UploadSerializer,
"actor": actor,
"page_size": 2,
}
expected = {
"@context": jsonld.get_default_context(),
"type": "Collection",
"id": conf["id"],
"actor": actor.fid,
"totalItems": len(uploads),
"current": conf["id"] + "?page=1",
"last": conf["id"] + "?page=3",
"first": conf["id"] + "?page=1",
}
serializer = serializers.PaginatedCollectionSerializer(conf)
assert serializer.data == expected
def test_paginated_collection_serializer_validation():
data = {
"@context": jsonld.get_default_context_fw(),
"type": "Collection",
"id": "https://test.federation/test",
"totalItems": 5,
"actor": "http://test.actor",
"first": "https://test.federation/test?page=1",
"last": "https://test.federation/test?page=1",
"items": [],
}
serializer = serializers.PaginatedCollectionSerializer(data=data)
assert serializer.is_valid(raise_exception=True) is True
assert serializer.validated_data["totalItems"] == 5
assert serializer.validated_data["id"] == data["id"]
assert serializer.validated_data["actor"] == data["actor"]
def test_collection_page_serializer_validation():
base = "https://test.federation/test"
data = {
"@context": jsonld.get_default_context(),
"type": "CollectionPage",
"id": base + "?page=2",
"totalItems": 5,
"actor": "https://test.actor",
"items": [],
"first": "https://test.federation/test?page=1",
"last": "https://test.federation/test?page=3",
"prev": base + "?page=1",
"next": base + "?page=3",
"partOf": base,
}
serializer = serializers.CollectionPageSerializer(data=data)
assert serializer.is_valid(raise_exception=True) is True
assert serializer.validated_data["totalItems"] == 5
assert serializer.validated_data["id"] == data["id"]
assert serializer.validated_data["actor"] == data["actor"]
assert serializer.validated_data["items"] == []
assert serializer.validated_data["prev"] == data["prev"]
assert serializer.validated_data["next"] == data["next"]
assert serializer.validated_data["partOf"] == data["partOf"]
def test_collection_page_serializer_can_validate_child():
data = {
"@context": jsonld.get_default_context(),
"type": "CollectionPage",
"id": "https://test.page?page=2",
"actor": "https://test.actor",
"first": "https://test.page?page=1",
"last": "https://test.page?page=3",
"partOf": "https://test.page",
"totalItems": 1,
"items": [{"in": "valid"}],
}
serializer = serializers.CollectionPageSerializer(
data=data, context={"item_serializer": serializers.UploadSerializer}
)
# child are validated but not included in data if not valid
assert serializer.is_valid(raise_exception=True) is True
assert len(serializer.validated_data["items"]) == 0
def test_collection_page_serializer(factories):
uploads = factories["music.Upload"].create_batch(size=5)
actor = factories["federation.Actor"](local=True)
conf = {
"id": "https://test.federation/test",
"item_serializer": serializers.UploadSerializer,
"actor": actor,
"page": Paginator(uploads, 2).page(2),
}
expected = {
"@context": jsonld.get_default_context(),
"type": "CollectionPage",
"id": conf["id"] + "?page=2",
"actor": actor.fid,
"totalItems": len(uploads),
"partOf": conf["id"],
"prev": conf["id"] + "?page=1",
"next": conf["id"] + "?page=3",
"first": conf["id"] + "?page=1",
"last": conf["id"] + "?page=3",
"items": [
conf["item_serializer"](
i, context={"actor": actor, "include_ap_context": False}
).data
for i in conf["page"].object_list
],
}
serializer = serializers.CollectionPageSerializer(conf)
assert serializer.data == expected
def test_music_library_serializer_to_ap(factories):
library = factories["music.Library"](privacy_level="everyone")
# pending, errored and skippednot included
factories["music.Upload"](import_status="pending")
factories["music.Upload"](import_status="errored")
factories["music.Upload"](import_status="finished")
serializer = serializers.LibrarySerializer(library)
expected = {
"@context": jsonld.get_default_context(),
"audience": "https://www.w3.org/ns/activitystreams#Public",
"type": "Library",
"id": library.fid,
"name": library.name,
"summary": library.description,
"actor": library.actor.fid,
"totalItems": 0,
"current": library.fid + "?page=1",
"last": library.fid + "?page=1",
"first": library.fid + "?page=1",
"followers": library.followers_url,
}
assert serializer.data == expected
def test_music_library_serializer_from_public(factories, mocker):
actor = factories["federation.Actor"]()
retrieve = mocker.patch(
"funkwhale_api.federation.utils.retrieve_ap_object", return_value=actor
)
data = {
"@context": jsonld.get_default_context(),
"audience": "https://www.w3.org/ns/activitystreams#Public",
"name": "Hello",
"summary": "World",
"type": "Library",
"id": "https://library.id",
"followers": "https://library.id/followers",
"actor": actor.fid,
"totalItems": 12,
"first": "https://library.id?page=1",
"last": "https://library.id?page=2",
}
serializer = serializers.LibrarySerializer(data=data)
assert serializer.is_valid(raise_exception=True)
library = serializer.save()
assert library.actor == actor
assert library.fid == data["id"]
assert library.uploads_count == data["totalItems"]
assert library.privacy_level == "everyone"
assert library.name == "Hello"
assert library.description == "World"
assert library.followers_url == data["followers"]
retrieve.assert_called_once_with(
actor.fid,
actor=None,
queryset=actor.__class__,
serializer_class=serializers.ActorSerializer,
)
def test_music_library_serializer_from_private(factories, mocker):
actor = factories["federation.Actor"]()
retrieve = mocker.patch(
"funkwhale_api.federation.utils.retrieve_ap_object", return_value=actor
)
data = {
"@context": jsonld.get_default_context_fw(),
"audience": "",
"name": "Hello",
"summary": "World",
"type": "Library",
"id": "https://library.id",
"followers": "https://library.id/followers",
"actor": actor.fid,
"totalItems": 12,
"first": "https://library.id?page=1",
"last": "https://library.id?page=2",
}
serializer = serializers.LibrarySerializer(data=data)
assert serializer.is_valid(raise_exception=True)
library = serializer.save()
assert library.actor == actor
assert library.fid == data["id"]
assert library.uploads_count == data["totalItems"]
assert library.privacy_level == "me"
assert library.name == "Hello"
assert library.description == "World"
assert library.followers_url == data["followers"]
retrieve.assert_called_once_with(
actor.fid,
actor=None,
queryset=actor.__class__,
serializer_class=serializers.ActorSerializer,
)
def test_activity_pub_artist_serializer_to_ap(factories):
artist = factories["music.Artist"](attributed=True)
expected = {
"@context": serializers.AP_CONTEXT,
"type": "Artist",
"id": artist.fid,
"name": artist.name,
"musicbrainzId": artist.mbid,
"published": artist.creation_date.isoformat(),
"attributedTo": artist.attributed_to.fid,
}
serializer = serializers.ArtistSerializer(artist)
assert serializer.data == expected
def test_activity_pub_album_serializer_to_ap(factories):
album = factories["music.Album"](attributed=True)
expected = {
"@context": serializers.AP_CONTEXT,
"type": "Album",
"id": album.fid,
"name": album.title,
"cover": {
"type": "Link",
"mediaType": "image/jpeg",
"href": utils.full_url(album.cover.url),
},
"musicbrainzId": album.mbid,
"published": album.creation_date.isoformat(),
"released": album.release_date.isoformat(),
"artists": [
serializers.ArtistSerializer(
album.artist, context={"include_ap_context": False}
).data
],
"attributedTo": album.attributed_to.fid,
}
serializer = serializers.AlbumSerializer(album)
assert serializer.data == expected
def test_activity_pub_track_serializer_to_ap(factories):
track = factories["music.Track"](
license="cc-by-4.0", copyright="test", disc_number=3, attributed=True
)
expected = {
"@context": serializers.AP_CONTEXT,
"published": track.creation_date.isoformat(),
"type": "Track",
"musicbrainzId": track.mbid,
"id": track.fid,
"name": track.title,
"position": track.position,
"disc": track.disc_number,
"license": track.license.conf["identifiers"][0],
"copyright": "test",
"artists": [
serializers.ArtistSerializer(
track.artist, context={"include_ap_context": False}
).data
],
"album": serializers.AlbumSerializer(
track.album, context={"include_ap_context": False}
).data,
"attributedTo": track.attributed_to.fid,
}
serializer = serializers.TrackSerializer(track)
assert serializer.data == expected
def test_activity_pub_track_serializer_from_ap(factories, r_mock, mocker):
track_attributed_to = factories["federation.Actor"]()
album_attributed_to = factories["federation.Actor"]()
album_artist_attributed_to = factories["federation.Actor"]()
artist_attributed_to = factories["federation.Actor"]()
activity = factories["federation.Activity"]()
published = timezone.now()
released = timezone.now().date()
data = {
"@context": jsonld.get_default_context(),
"type": "Track",
"id": "http://hello.track",
"published": published.isoformat(),
"musicbrainzId": str(uuid.uuid4()),
"name": "Black in back",
"position": 5,
"disc": 1,
"attributedTo": track_attributed_to.fid,
"album": {
"type": "Album",
"id": "http://hello.album",
"name": "Purple album",
"musicbrainzId": str(uuid.uuid4()),
"published": published.isoformat(),
"released": released.isoformat(),
"attributedTo": album_attributed_to.fid,
"cover": {
"type": "Link",
"href": "https://cover.image/test.png",
"mediaType": "image/png",
},
"artists": [
{
"type": "Artist",
"id": "http://hello.artist",
"name": "John Smith",
"musicbrainzId": str(uuid.uuid4()),
"published": published.isoformat(),
"attributedTo": album_artist_attributed_to.fid,
}
],
},
"artists": [
{
"type": "Artist",
"id": "http://hello.trackartist",
"name": "Bob Smith",
"musicbrainzId": str(uuid.uuid4()),
"attributedTo": artist_attributed_to.fid,
"published": published.isoformat(),
}
],
}
r_mock.get(data["album"]["cover"]["href"], body=io.BytesIO(b"coucou"))
serializer = serializers.TrackSerializer(data=data, context={"activity": activity})
assert serializer.is_valid(raise_exception=True)
track = serializer.save()
album = track.album
artist = track.artist
album_artist = track.album.artist
assert track.from_activity == activity
assert track.fid == data["id"]
assert track.title == data["name"]
assert track.position == data["position"]
assert track.disc_number == data["disc"]
assert track.creation_date == published
assert track.attributed_to == track_attributed_to
assert str(track.mbid) == data["musicbrainzId"]
assert album.from_activity == activity
assert album.cover.read() == b"coucou"
assert album.cover.path.endswith(".png")
assert album.title == data["album"]["name"]
assert album.fid == data["album"]["id"]
assert str(album.mbid) == data["album"]["musicbrainzId"]
assert album.creation_date == published
assert album.release_date == released
assert album.attributed_to == album_attributed_to
assert artist.from_activity == activity
assert artist.name == data["artists"][0]["name"]
assert artist.fid == data["artists"][0]["id"]
assert str(artist.mbid) == data["artists"][0]["musicbrainzId"]
assert artist.creation_date == published
assert artist.attributed_to == artist_attributed_to
assert album_artist.from_activity == activity
assert album_artist.name == data["album"]["artists"][0]["name"]
assert album_artist.fid == data["album"]["artists"][0]["id"]
assert str(album_artist.mbid) == data["album"]["artists"][0]["musicbrainzId"]
assert album_artist.creation_date == published
assert album_artist.attributed_to == album_artist_attributed_to
def test_activity_pub_upload_serializer_from_ap(factories, mocker, r_mock):
activity = factories["federation.Activity"]()
library = factories["music.Library"]()
published = timezone.now()
updated = timezone.now()
released = timezone.now().date()
data = {
"@context": serializers.AP_CONTEXT,
"type": "Audio",
"id": "https://track.file",
"name": "Ignored",
"published": published.isoformat(),
"updated": updated.isoformat(),
"duration": 43,
"bitrate": 42,
"size": 66,
"url": {"href": "https://audio.file", "type": "Link", "mediaType": "audio/mp3"},
"library": library.fid,
"track": {
"type": "Track",
"id": "http://hello.track",
"published": published.isoformat(),
"musicbrainzId": str(uuid.uuid4()),
"name": "Black in back",
"position": 5,
"album": {
"type": "Album",
"id": "http://hello.album",
"name": "Purple album",
"musicbrainzId": str(uuid.uuid4()),
"published": published.isoformat(),
"released": released.isoformat(),
"cover": {
"type": "Link",
"href": "https://cover.image/test.png",
"mediaType": "image/png",
},
"artists": [
{
"type": "Artist",
"id": "http://hello.artist",
"name": "John Smith",
"musicbrainzId": str(uuid.uuid4()),
"published": published.isoformat(),
}
],
},
"artists": [
{
"type": "Artist",
"id": "http://hello.trackartist",
"name": "Bob Smith",
"musicbrainzId": str(uuid.uuid4()),
"published": published.isoformat(),
}
],
},
}
r_mock.get(data["track"]["album"]["cover"]["href"], body=io.BytesIO(b"coucou"))
serializer = serializers.UploadSerializer(data=data, context={"activity": activity})
assert serializer.is_valid(raise_exception=True)
track_create = mocker.spy(serializers.TrackSerializer, "create")
upload = serializer.save()
assert upload.track.from_activity == activity
assert upload.from_activity == activity
assert track_create.call_count == 1
assert upload.fid == data["id"]
assert upload.track.fid == data["track"]["id"]
assert upload.duration == data["duration"]
assert upload.size == data["size"]
assert upload.bitrate == data["bitrate"]
assert upload.source == data["url"]["href"]
assert upload.mimetype == data["url"]["mediaType"]
assert upload.creation_date == published
assert upload.import_status == "finished"
assert upload.modification_date == updated
def test_activity_pub_upload_serializer_validtes_library_actor(factories, mocker):
library = factories["music.Library"]()
usurpator = factories["federation.Actor"]()
serializer = serializers.UploadSerializer(data={}, context={"actor": usurpator})
with pytest.raises(serializers.serializers.ValidationError):
serializer.validate_library(library.fid)
def test_activity_pub_audio_serializer_to_ap(factories):
upload = factories["music.Upload"](
mimetype="audio/mp3", bitrate=42, duration=43, size=44
)
expected = {
"@context": serializers.AP_CONTEXT,
"type": "Audio",
"id": upload.fid,
"name": upload.track.full_name,
"published": upload.creation_date.isoformat(),
"updated": upload.modification_date.isoformat(),
"duration": upload.duration,
"bitrate": upload.bitrate,
"size": upload.size,
"url": {
"href": utils.full_url(upload.listen_url),
"type": "Link",
"mediaType": "audio/mp3",
},
"library": upload.library.fid,
"track": serializers.TrackSerializer(
upload.track, context={"include_ap_context": False}
).data,
}
serializer = serializers.UploadSerializer(upload)
assert serializer.data == expected
def test_local_actor_serializer_to_ap(factories):
expected = {
"@context": jsonld.get_default_context(),
"id": "https://test.federation/user",
"type": "Person",
"following": "https://test.federation/user/following",
"followers": "https://test.federation/user/followers",
"inbox": "https://test.federation/user/inbox",
"outbox": "https://test.federation/user/outbox",
"preferredUsername": "user",
"name": "Real User",
"summary": "Hello world",
"manuallyApprovesFollowers": False,
"publicKey": {
"id": "https://test.federation/user#main-key",
"owner": "https://test.federation/user",
"publicKeyPem": "yolo",
},
"endpoints": {"sharedInbox": "https://test.federation/inbox"},
}
ac = models.Actor.objects.create(
fid=expected["id"],
inbox_url=expected["inbox"],
outbox_url=expected["outbox"],
shared_inbox_url=expected["endpoints"]["sharedInbox"],
followers_url=expected["followers"],
following_url=expected["following"],
public_key=expected["publicKey"]["publicKeyPem"],
preferred_username=expected["preferredUsername"],
name=expected["name"],
domain=models.Domain.objects.create(pk="test.federation"),
summary=expected["summary"],
type="Person",
manually_approves_followers=False,
)
user = factories["users.User"]()
user.actor = ac
user.save()
ac.refresh_from_db()
expected["icon"] = {
"type": "Image",
"mediaType": "image/jpeg",
"url": utils.full_url(user.avatar.crop["400x400"].url),
}
serializer = serializers.ActorSerializer(ac)
assert serializer.data == expected
def test_activity_serializer_validate_recipients_empty(db):
s = serializers.BaseActivitySerializer()
with pytest.raises(serializers.serializers.ValidationError):
s.validate_recipients({})
with pytest.raises(serializers.serializers.ValidationError):
s.validate_recipients({"to": []})
with pytest.raises(serializers.serializers.ValidationError):
s.validate_recipients({"cc": []})
def test_track_serializer_update_license(factories):
licenses.load(licenses.LICENSES)
obj = factories["music.Track"](license=None)
serializer = serializers.TrackSerializer()
serializer.update(obj, {"license": "http://creativecommons.org/licenses/by/2.0/"})
obj.refresh_from_db()
assert obj.license_id == "cc-by-2.0"

15
api/tests/music/test_models.py
View file

@ -533,3 +533,18 @@ def test_queryset_local_entities(factories, settings, factory):
factories[factory](fid="https://noope/3")
assert list(obj1.__class__.objects.local().order_by("id")) == [obj1, obj2]
@pytest.mark.parametrize(
"federation_hostname, fid, expected",
[
("test.domain", "http://test.domain/", True),
("test.domain", None, True),
("test.domain", "https://test.domain/", True),
("test.otherdomain", "http://test.domain/", False),
],
)
def test_api_model_mixin_is_local(federation_hostname, fid, expected, settings):
settings.FEDERATION_HOSTNAME = federation_hostname
obj = models.Track(fid=fid)
assert obj.is_local is expected

13
api/tests/music/test_mutations.py
View file

@ -56,3 +56,16 @@ def test_track_position_mutation(factories):
track.refresh_from_db()
assert track.position == 12
def test_track_mutation_apply_outbox(factories, mocker):
dispatch = mocker.patch("funkwhale_api.federation.routes.outbox.dispatch")
track = factories["music.Track"](position=4)
mutation = factories["common.Mutation"](
type="update", target=track, payload={"position": 12}
)
mutation.apply()
dispatch.assert_called_once_with(
{"type": "Update", "object": {"type": "Track"}}, context={"track": track}
)

10
api/tests/music/test_serializers.py
View file

@ -34,6 +34,7 @@ def test_artist_album_serializer(factories, to_api_date):
album = album.__class__.objects.with_tracks_count().get(pk=album.pk)
expected = {
"id": album.id,
"fid": album.fid,
"mbid": str(album.mbid),
"title": album.title,
"artist": album.artist.id,
@ -47,6 +48,7 @@ def test_artist_album_serializer(factories, to_api_date):
"small_square_crop": album.cover.crop["50x50"].url,
},
"release_date": to_api_date(album.release_date),
"is_local": album.is_local,
}
serializer = serializers.ArtistAlbumSerializer(album)
@ -61,8 +63,10 @@ def test_artist_with_albums_serializer(factories, to_api_date):
expected = {
"id": artist.id,
"fid": artist.fid,
"mbid": str(artist.mbid),
"name": artist.name,
"is_local": artist.is_local,
"creation_date": to_api_date(artist.creation_date),
"albums": [serializers.ArtistAlbumSerializer(album).data],
}
@ -79,6 +83,7 @@ def test_album_track_serializer(factories, to_api_date):
expected = {
"id": track.id,
"fid": track.fid,
"artist": serializers.ArtistSimpleSerializer(track.artist).data,
"album": track.album.id,
"mbid": str(track.mbid),
@ -91,6 +96,7 @@ def test_album_track_serializer(factories, to_api_date):
"duration": None,
"license": track.license.code,
"copyright": track.copyright,
"is_local": track.is_local,
}
serializer = serializers.AlbumTrackSerializer(track)
assert serializer.data == expected
@ -154,6 +160,7 @@ def test_album_serializer(factories, to_api_date):
album = track1.album
expected = {
"id": album.id,
"fid": album.fid,
"mbid": str(album.mbid),
"title": album.title,
"artist": serializers.ArtistSimpleSerializer(album.artist).data,
@ -167,6 +174,7 @@ def test_album_serializer(factories, to_api_date):
},
"release_date": to_api_date(album.release_date),
"tracks": serializers.AlbumTrackSerializer([track2, track1], many=True).data,
"is_local": album.is_local,
}
serializer = serializers.AlbumSerializer(album)
@ -181,6 +189,7 @@ def test_track_serializer(factories, to_api_date):
setattr(track, "playable_uploads", [upload])
expected = {
"id": track.id,
"fid": track.fid,
"artist": serializers.ArtistSimpleSerializer(track.artist).data,
"album": serializers.TrackAlbumSerializer(track.album).data,
"mbid": str(track.mbid),
@ -193,6 +202,7 @@ def test_track_serializer(factories, to_api_date):
"listen_url": track.listen_url,
"license": upload.track.license.code,
"copyright": upload.track.copyright,
"is_local": upload.track.is_local,
}
serializer = serializers.TrackSerializer(track)
assert serializer.data == expected

65
api/tests/music/test_tasks.py
View file

@ -42,9 +42,38 @@ def test_can_create_track_from_file_metadata_no_mbid(db, mocker):
assert track.album.release_date == datetime.date(2012, 8, 15)
assert track.artist.name == metadata["artists"][0]["name"]
assert track.artist.mbid is None
assert track.artist.attributed_to is None
match_license.assert_called_once_with(metadata["license"], metadata["copyright"])
def test_can_create_track_from_file_metadata_attributed_to(factories, mocker):
actor = factories["federation.Actor"]()
metadata = {
"title": "Test track",
"artists": [{"name": "Test artist"}],
"album": {"title": "Test album", "release_date": datetime.date(2012, 8, 15)},
"position": 4,
"disc_number": 2,
"copyright": "2018 Someone",
}
track = tasks.get_track_from_import_metadata(metadata, attributed_to=actor)
assert track.title == metadata["title"]
assert track.mbid is None
assert track.position == 4
assert track.disc_number == 2
assert track.copyright == metadata["copyright"]
assert track.attributed_to == actor
assert track.album.title == metadata["album"]["title"]
assert track.album.mbid is None
assert track.album.release_date == datetime.date(2012, 8, 15)
assert track.album.attributed_to == actor
assert track.artist.name == metadata["artists"][0]["name"]
assert track.artist.mbid is None
assert track.artist.attributed_to == actor
def test_can_create_track_from_file_metadata_mbid(factories, mocker):
metadata = {
"title": "Test track",
@ -229,6 +258,7 @@ def test_upload_import(now, factories, temp_signal, mocker):
outbox = mocker.patch("funkwhale_api.federation.routes.outbox.dispatch")
update_album_cover = mocker.patch("funkwhale_api.music.tasks.update_album_cover")
get_picture = mocker.patch("funkwhale_api.music.metadata.Metadata.get_picture")
get_track_from_import_metadata = mocker.spy(tasks, "get_track_from_import_metadata")
track = factories["music.Track"](album__cover="")
upload = factories["music.Upload"](
track=None, import_metadata={"funkwhale": {"track": {"uuid": str(track.uuid)}}}
@ -246,6 +276,10 @@ def test_upload_import(now, factories, temp_signal, mocker):
update_album_cover.assert_called_once_with(
upload.track.album, cover_data=get_picture.return_value, source=upload.source
)
assert (
get_track_from_import_metadata.call_args[-1]["attributed_to"]
== upload.library.actor
)
handler.assert_called_once_with(
upload=upload,
old_status="pending",
@ -478,9 +512,15 @@ def test_update_album_cover_file_cover_separate_file(ext, mimetype, factories, m
)
def test_federation_audio_track_to_metadata(now):
def test_federation_audio_track_to_metadata(now, mocker):
published = now
released = now.date()
references = {
"http://track.attributed": mocker.Mock(),
"http://album.attributed": mocker.Mock(),
"http://album-artist.attributed": mocker.Mock(),
"http://artist.attributed": mocker.Mock(),
}
payload = {
"@context": jsonld.get_default_context(),
"type": "Track",
@ -492,6 +532,7 @@ def test_federation_audio_track_to_metadata(now):
"published": published.isoformat(),
"license": "http://creativecommons.org/licenses/by-sa/4.0/",
"copyright": "2018 Someone",
"attributedTo": "http://track.attributed",
"album": {
"published": published.isoformat(),
"type": "Album",
@ -499,6 +540,7 @@ def test_federation_audio_track_to_metadata(now):
"name": "Purple album",
"musicbrainzId": str(uuid.uuid4()),
"released": released.isoformat(),
"attributedTo": "http://album.attributed",
"artists": [
{
"type": "Artist",
@ -506,6 +548,7 @@ def test_federation_audio_track_to_metadata(now):
"id": "http://hello.artist",
"name": "John Smith",
"musicbrainzId": str(uuid.uuid4()),
"attributedTo": "http://album-artist.attributed",
}
],
"cover": {
@ -521,6 +564,7 @@ def test_federation_audio_track_to_metadata(now):
"id": "http://hello.trackartist",
"name": "Bob Smith",
"musicbrainzId": str(uuid.uuid4()),
"attributedTo": "http://artist.attributed",
}
],
}
@ -535,8 +579,10 @@ def test_federation_audio_track_to_metadata(now):
"mbid": payload["musicbrainzId"],
"fdate": serializer.validated_data["published"],
"fid": payload["id"],
"attributed_to": references["http://track.attributed"],
"album": {
"title": payload["album"]["name"],
"attributed_to": references["http://album.attributed"],
"release_date": released,
"mbid": payload["album"]["musicbrainzId"],
"fid": payload["album"]["id"],
@ -546,6 +592,7 @@ def test_federation_audio_track_to_metadata(now):
"name": a["name"],
"mbid": a["musicbrainzId"],
"fid": a["id"],
"attributed_to": references["http://album-artist.attributed"],
"fdate": serializer.validated_data["album"]["artists"][i][
"published"
],
@ -561,6 +608,7 @@ def test_federation_audio_track_to_metadata(now):
"mbid": a["musicbrainzId"],
"fid": a["id"],
"fdate": serializer.validated_data["artists"][i]["published"],
"attributed_to": references["http://artist.attributed"],
}
for i, a in enumerate(payload["artists"])
],
@ -570,7 +618,9 @@ def test_federation_audio_track_to_metadata(now):
},
}
result = tasks.federation_audio_track_to_metadata(serializer.validated_data)
result = tasks.federation_audio_track_to_metadata(
serializer.validated_data, references
)
assert result == expected
@ -747,3 +797,14 @@ def test_get_prunable_artists(factories):
factories["music.Track"](album__artist=non_prunable_album_artist)
assert list(tasks.get_prunable_artists()) == [prunable_artist]
def test_update_library_entity(factories, mocker):
artist = factories["music.Artist"]()
save = mocker.spy(artist, "save")
tasks.update_library_entity(artist, {"name": "Hello"})
save.assert_called_once_with(update_fields=["name"])
artist.refresh_from_db()
assert artist.name == "Hello"