forked from mirrors/bookwyrm
215 lines
7.8 KiB
Python
215 lines
7.8 KiB
Python
''' tests the base functionality for activitypub dataclasses '''
|
|
from io import BytesIO
|
|
import json
|
|
import pathlib
|
|
from unittest.mock import patch
|
|
|
|
from dataclasses import dataclass
|
|
from django.test import TestCase
|
|
from PIL import Image
|
|
import responses
|
|
|
|
from bookwyrm import activitypub
|
|
from bookwyrm.activitypub.base_activity import ActivityObject, \
|
|
resolve_remote_id, set_related_field
|
|
from bookwyrm.activitypub import ActivitySerializerError
|
|
from bookwyrm import models
|
|
|
|
class BaseActivity(TestCase):
|
|
''' the super class for model-linked activitypub dataclasses '''
|
|
def setUp(self):
|
|
''' we're probably going to re-use this so why copy/paste '''
|
|
self.user = models.User.objects.create_user(
|
|
'mouse', 'mouse@mouse.mouse', 'mouseword', local=True)
|
|
self.user.remote_id = 'http://example.com/a/b'
|
|
self.user.save()
|
|
|
|
datafile = pathlib.Path(__file__).parent.joinpath(
|
|
'../data/ap_user.json'
|
|
)
|
|
self.userdata = json.loads(datafile.read_bytes())
|
|
# don't try to load the user icon
|
|
del self.userdata['icon']
|
|
|
|
image_file = pathlib.Path(__file__).parent.joinpath(
|
|
'../../static/images/default_avi.jpg')
|
|
image = Image.open(image_file)
|
|
output = BytesIO()
|
|
image.save(output, format=image.format)
|
|
self.image_data = output.getvalue()
|
|
|
|
def test_init(self):
|
|
''' simple successfuly init '''
|
|
instance = ActivityObject(id='a', type='b')
|
|
self.assertTrue(hasattr(instance, 'id'))
|
|
self.assertTrue(hasattr(instance, 'type'))
|
|
|
|
def test_init_missing(self):
|
|
''' init with missing required params '''
|
|
with self.assertRaises(ActivitySerializerError):
|
|
ActivityObject()
|
|
|
|
def test_init_extra_fields(self):
|
|
''' init ignoring additional fields '''
|
|
instance = ActivityObject(id='a', type='b', fish='c')
|
|
self.assertTrue(hasattr(instance, 'id'))
|
|
self.assertTrue(hasattr(instance, 'type'))
|
|
|
|
def test_init_default_field(self):
|
|
''' replace an existing required field with a default field '''
|
|
@dataclass(init=False)
|
|
class TestClass(ActivityObject):
|
|
''' test class with default field '''
|
|
type: str = 'TestObject'
|
|
|
|
instance = TestClass(id='a')
|
|
self.assertEqual(instance.id, 'a')
|
|
self.assertEqual(instance.type, 'TestObject')
|
|
|
|
def test_serialize(self):
|
|
''' simple function for converting dataclass to dict '''
|
|
instance = ActivityObject(id='a', type='b')
|
|
serialized = instance.serialize()
|
|
self.assertIsInstance(serialized, dict)
|
|
self.assertEqual(serialized['id'], 'a')
|
|
self.assertEqual(serialized['type'], 'b')
|
|
|
|
@responses.activate
|
|
def test_resolve_remote_id(self):
|
|
''' look up or load remote data '''
|
|
# existing item
|
|
result = resolve_remote_id(models.User, 'http://example.com/a/b')
|
|
self.assertEqual(result, self.user)
|
|
|
|
# remote item
|
|
responses.add(
|
|
responses.GET,
|
|
'https://example.com/user/mouse',
|
|
json=self.userdata,
|
|
status=200)
|
|
|
|
with patch('bookwyrm.models.user.set_remote_server.delay'):
|
|
result = resolve_remote_id(
|
|
models.User, 'https://example.com/user/mouse')
|
|
self.assertIsInstance(result, models.User)
|
|
self.assertEqual(result.remote_id, 'https://example.com/user/mouse')
|
|
self.assertEqual(result.name, 'MOUSE?? MOUSE!!')
|
|
|
|
def test_to_model(self):
|
|
''' the big boy of this module. it feels janky to test this with actual
|
|
models rather than a test model, but I don't know how to make a test
|
|
model so here we are. '''
|
|
instance = ActivityObject(id='a', type='b')
|
|
with self.assertRaises(ActivitySerializerError):
|
|
instance.to_model(models.User)
|
|
|
|
# test setting simple fields
|
|
self.assertEqual(self.user.name, '')
|
|
update_data = activitypub.Person(**self.user.to_activity())
|
|
update_data.name = 'New Name'
|
|
update_data.to_model(models.User, self.user)
|
|
|
|
self.assertEqual(self.user.name, 'New Name')
|
|
|
|
def test_to_model_foreign_key(self):
|
|
''' test setting one to one/foreign key '''
|
|
update_data = activitypub.Person(**self.user.to_activity())
|
|
update_data.publicKey['publicKeyPem'] = 'hi im secure'
|
|
update_data.to_model(models.User, self.user)
|
|
self.assertEqual(self.user.key_pair.public_key, 'hi im secure')
|
|
|
|
@responses.activate
|
|
def test_to_model_image(self):
|
|
''' update an image field '''
|
|
update_data = activitypub.Person(**self.user.to_activity())
|
|
update_data.icon = {'url': 'http://www.example.com/image.jpg'}
|
|
responses.add(
|
|
responses.GET,
|
|
'http://www.example.com/image.jpg',
|
|
body=self.image_data,
|
|
status=200)
|
|
|
|
self.assertIsNone(self.user.avatar.name)
|
|
with self.assertRaises(ValueError):
|
|
self.user.avatar.file #pylint: disable=pointless-statement
|
|
|
|
update_data.to_model(models.User, self.user)
|
|
self.assertIsNotNone(self.user.avatar.name)
|
|
self.assertIsNotNone(self.user.avatar.file)
|
|
|
|
def test_to_model_many_to_many(self):
|
|
''' annoying that these all need special handling '''
|
|
status = models.Status.objects.create(
|
|
content='test status',
|
|
user=self.user,
|
|
)
|
|
book = models.Edition.objects.create(
|
|
title='Test Edition', remote_id='http://book.com/book')
|
|
update_data = activitypub.Note(**status.to_activity())
|
|
update_data.tag = [
|
|
{
|
|
'type': 'Mention',
|
|
'name': 'gerald',
|
|
'href': 'http://example.com/a/b'
|
|
},
|
|
{
|
|
'type': 'Edition',
|
|
'name': 'gerald j. books',
|
|
'href': 'http://book.com/book'
|
|
},
|
|
]
|
|
update_data.to_model(models.Status, instance=status)
|
|
self.assertEqual(status.mention_users.first(), self.user)
|
|
self.assertEqual(status.mention_books.first(), book)
|
|
|
|
|
|
@responses.activate
|
|
def test_to_model_one_to_many(self):
|
|
''' these are reversed relationships, where the secondary object
|
|
keys the primary object but not vice versa '''
|
|
status = models.Status.objects.create(
|
|
content='test status',
|
|
user=self.user,
|
|
)
|
|
update_data = activitypub.Note(**status.to_activity())
|
|
update_data.attachment = [{
|
|
'url': 'http://www.example.com/image.jpg',
|
|
'name': 'alt text',
|
|
'type': 'Image',
|
|
}]
|
|
|
|
responses.add(
|
|
responses.GET,
|
|
'http://www.example.com/image.jpg',
|
|
body=self.image_data,
|
|
status=200)
|
|
|
|
# sets the celery task call to the function call
|
|
with patch(
|
|
'bookwyrm.activitypub.base_activity.set_related_field.delay'):
|
|
update_data.to_model(models.Status, instance=status)
|
|
self.assertIsNone(status.attachments.first())
|
|
|
|
|
|
@responses.activate
|
|
def test_set_related_field(self):
|
|
''' celery task to add back-references to created objects '''
|
|
status = models.Status.objects.create(
|
|
content='test status',
|
|
user=self.user,
|
|
)
|
|
data = {
|
|
'url': 'http://www.example.com/image.jpg',
|
|
'name': 'alt text',
|
|
'type': 'Image',
|
|
}
|
|
responses.add(
|
|
responses.GET,
|
|
'http://www.example.com/image.jpg',
|
|
body=self.image_data,
|
|
status=200)
|
|
set_related_field(
|
|
'Image', 'Status', 'status', status.remote_id, data)
|
|
|
|
self.assertIsInstance(status.attachments.first(), models.Image)
|
|
self.assertIsNotNone(status.attachments.first().image)
|