Newer
Older
from django.core.paginator import Paginator
from django.urls import reverse
from django.utils import timezone
from funkwhale_api.federation import actors
from funkwhale_api.federation import activity
Eliot Berriot
committed
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.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(
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_wellknown_nodeinfo(db, preferences, api_client, settings):
expected = {
'links': [
{
'rel': 'http://nodeinfo.diaspora.software/ns/schema/2.0',
'href': '{}{}'.format(
settings.FUNKWHALE_URL,
reverse('api:v1:instance:nodeinfo-2.0')
)
}
]
}
url = reverse('federation:well-known-nodeinfo')
response = api_client.get(url)
assert response.status_code == 200
assert response['Content-Type'] == 'application/jrd+json'
assert response.data == expected
def test_wellknown_nodeinfo_disabled(db, preferences, api_client):
preferences['instance__nodeinfo_enabled'] = False
url = reverse('federation:well-known-nodeinfo')
response = api_client.get(url)
assert response.status_code == 404
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')
Eliot Berriot
committed
def test_follow_library(superuser_api_client, mocker, factories, r_mock):
Eliot Berriot
committed
library_actor = actors.SYSTEM_ACTORS['library'].get_actor_instance()
actor = factories['federation.Actor']()
Eliot Berriot
committed
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,
}
Eliot Berriot
committed
url = reverse('api:v1:federation:libraries-list')
response = superuser_api_client.post(
Eliot Berriot
committed
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
Eliot Berriot
committed
on_commit.assert_called_once_with(
activity.deliver,
Eliot Berriot
committed
serializers.FollowSerializer(follow).data,
on_behalf_of=library_actor,
to=[actor.url]
)
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
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,
]
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
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',
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,
}
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
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