Verified Commit 78d0de0e authored by Eliot Berriot's avatar Eliot Berriot
Browse files

Merge branch 'release/0.8'

parents 3673f624 1a5b7ed2
......@@ -10,7 +10,7 @@ from funkwhale_api.users.factories import UserFactory
@pytest.mark.parametrize('user,expected', [
(AnonymousUser(), Q(privacy_level='everyone')),
(UserFactory.build(pk=1),
Q(privacy_level__in=['me', 'followers', 'instance', 'everyone'])),
Q(privacy_level__in=['followers', 'instance', 'everyone'])),
])
def test_privacy_level_query(user,expected):
query = fields.privacy_level_query(user)
......
......@@ -2,7 +2,6 @@ import pytest
from rest_framework.views import APIView
from django.contrib.auth.models import AnonymousUser
from django.http import Http404
from funkwhale_api.common import permissions
......@@ -19,24 +18,26 @@ def test_owner_permission_owner_field_ok(nodb_factories, api_request):
assert check is True
def test_owner_permission_owner_field_not_ok(nodb_factories, api_request):
def test_owner_permission_owner_field_not_ok(
anonymous_user, nodb_factories, api_request):
playlist = nodb_factories['playlists.Playlist']()
view = APIView.as_view()
permission = permissions.OwnerPermission()
request = api_request.get('/')
setattr(request, 'user', AnonymousUser())
setattr(request, 'user', anonymous_user)
with pytest.raises(Http404):
permission.has_object_permission(request, view, playlist)
def test_owner_permission_read_only(nodb_factories, api_request):
def test_owner_permission_read_only(
anonymous_user, nodb_factories, api_request):
playlist = nodb_factories['playlists.Playlist']()
view = APIView.as_view()
setattr(view, 'owner_checks', ['write'])
permission = permissions.OwnerPermission()
request = api_request.get('/')
setattr(request, 'user', AnonymousUser())
setattr(request, 'user', anonymous_user)
check = permission.has_object_permission(request, view, playlist)
assert check is True
import factory
import tempfile
import shutil
import pytest
import requests_mock
import shutil
import tempfile
from django.contrib.auth.models import AnonymousUser
from django.core.cache import cache as django_cache
from django.test import client
from dynamic_preferences.registries import global_preferences_registry
from rest_framework.test import APIClient
......@@ -31,7 +35,11 @@ def cache():
def factories(db):
from funkwhale_api import factories
for v in factories.registry.values():
v._meta.strategy = factory.CREATE_STRATEGY
try:
v._meta.strategy = factory.CREATE_STRATEGY
except AttributeError:
# probably not a class based factory
pass
yield factories.registry
......@@ -39,12 +47,16 @@ def factories(db):
def nodb_factories():
from funkwhale_api import factories
for v in factories.registry.values():
v._meta.strategy = factory.BUILD_STRATEGY
try:
v._meta.strategy = factory.BUILD_STRATEGY
except AttributeError:
# probably not a class based factory
pass
yield factories.registry
@pytest.fixture
def preferences(db):
def preferences(db, cache):
manager = global_preferences_registry.manager()
manager.all()
yield manager
......@@ -66,6 +78,11 @@ def logged_in_client(db, factories, client):
delattr(client, 'user')
@pytest.fixture
def anonymous_user():
return AnonymousUser()
@pytest.fixture
def api_client(client):
return APIClient()
......@@ -103,6 +120,11 @@ def api_request():
return APIRequestFactory()
@pytest.fixture
def fake_request():
return client.RequestFactory()
@pytest.fixture
def activity_registry():
r = record.registry
......@@ -126,3 +148,17 @@ def activity_registry():
@pytest.fixture
def activity_muted(activity_registry, mocker):
yield mocker.patch.object(record, 'send')
@pytest.fixture(autouse=True)
def media_root(settings):
tmp_dir = tempfile.mkdtemp()
settings.MEDIA_ROOT = tmp_dir
yield settings.MEDIA_ROOT
shutil.rmtree(tmp_dir)
@pytest.fixture
def r_mock():
with requests_mock.mock() as m:
yield m
import pytest
@pytest.fixture
def authenticated_actor(nodb_factories, mocker):
actor = nodb_factories['federation.Actor']()
mocker.patch(
'funkwhale_api.federation.authentication.SignatureAuthentication.authenticate_actor',
return_value=actor)
yield actor
from funkwhale_api.federation import activity
def test_deliver(nodb_factories, r_mock, mocker):
to = nodb_factories['federation.Actor']()
mocker.patch(
'funkwhale_api.federation.actors.get_actor',
return_value=to)
sender = nodb_factories['federation.Actor']()
ac = {
'id': 'http://test.federation/activity',
'type': 'Create',
'actor': sender.url,
'object': {
'id': 'http://test.federation/note',
'type': 'Note',
'content': 'Hello',
}
}
r_mock.post(to.inbox_url)
activity.deliver(
ac,
to=[to.url],
on_behalf_of=sender,
)
request = r_mock.request_history[0]
assert r_mock.called is True
assert r_mock.call_count == 1
assert request.url == to.inbox_url
assert request.headers['content-type'] == 'application/activity+json'
import pytest
from django.urls import reverse
from django.utils import timezone
from rest_framework import exceptions
from funkwhale_api.federation import actors
from funkwhale_api.federation import serializers
from funkwhale_api.federation import utils
def test_actor_fetching(r_mock):
payload = {
'id': 'https://actor.mock/users/actor#main-key',
'owner': 'test',
'publicKeyPem': 'test_pem',
}
actor_url = 'https://actor.mock/'
r_mock.get(actor_url, json=payload)
r = actors.get_actor_data(actor_url)
assert r == payload
def test_get_library(settings, preferences):
preferences['federation__public_key'] = 'public_key'
expected = {
'preferred_username': 'library',
'domain': settings.FEDERATION_HOSTNAME,
'type': 'Person',
'name': '{}\'s library'.format(settings.FEDERATION_HOSTNAME),
'manually_approves_followers': True,
'url': utils.full_url(
reverse(
'federation:instance-actors-detail',
kwargs={'actor': 'library'})),
'shared_inbox_url': utils.full_url(
reverse(
'federation:instance-actors-inbox',
kwargs={'actor': 'library'})),
'inbox_url': utils.full_url(
reverse(
'federation:instance-actors-inbox',
kwargs={'actor': 'library'})),
'outbox_url': utils.full_url(
reverse(
'federation:instance-actors-outbox',
kwargs={'actor': 'library'})),
'public_key': 'public_key',
'summary': 'Bot account to federate with {}\'s library'.format(
settings.FEDERATION_HOSTNAME),
}
actor = actors.SYSTEM_ACTORS['library'].get_actor_instance()
for key, value in expected.items():
assert getattr(actor, key) == value
def test_get_test(settings, preferences):
preferences['federation__public_key'] = 'public_key'
expected = {
'preferred_username': 'test',
'domain': settings.FEDERATION_HOSTNAME,
'type': 'Person',
'name': '{}\'s test account'.format(settings.FEDERATION_HOSTNAME),
'manually_approves_followers': False,
'url': utils.full_url(
reverse(
'federation:instance-actors-detail',
kwargs={'actor': 'test'})),
'shared_inbox_url': utils.full_url(
reverse(
'federation:instance-actors-inbox',
kwargs={'actor': 'test'})),
'inbox_url': utils.full_url(
reverse(
'federation:instance-actors-inbox',
kwargs={'actor': 'test'})),
'outbox_url': utils.full_url(
reverse(
'federation:instance-actors-outbox',
kwargs={'actor': 'test'})),
'public_key': 'public_key',
'summary': 'Bot account to test federation with {}. Send me /ping and I\'ll answer you.'.format(
settings.FEDERATION_HOSTNAME),
}
actor = actors.SYSTEM_ACTORS['test'].get_actor_instance()
for key, value in expected.items():
assert getattr(actor, key) == value
def test_test_get_outbox():
expected = {
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1",
{}
],
"id": utils.full_url(
reverse(
'federation:instance-actors-outbox',
kwargs={'actor': 'test'})),
"type": "OrderedCollection",
"totalItems": 0,
"orderedItems": []
}
data = actors.SYSTEM_ACTORS['test'].get_outbox({}, actor=None)
assert data == expected
def test_test_post_inbox_requires_authenticated_actor():
with pytest.raises(exceptions.PermissionDenied):
actors.SYSTEM_ACTORS['test'].post_inbox({}, actor=None)
def test_test_post_outbox_validates_actor(nodb_factories):
actor = nodb_factories['federation.Actor']()
data = {
'actor': 'noop'
}
with pytest.raises(exceptions.ValidationError) as exc_info:
actors.SYSTEM_ACTORS['test'].post_inbox(data, actor=actor)
msg = 'The actor making the request do not match'
assert msg in exc_info.value
def test_test_post_outbox_handles_create_note(
settings, mocker, factories):
deliver = mocker.patch(
'funkwhale_api.federation.activity.deliver')
actor = factories['federation.Actor']()
now = timezone.now()
mocker.patch('django.utils.timezone.now', return_value=now)
data = {
'actor': actor.url,
'type': 'Create',
'id': 'http://test.federation/activity',
'object': {
'type': 'Note',
'id': 'http://test.federation/object',
'content': '<p><a>@mention</a> /ping</p>'
}
}
test_actor = actors.SYSTEM_ACTORS['test'].get_actor_instance()
expected_note = factories['federation.Note'](
id='https://test.federation/activities/note/{}'.format(
now.timestamp()
),
content='Pong!',
published=now.isoformat(),
inReplyTo=data['object']['id'],
cc=[],
summary=None,
sensitive=False,
attributedTo=test_actor.url,
attachment=[],
to=[actor.url],
url='https://{}/activities/note/{}'.format(
settings.FEDERATION_HOSTNAME, now.timestamp()
),
tag=[{
'href': actor.url,
'name': actor.mention_username,
'type': 'Mention',
}]
)
expected_activity = {
'@context': [
'https://www.w3.org/ns/activitystreams',
'https://w3id.org/security/v1',
{}
],
'actor': test_actor.url,
'id': 'https://{}/activities/note/{}/activity'.format(
settings.FEDERATION_HOSTNAME, now.timestamp()
),
'to': actor.url,
'type': 'Create',
'published': now.isoformat(),
'object': expected_note,
'cc': [],
}
actors.SYSTEM_ACTORS['test'].post_inbox(data, actor=actor)
deliver.assert_called_once_with(
expected_activity,
to=[actor.url],
on_behalf_of=actors.SYSTEM_ACTORS['test'].get_actor_instance()
)
from funkwhale_api.federation import authentication
from funkwhale_api.federation import keys
from funkwhale_api.federation import signing
def test_authenticate(nodb_factories, mocker, api_request):
private, public = keys.get_key_pair()
actor_url = 'https://test.federation/actor'
mocker.patch(
'funkwhale_api.federation.actors.get_actor_data',
return_value={
'id': actor_url,
'outbox': 'https://test.com',
'inbox': 'https://test.com',
'publicKey': {
'publicKeyPem': public.decode('utf-8'),
'owner': actor_url,
'id': actor_url + '#main-key',
}
})
signed_request = nodb_factories['federation.SignedRequest'](
auth__key=private,
auth__key_id=actor_url + '#main-key',
auth__headers=[
'date',
]
)
prepared = signed_request.prepare()
django_request = api_request.get(
'/',
**{
'HTTP_DATE': prepared.headers['date'],
'HTTP_SIGNATURE': prepared.headers['signature'],
}
)
authenticator = authentication.SignatureAuthentication()
user, _ = authenticator.authenticate(django_request)
actor = django_request.actor
assert user.is_anonymous is True
assert actor.public_key == public.decode('utf-8')
assert actor.url == actor_url
from django.core.management import call_command
def test_generate_instance_key_pair(preferences, mocker):
mocker.patch(
'funkwhale_api.federation.keys.get_key_pair',
return_value=(b'private', b'public'))
assert preferences['federation__public_key'] == ''
assert preferences['federation__private_key'] == ''
call_command('generate_keys', interactive=False)
assert preferences['federation__private_key'] == 'private'
assert preferences['federation__public_key'] == 'public'
import pytest
from funkwhale_api.federation import keys
@pytest.mark.parametrize('raw, expected', [
('algorithm="test",keyId="https://test.com"', 'https://test.com'),
('keyId="https://test.com",algorithm="test"', 'https://test.com'),
])
def test_get_key_from_header(raw, expected):
r = keys.get_key_id_from_signature_header(raw)
assert r == expected
@pytest.mark.parametrize('raw', [
'algorithm="test",keyid="badCase"',
'algorithm="test",wrong="wrong"',
'keyId = "wrong"',
'keyId=\'wrong\'',
'keyId="notanurl"',
'keyId="wrong://test.com"',
])
def test_get_key_from_header_invalid(raw):
with pytest.raises(ValueError):
keys.get_key_id_from_signature_header(raw)
from django.urls import reverse
from funkwhale_api.federation import keys
from funkwhale_api.federation import models
from funkwhale_api.federation import serializers
def test_actor_serializer_from_ap(db):
payload = {
'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',
'url': 'https://test.federation/@user',
'manuallyApprovesFollowers': False,
'publicKey': {
'id': 'https://test.federation/user#main-key',
'owner': 'https://test.federation/user',
'publicKeyPem': 'yolo'
},
'endpoints': {
'sharedInbox': 'https://test.federation/inbox'
},
}
serializer = serializers.ActorSerializer(data=payload)
assert serializer.is_valid()
actor = serializer.build()
assert actor.url == payload['id']
assert actor.inbox_url == payload['inbox']
assert actor.outbox_url == payload['outbox']
assert actor.shared_inbox_url == payload['endpoints']['sharedInbox']
assert actor.followers_url == payload['followers']
assert actor.following_url == payload['following']
assert actor.public_key == payload['publicKey']['publicKeyPem']
assert actor.preferred_username == payload['preferredUsername']
assert actor.name == payload['name']
assert actor.domain == 'test.federation'
assert actor.summary == payload['summary']
assert actor.type == 'Person'
assert actor.manually_approves_followers == payload['manuallyApprovesFollowers']
def test_actor_serializer_only_mandatory_field_from_ap(db):
payload = {
'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()
actor = serializer.build()
assert actor.url == 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 == 'test.federation'
assert actor.type == 'Person'
assert actor.manually_approves_followers is None
def test_actor_serializer_to_ap():
expected = {
'@context': [
'https://www.w3.org/ns/activitystreams',
'https://w3id.org/security/v1',
{},
],
'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(
url=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='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(
url=expected['links'][0]['href'],
preferred_username='service',
domain='test.federation',