forked from mirrors/bookwyrm
384 lines
16 KiB
Python
384 lines
16 KiB
Python
''' testing model activitypub utilities '''
|
|
from unittest.mock import patch
|
|
from collections import namedtuple
|
|
from dataclasses import dataclass
|
|
import re
|
|
from django import db
|
|
from django.test import TestCase
|
|
|
|
from bookwyrm.activitypub.base_activity import ActivityObject
|
|
from bookwyrm import models
|
|
from bookwyrm.models import base_model
|
|
from bookwyrm.models.activitypub_mixin import ActivitypubMixin
|
|
from bookwyrm.models.activitypub_mixin import ActivityMixin, ObjectMixin
|
|
|
|
class ActivitypubMixins(TestCase):
|
|
''' functionality shared across models '''
|
|
def setUp(self):
|
|
''' shared data '''
|
|
self.local_user = models.User.objects.create_user(
|
|
'mouse', 'mouse@mouse.com', 'mouseword',
|
|
local=True, localname='mouse')
|
|
self.local_user.remote_id = 'http://example.com/a/b'
|
|
self.local_user.save(broadcast=False)
|
|
with patch('bookwyrm.models.user.set_remote_server.delay'):
|
|
self.remote_user = models.User.objects.create_user(
|
|
'rat', 'rat@rat.com', 'ratword',
|
|
local=False,
|
|
remote_id='https://example.com/users/rat',
|
|
inbox='https://example.com/users/rat/inbox',
|
|
outbox='https://example.com/users/rat/outbox',
|
|
)
|
|
|
|
self.object_mock = {
|
|
'to': 'to field', 'cc': 'cc field',
|
|
'content': 'hi', 'id': 'bip', 'type': 'Test',
|
|
'published': '2020-12-04T17:52:22.623807+00:00',
|
|
}
|
|
|
|
|
|
# ActivitypubMixin
|
|
def test_to_activity(self):
|
|
''' model to ActivityPub json '''
|
|
@dataclass(init=False)
|
|
class TestActivity(ActivityObject):
|
|
''' real simple mock '''
|
|
type: str = 'Test'
|
|
|
|
class TestModel(ActivitypubMixin, base_model.BookWyrmModel):
|
|
''' real simple mock model because BookWyrmModel is abstract '''
|
|
|
|
instance = TestModel()
|
|
instance.remote_id = 'https://www.example.com/test'
|
|
instance.activity_serializer = TestActivity
|
|
|
|
activity = instance.to_activity()
|
|
self.assertIsInstance(activity, dict)
|
|
self.assertEqual(activity['id'], 'https://www.example.com/test')
|
|
self.assertEqual(activity['type'], 'Test')
|
|
|
|
|
|
def test_find_existing_by_remote_id(self):
|
|
''' attempt to match a remote id to an object in the db '''
|
|
# uses a different remote id scheme
|
|
# this isn't really part of this test directly but it's helpful to state
|
|
book = models.Edition.objects.create(
|
|
title='Test Edition', remote_id='http://book.com/book')
|
|
|
|
self.assertEqual(book.origin_id, 'http://book.com/book')
|
|
self.assertNotEqual(book.remote_id, 'http://book.com/book')
|
|
|
|
# uses subclasses
|
|
with patch('bookwyrm.models.activitypub_mixin.broadcast_task.delay'):
|
|
models.Comment.objects.create(
|
|
user=self.local_user, content='test status', book=book, \
|
|
remote_id='https://comment.net')
|
|
|
|
result = models.User.find_existing_by_remote_id('hi')
|
|
self.assertIsNone(result)
|
|
|
|
result = models.User.find_existing_by_remote_id(
|
|
'http://example.com/a/b')
|
|
self.assertEqual(result, self.local_user)
|
|
|
|
# test using origin id
|
|
result = models.Edition.find_existing_by_remote_id(
|
|
'http://book.com/book')
|
|
self.assertEqual(result, book)
|
|
|
|
# test subclass match
|
|
result = models.Status.find_existing_by_remote_id(
|
|
'https://comment.net')
|
|
|
|
|
|
def test_find_existing(self):
|
|
''' match a blob of data to a model '''
|
|
book = models.Edition.objects.create(
|
|
title='Test edition',
|
|
openlibrary_key='OL1234',
|
|
)
|
|
|
|
result = models.Edition.find_existing(
|
|
{'openlibraryKey': 'OL1234'})
|
|
self.assertEqual(result, book)
|
|
|
|
|
|
def test_get_recipients_public_object(self):
|
|
''' determines the recipients for an object's broadcast '''
|
|
MockSelf = namedtuple('Self', ('privacy'))
|
|
mock_self = MockSelf('public')
|
|
recipients = ActivitypubMixin.get_recipients(mock_self)
|
|
self.assertEqual(len(recipients), 1)
|
|
self.assertEqual(recipients[0], self.remote_user.inbox)
|
|
|
|
|
|
def test_get_recipients_public_user_object_no_followers(self):
|
|
''' determines the recipients for a user's object broadcast '''
|
|
MockSelf = namedtuple('Self', ('privacy', 'user'))
|
|
mock_self = MockSelf('public', self.local_user)
|
|
|
|
recipients = ActivitypubMixin.get_recipients(mock_self)
|
|
self.assertEqual(len(recipients), 0)
|
|
|
|
|
|
def test_get_recipients_public_user_object(self):
|
|
''' determines the recipients for a user's object broadcast '''
|
|
MockSelf = namedtuple('Self', ('privacy', 'user'))
|
|
mock_self = MockSelf('public', self.local_user)
|
|
self.local_user.followers.add(self.remote_user)
|
|
|
|
recipients = ActivitypubMixin.get_recipients(mock_self)
|
|
self.assertEqual(len(recipients), 1)
|
|
self.assertEqual(recipients[0], self.remote_user.inbox)
|
|
|
|
|
|
def test_get_recipients_public_user_object_with_mention(self):
|
|
''' determines the recipients for a user's object broadcast '''
|
|
MockSelf = namedtuple('Self', ('privacy', 'user'))
|
|
mock_self = MockSelf('public', self.local_user)
|
|
self.local_user.followers.add(self.remote_user)
|
|
with patch('bookwyrm.models.user.set_remote_server.delay'):
|
|
another_remote_user = models.User.objects.create_user(
|
|
'nutria', 'nutria@nutria.com', 'nutriaword',
|
|
local=False,
|
|
remote_id='https://example.com/users/nutria',
|
|
inbox='https://example.com/users/nutria/inbox',
|
|
outbox='https://example.com/users/nutria/outbox',
|
|
)
|
|
MockSelf = namedtuple('Self', ('privacy', 'user', 'recipients'))
|
|
mock_self = MockSelf('public', self.local_user, [another_remote_user])
|
|
|
|
recipients = ActivitypubMixin.get_recipients(mock_self)
|
|
self.assertEqual(len(recipients), 2)
|
|
self.assertEqual(recipients[0], another_remote_user.inbox)
|
|
self.assertEqual(recipients[1], self.remote_user.inbox)
|
|
|
|
|
|
def test_get_recipients_direct(self):
|
|
''' determines the recipients for a user's object broadcast '''
|
|
MockSelf = namedtuple('Self', ('privacy', 'user'))
|
|
mock_self = MockSelf('public', self.local_user)
|
|
self.local_user.followers.add(self.remote_user)
|
|
with patch('bookwyrm.models.user.set_remote_server.delay'):
|
|
another_remote_user = models.User.objects.create_user(
|
|
'nutria', 'nutria@nutria.com', 'nutriaword',
|
|
local=False,
|
|
remote_id='https://example.com/users/nutria',
|
|
inbox='https://example.com/users/nutria/inbox',
|
|
outbox='https://example.com/users/nutria/outbox',
|
|
)
|
|
MockSelf = namedtuple('Self', ('privacy', 'user', 'recipients'))
|
|
mock_self = MockSelf('direct', self.local_user, [another_remote_user])
|
|
|
|
recipients = ActivitypubMixin.get_recipients(mock_self)
|
|
self.assertEqual(len(recipients), 1)
|
|
self.assertEqual(recipients[0], another_remote_user.inbox)
|
|
|
|
|
|
def test_get_recipients_combine_inboxes(self):
|
|
''' should combine users with the same shared_inbox '''
|
|
self.remote_user.shared_inbox = 'http://example.com/inbox'
|
|
self.remote_user.save(broadcast=False)
|
|
with patch('bookwyrm.models.user.set_remote_server.delay'):
|
|
another_remote_user = models.User.objects.create_user(
|
|
'nutria', 'nutria@nutria.com', 'nutriaword',
|
|
local=False,
|
|
remote_id='https://example.com/users/nutria',
|
|
inbox='https://example.com/users/nutria/inbox',
|
|
shared_inbox='http://example.com/inbox',
|
|
outbox='https://example.com/users/nutria/outbox',
|
|
)
|
|
MockSelf = namedtuple('Self', ('privacy', 'user'))
|
|
mock_self = MockSelf('public', self.local_user)
|
|
self.local_user.followers.add(self.remote_user)
|
|
self.local_user.followers.add(another_remote_user)
|
|
|
|
recipients = ActivitypubMixin.get_recipients(mock_self)
|
|
self.assertEqual(len(recipients), 1)
|
|
self.assertEqual(recipients[0], 'http://example.com/inbox')
|
|
|
|
|
|
def test_get_recipients_software(self):
|
|
''' should differentiate between bookwyrm and other remote users '''
|
|
with patch('bookwyrm.models.user.set_remote_server.delay'):
|
|
another_remote_user = models.User.objects.create_user(
|
|
'nutria', 'nutria@nutria.com', 'nutriaword',
|
|
local=False,
|
|
remote_id='https://example.com/users/nutria',
|
|
inbox='https://example.com/users/nutria/inbox',
|
|
outbox='https://example.com/users/nutria/outbox',
|
|
bookwyrm_user=False,
|
|
)
|
|
MockSelf = namedtuple('Self', ('privacy', 'user'))
|
|
mock_self = MockSelf('public', self.local_user)
|
|
self.local_user.followers.add(self.remote_user)
|
|
self.local_user.followers.add(another_remote_user)
|
|
|
|
recipients = ActivitypubMixin.get_recipients(mock_self)
|
|
self.assertEqual(len(recipients), 2)
|
|
|
|
recipients = ActivitypubMixin.get_recipients(
|
|
mock_self, software='bookwyrm')
|
|
self.assertEqual(len(recipients), 1)
|
|
self.assertEqual(recipients[0], self.remote_user.inbox)
|
|
|
|
recipients = ActivitypubMixin.get_recipients(
|
|
mock_self, software='other')
|
|
self.assertEqual(len(recipients), 1)
|
|
self.assertEqual(recipients[0], another_remote_user.inbox)
|
|
|
|
|
|
# ObjectMixin
|
|
def test_object_save_create(self):
|
|
''' should save uneventufully when broadcast is disabled '''
|
|
class Success(Exception):
|
|
''' this means we got to the right method '''
|
|
|
|
class ObjectModel(ObjectMixin, base_model.BookWyrmModel):
|
|
''' real simple mock model because BookWyrmModel is abstract '''
|
|
user = models.fields.ForeignKey('User', on_delete=db.models.CASCADE)
|
|
def save(self, *args, **kwargs):
|
|
with patch('django.db.models.Model.save'):
|
|
super().save(*args, **kwargs)
|
|
def broadcast(self, activity, sender, **kwargs):#pylint: disable=arguments-differ
|
|
''' do something '''
|
|
raise Success()
|
|
def to_create_activity(self, user):#pylint: disable=arguments-differ
|
|
return {}
|
|
|
|
with self.assertRaises(Success):
|
|
ObjectModel(user=self.local_user).save()
|
|
|
|
ObjectModel(user=self.remote_user).save()
|
|
ObjectModel(user=self.local_user).save(broadcast=False)
|
|
ObjectModel(user=None).save()
|
|
|
|
|
|
def test_object_save_update(self):
|
|
''' should save uneventufully when broadcast is disabled '''
|
|
class Success(Exception):
|
|
''' this means we got to the right method '''
|
|
|
|
class UpdateObjectModel(ObjectMixin, base_model.BookWyrmModel):
|
|
''' real simple mock model because BookWyrmModel is abstract '''
|
|
user = models.fields.ForeignKey('User', on_delete=db.models.CASCADE)
|
|
last_edited_by = models.fields.ForeignKey(
|
|
'User', on_delete=db.models.CASCADE)
|
|
def save(self, *args, **kwargs):
|
|
with patch('django.db.models.Model.save'):
|
|
super().save(*args, **kwargs)
|
|
def to_update_activity(self, user):
|
|
raise Success()
|
|
|
|
with self.assertRaises(Success):
|
|
UpdateObjectModel(id=1, user=self.local_user).save()
|
|
with self.assertRaises(Success):
|
|
UpdateObjectModel(id=1, last_edited_by=self.local_user).save()
|
|
|
|
|
|
def test_object_save_delete(self):
|
|
''' should create delete activities when objects are deleted by flag '''
|
|
class ActivitySuccess(Exception):
|
|
''' this means we got to the right method '''
|
|
|
|
class DeletableObjectModel(ObjectMixin, base_model.BookWyrmModel):
|
|
''' real simple mock model because BookWyrmModel is abstract '''
|
|
user = models.fields.ForeignKey('User', on_delete=db.models.CASCADE)
|
|
deleted = models.fields.BooleanField()
|
|
def save(self, *args, **kwargs):
|
|
with patch('django.db.models.Model.save'):
|
|
super().save(*args, **kwargs)
|
|
def to_delete_activity(self, user):
|
|
raise ActivitySuccess()
|
|
|
|
with self.assertRaises(ActivitySuccess):
|
|
DeletableObjectModel(
|
|
id=1, user=self.local_user, deleted=True).save()
|
|
|
|
|
|
def test_to_create_activity(self):
|
|
''' wrapper for ActivityPub "create" action '''
|
|
MockSelf = namedtuple('Self', ('remote_id', 'to_activity'))
|
|
mock_self = MockSelf(
|
|
'https://example.com/status/1',
|
|
lambda *args: self.object_mock
|
|
)
|
|
activity = ObjectMixin.to_create_activity(
|
|
mock_self, self.local_user)
|
|
self.assertEqual(
|
|
activity['id'],
|
|
'https://example.com/status/1/activity'
|
|
)
|
|
self.assertEqual(activity['actor'], self.local_user.remote_id)
|
|
self.assertEqual(activity['type'], 'Create')
|
|
self.assertEqual(activity['to'], 'to field')
|
|
self.assertEqual(activity['cc'], 'cc field')
|
|
self.assertIsInstance(activity['object'], dict)
|
|
self.assertEqual(
|
|
activity['signature'].creator,
|
|
'%s#main-key' % self.local_user.remote_id
|
|
)
|
|
|
|
def test_to_delete_activity(self):
|
|
''' wrapper for Delete activity '''
|
|
MockSelf = namedtuple('Self', ('remote_id', 'to_activity'))
|
|
mock_self = MockSelf(
|
|
'https://example.com/status/1',
|
|
lambda *args: self.object_mock
|
|
)
|
|
activity = ObjectMixin.to_delete_activity(
|
|
mock_self, self.local_user)
|
|
self.assertEqual(
|
|
activity['id'],
|
|
'https://example.com/status/1/activity'
|
|
)
|
|
self.assertEqual(activity['actor'], self.local_user.remote_id)
|
|
self.assertEqual(activity['type'], 'Delete')
|
|
self.assertEqual(
|
|
activity['to'],
|
|
['%s/followers' % self.local_user.remote_id])
|
|
self.assertEqual(
|
|
activity['cc'],
|
|
['https://www.w3.org/ns/activitystreams#Public'])
|
|
|
|
|
|
def test_to_update_activity(self):
|
|
''' ditto above but for Update '''
|
|
MockSelf = namedtuple('Self', ('remote_id', 'to_activity'))
|
|
mock_self = MockSelf(
|
|
'https://example.com/status/1',
|
|
lambda *args: self.object_mock
|
|
)
|
|
activity = ObjectMixin.to_update_activity(
|
|
mock_self, self.local_user)
|
|
self.assertIsNotNone(
|
|
re.match(
|
|
r'^https:\/\/example\.com\/status\/1#update\/.*',
|
|
activity['id']
|
|
)
|
|
)
|
|
self.assertEqual(activity['actor'], self.local_user.remote_id)
|
|
self.assertEqual(activity['type'], 'Update')
|
|
self.assertEqual(
|
|
activity['to'],
|
|
['https://www.w3.org/ns/activitystreams#Public'])
|
|
self.assertIsInstance(activity['object'], dict)
|
|
|
|
|
|
# Activity mixin
|
|
def test_to_undo_activity(self):
|
|
''' and again, for Undo '''
|
|
MockSelf = namedtuple('Self', ('remote_id', 'to_activity', 'user'))
|
|
mock_self = MockSelf(
|
|
'https://example.com/status/1',
|
|
lambda *args: self.object_mock,
|
|
self.local_user,
|
|
)
|
|
activity = ActivityMixin.to_undo_activity(mock_self)
|
|
self.assertEqual(
|
|
activity['id'],
|
|
'https://example.com/status/1#undo'
|
|
)
|
|
self.assertEqual(activity['actor'], self.local_user.remote_id)
|
|
self.assertEqual(activity['type'], 'Undo')
|
|
self.assertIsInstance(activity['object'], dict)
|