kopia lustrzana https://dev.funkwhale.audio/funkwhale/funkwhale
383 wiersze
13 KiB
Python
383 wiersze
13 KiB
Python
from django.core.paginator import Paginator
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
import pytest
|
|
|
|
from funkwhale_api.federation import actors
|
|
from funkwhale_api.federation import activity
|
|
from funkwhale_api.federation import models
|
|
from funkwhale_api.federation import serializers
|
|
from funkwhale_api.federation import utils
|
|
from funkwhale_api.federation import webfinger
|
|
|
|
|
|
@pytest.mark.parametrize('system_actor', actors.SYSTEM_ACTORS.keys())
|
|
def test_instance_actors(system_actor, db, api_client):
|
|
actor = actors.SYSTEM_ACTORS[system_actor].get_actor_instance()
|
|
url = reverse(
|
|
'federation:instance-actors-detail',
|
|
kwargs={'actor': system_actor})
|
|
response = api_client.get(url)
|
|
serializer = serializers.ActorSerializer(actor)
|
|
|
|
if system_actor == 'library':
|
|
response.data.pop('url')
|
|
assert response.status_code == 200
|
|
assert response.data == serializer.data
|
|
|
|
|
|
@pytest.mark.parametrize('route,kwargs', [
|
|
('instance-actors-outbox', {'actor': 'library'}),
|
|
('instance-actors-inbox', {'actor': 'library'}),
|
|
('instance-actors-detail', {'actor': 'library'}),
|
|
('well-known-webfinger', {}),
|
|
])
|
|
def test_instance_endpoints_405_if_federation_disabled(
|
|
authenticated_actor, db, preferences, api_client, route, kwargs):
|
|
preferences['federation__enabled'] = False
|
|
url = reverse('federation:{}'.format(route), kwargs=kwargs)
|
|
response = api_client.get(url)
|
|
|
|
assert response.status_code == 405
|
|
|
|
|
|
def test_wellknown_webfinger_validates_resource(
|
|
db, api_client, settings, mocker):
|
|
clean = mocker.spy(webfinger, 'clean_resource')
|
|
url = reverse('federation:well-known-webfinger')
|
|
response = api_client.get(url, data={'resource': 'something'})
|
|
|
|
clean.assert_called_once_with('something')
|
|
assert url == '/.well-known/webfinger'
|
|
assert response.status_code == 400
|
|
assert response.data['errors']['resource'] == (
|
|
'Missing webfinger resource type'
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize('system_actor', actors.SYSTEM_ACTORS.keys())
|
|
def test_wellknown_webfinger_system(
|
|
system_actor, db, api_client, settings, mocker):
|
|
actor = actors.SYSTEM_ACTORS[system_actor].get_actor_instance()
|
|
url = reverse('federation:well-known-webfinger')
|
|
response = api_client.get(
|
|
url, data={'resource': 'acct:{}'.format(actor.webfinger_subject)})
|
|
serializer = serializers.ActorWebfingerSerializer(actor)
|
|
|
|
assert response.status_code == 200
|
|
assert response['Content-Type'] == 'application/jrd+json'
|
|
assert response.data == serializer.data
|
|
|
|
|
|
def test_audio_file_list_requires_authenticated_actor(
|
|
db, preferences, api_client):
|
|
preferences['federation__music_needs_approval'] = True
|
|
url = reverse('federation:music:files-list')
|
|
response = api_client.get(url)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_audio_file_list_actor_no_page(
|
|
db, preferences, api_client, factories):
|
|
preferences['federation__music_needs_approval'] = False
|
|
preferences['federation__collection_page_size'] = 2
|
|
library = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
tfs = factories['music.TrackFile'].create_batch(size=5)
|
|
conf = {
|
|
'id': utils.full_url(reverse('federation:music:files-list')),
|
|
'page_size': 2,
|
|
'items': list(reversed(tfs)), # we order by -creation_date
|
|
'item_serializer': serializers.AudioSerializer,
|
|
'actor': library
|
|
}
|
|
expected = serializers.PaginatedCollectionSerializer(conf).data
|
|
url = reverse('federation:music:files-list')
|
|
response = api_client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.data == expected
|
|
|
|
|
|
def test_audio_file_list_actor_page(
|
|
db, preferences, api_client, factories):
|
|
preferences['federation__music_needs_approval'] = False
|
|
preferences['federation__collection_page_size'] = 2
|
|
library = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
tfs = factories['music.TrackFile'].create_batch(size=5)
|
|
conf = {
|
|
'id': utils.full_url(reverse('federation:music:files-list')),
|
|
'page': Paginator(list(reversed(tfs)), 2).page(2),
|
|
'item_serializer': serializers.AudioSerializer,
|
|
'actor': library
|
|
}
|
|
expected = serializers.CollectionPageSerializer(conf).data
|
|
url = reverse('federation:music:files-list')
|
|
response = api_client.get(url, data={'page': 2})
|
|
|
|
assert response.status_code == 200
|
|
assert response.data == expected
|
|
|
|
|
|
def test_audio_file_list_actor_page_exclude_federated_files(
|
|
db, preferences, api_client, factories):
|
|
preferences['federation__music_needs_approval'] = False
|
|
library = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
tfs = factories['music.TrackFile'].create_batch(size=5, federation=True)
|
|
|
|
url = reverse('federation:music:files-list')
|
|
response = api_client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.data['totalItems'] == 0
|
|
|
|
|
|
def test_audio_file_list_actor_page_error(
|
|
db, preferences, api_client, factories):
|
|
preferences['federation__music_needs_approval'] = False
|
|
url = reverse('federation:music:files-list')
|
|
response = api_client.get(url, data={'page': 'nope'})
|
|
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_audio_file_list_actor_page_error_too_far(
|
|
db, preferences, api_client, factories):
|
|
preferences['federation__music_needs_approval'] = False
|
|
url = reverse('federation:music:files-list')
|
|
response = api_client.get(url, data={'page': 5000})
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_library_actor_includes_library_link(db, preferences, api_client):
|
|
actor = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
url = reverse(
|
|
'federation:instance-actors-detail',
|
|
kwargs={'actor': 'library'})
|
|
response = api_client.get(url)
|
|
expected_links = [
|
|
{
|
|
'type': 'Link',
|
|
'name': 'library',
|
|
'mediaType': 'application/activity+json',
|
|
'href': utils.full_url(reverse('federation:music:files-list'))
|
|
}
|
|
]
|
|
assert response.status_code == 200
|
|
assert response.data['url'] == expected_links
|
|
|
|
|
|
def test_can_fetch_library(superuser_api_client, mocker):
|
|
result = {'test': 'test'}
|
|
scan = mocker.patch(
|
|
'funkwhale_api.federation.library.scan_from_account_name',
|
|
return_value=result)
|
|
|
|
url = reverse('api:v1:federation:libraries-fetch')
|
|
response = superuser_api_client.get(
|
|
url, data={'account': 'test@test.library'})
|
|
|
|
assert response.status_code == 200
|
|
assert response.data == result
|
|
scan.assert_called_once_with('test@test.library')
|
|
|
|
|
|
def test_follow_library(superuser_api_client, mocker, factories, r_mock):
|
|
library_actor = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
actor = factories['federation.Actor']()
|
|
follow = {'test': 'follow'}
|
|
on_commit = mocker.patch(
|
|
'funkwhale_api.common.utils.on_commit')
|
|
actor_data = serializers.ActorSerializer(actor).data
|
|
actor_data['url'] = [{
|
|
'href': 'https://test.library',
|
|
'name': 'library',
|
|
'type': 'Link',
|
|
}]
|
|
library_conf = {
|
|
'id': 'https://test.library',
|
|
'items': range(10),
|
|
'actor': actor,
|
|
'page_size': 5,
|
|
}
|
|
library_data = serializers.PaginatedCollectionSerializer(library_conf).data
|
|
r_mock.get(actor.url, json=actor_data)
|
|
r_mock.get('https://test.library', json=library_data)
|
|
data = {
|
|
'actor': actor.url,
|
|
'autoimport': False,
|
|
'federation_enabled': True,
|
|
'download_files': False,
|
|
}
|
|
|
|
url = reverse('api:v1:federation:libraries-list')
|
|
response = superuser_api_client.post(
|
|
url, data)
|
|
|
|
assert response.status_code == 201
|
|
|
|
follow = models.Follow.objects.get(
|
|
actor=library_actor,
|
|
target=actor,
|
|
approved=None,
|
|
)
|
|
library = follow.library
|
|
|
|
assert response.data == serializers.APILibraryCreateSerializer(
|
|
library).data
|
|
|
|
on_commit.assert_called_once_with(
|
|
activity.deliver,
|
|
serializers.FollowSerializer(follow).data,
|
|
on_behalf_of=library_actor,
|
|
to=[actor.url]
|
|
)
|
|
|
|
|
|
def test_can_list_system_actor_following(factories, superuser_api_client):
|
|
library_actor = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
follow1 = factories['federation.Follow'](actor=library_actor)
|
|
follow2 = factories['federation.Follow']()
|
|
|
|
url = reverse('api:v1:federation:libraries-following')
|
|
response = superuser_api_client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.data['results'] == [
|
|
serializers.APIFollowSerializer(follow1).data
|
|
]
|
|
|
|
|
|
def test_can_list_system_actor_followers(factories, superuser_api_client):
|
|
library_actor = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
follow1 = factories['federation.Follow'](actor=library_actor)
|
|
follow2 = factories['federation.Follow'](target=library_actor)
|
|
|
|
url = reverse('api:v1:federation:libraries-followers')
|
|
response = superuser_api_client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.data['results'] == [
|
|
serializers.APIFollowSerializer(follow2).data
|
|
]
|
|
|
|
|
|
def test_can_list_libraries(factories, superuser_api_client):
|
|
library1 = factories['federation.Library']()
|
|
library2 = factories['federation.Library']()
|
|
|
|
url = reverse('api:v1:federation:libraries-list')
|
|
response = superuser_api_client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.data['results'] == [
|
|
serializers.APILibrarySerializer(library1).data,
|
|
serializers.APILibrarySerializer(library2).data,
|
|
]
|
|
|
|
|
|
def test_can_detail_library(factories, superuser_api_client):
|
|
library = factories['federation.Library']()
|
|
|
|
url = reverse(
|
|
'api:v1:federation:libraries-detail',
|
|
kwargs={'uuid': str(library.uuid)})
|
|
response = superuser_api_client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.data == serializers.APILibrarySerializer(library).data
|
|
|
|
|
|
def test_can_patch_library(factories, superuser_api_client):
|
|
library = factories['federation.Library']()
|
|
data = {
|
|
'federation_enabled': not library.federation_enabled,
|
|
'download_files': not library.download_files,
|
|
'autoimport': not library.autoimport,
|
|
}
|
|
url = reverse(
|
|
'api:v1:federation:libraries-detail',
|
|
kwargs={'uuid': str(library.uuid)})
|
|
response = superuser_api_client.patch(url, data)
|
|
|
|
assert response.status_code == 200
|
|
library.refresh_from_db()
|
|
|
|
for k, v in data.items():
|
|
assert getattr(library, k) == v
|
|
|
|
|
|
def test_scan_library(factories, mocker, superuser_api_client):
|
|
scan = mocker.patch(
|
|
'funkwhale_api.federation.tasks.scan_library.delay',
|
|
return_value=mocker.Mock(id='id'))
|
|
library = factories['federation.Library']()
|
|
now = timezone.now()
|
|
data = {
|
|
'until': now,
|
|
}
|
|
url = reverse(
|
|
'api:v1:federation:libraries-scan',
|
|
kwargs={'uuid': str(library.uuid)})
|
|
response = superuser_api_client.post(url, data)
|
|
|
|
assert response.status_code == 200
|
|
assert response.data == {'task': 'id'}
|
|
scan.assert_called_once_with(
|
|
library_id=library.pk,
|
|
until=now
|
|
)
|
|
|
|
|
|
def test_list_library_tracks(factories, superuser_api_client):
|
|
library = factories['federation.Library']()
|
|
lts = list(reversed(factories['federation.LibraryTrack'].create_batch(
|
|
size=5, library=library)))
|
|
factories['federation.LibraryTrack'].create_batch(size=5)
|
|
url = reverse('api:v1:federation:library-tracks-list')
|
|
response = superuser_api_client.get(url, {'library': library.uuid})
|
|
|
|
assert response.status_code == 200
|
|
assert response.data == {
|
|
'results': serializers.APILibraryTrackSerializer(lts, many=True).data,
|
|
'count': 5,
|
|
'previous': None,
|
|
'next': None,
|
|
}
|
|
|
|
|
|
def test_can_update_follow_status(factories, superuser_api_client, mocker):
|
|
patched_accept = mocker.patch(
|
|
'funkwhale_api.federation.activity.accept_follow'
|
|
)
|
|
library_actor = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
follow = factories['federation.Follow'](target=library_actor)
|
|
|
|
payload = {
|
|
'follow': follow.pk,
|
|
'approved': True
|
|
}
|
|
url = reverse('api:v1:federation:libraries-followers')
|
|
response = superuser_api_client.patch(url, payload)
|
|
follow.refresh_from_db()
|
|
|
|
assert response.status_code == 200
|
|
assert follow.approved is True
|
|
patched_accept.assert_called_once_with(follow)
|
|
|
|
|
|
def test_can_filter_pending_follows(factories, superuser_api_client):
|
|
library_actor = actors.SYSTEM_ACTORS['library'].get_actor_instance()
|
|
follow = factories['federation.Follow'](
|
|
target=library_actor,
|
|
approved=True)
|
|
|
|
params = {'pending': True}
|
|
url = reverse('api:v1:federation:libraries-followers')
|
|
response = superuser_api_client.get(url, params)
|
|
|
|
assert response.status_code == 200
|
|
assert len(response.data['results']) == 0
|