moviewyrm/bookwyrm/tests/models/test_activitypub_mixin.py

424 lines
18 KiB
Python
Raw Normal View History

2021-03-08 16:49:10 +00:00
""" 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,
ActivityMixin,
ObjectMixin,
OrderedCollectionMixin,
to_ordered_collection_page,
)
from bookwyrm.settings import PAGE_LENGTH
2021-03-08 16:49:10 +00:00
# pylint: disable=invalid-name
2021-09-06 20:53:49 +00:00
@patch("bookwyrm.activitystreams.add_status_task.delay")
2021-11-12 17:17:00 +00:00
@patch("bookwyrm.models.activitypub_mixin.broadcast_task.apply_async")
class ActivitypubMixins(TestCase):
2021-04-26 16:15:42 +00:00
"""functionality shared across models"""
2021-03-08 16:49:10 +00:00
def setUp(self):
2021-04-26 16:15:42 +00:00
"""shared data"""
with patch("bookwyrm.suggested_users.rerank_suggestions_task.delay"), patch(
"bookwyrm.activitystreams.populate_stream_task.delay"
):
2021-08-03 17:25:53 +00:00
self.local_user = models.User.objects.create_user(
"mouse", "mouse@mouse.com", "mouseword", local=True, localname="mouse"
)
2021-08-02 23:05:40 +00:00
self.local_user.remote_id = "http://example.com/a/b"
2021-08-03 23:21:29 +00:00
self.local_user.save(broadcast=False, update_fields=["remote_id"])
2021-08-02 23:05:40 +00:00
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",
2021-05-26 21:57:29 +00:00
)
2021-02-17 16:35:17 +00:00
self.object_mock = {
2021-03-08 16:49:10 +00:00
"to": "to field",
"cc": "cc field",
"content": "hi",
"id": "bip",
"type": "Test",
"published": "2020-12-04T17:52:22.623807+00:00",
2021-02-17 16:35:17 +00:00
}
def test_to_activity(self, *_):
2021-04-26 16:15:42 +00:00
"""model to ActivityPub json"""
2021-03-08 16:49:10 +00:00
@dataclass(init=False)
class TestActivity(ActivityObject):
2021-04-26 16:15:42 +00:00
"""real simple mock"""
2021-03-08 16:49:10 +00:00
type: str = "Test"
class TestModel(ActivitypubMixin, base_model.BookWyrmModel):
2021-04-26 16:15:42 +00:00
"""real simple mock model because BookWyrmModel is abstract"""
instance = TestModel()
2021-03-08 16:49:10 +00:00
instance.remote_id = "https://www.example.com/test"
instance.activity_serializer = TestActivity
activity = instance.to_activity()
self.assertIsInstance(activity, dict)
2021-03-08 16:49:10 +00:00
self.assertEqual(activity["id"], "https://www.example.com/test")
self.assertEqual(activity["type"], "Test")
def test_find_existing_by_remote_id(self, *_):
2021-04-26 16:15:42 +00:00
"""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
2021-08-02 23:05:40 +00:00
book = models.Edition.objects.create(
title="Test Edition", remote_id="http://book.com/book"
)
2021-03-08 16:49:10 +00:00
self.assertEqual(book.origin_id, "http://book.com/book")
self.assertNotEqual(book.remote_id, "http://book.com/book")
# uses subclasses
2021-08-02 23:05:40 +00:00
models.Comment.objects.create(
user=self.local_user,
content="test status",
book=book,
remote_id="https://comment.net",
)
2021-03-08 16:49:10 +00:00
result = models.User.find_existing_by_remote_id("hi")
self.assertIsNone(result)
2021-03-08 16:49:10 +00:00
result = models.User.find_existing_by_remote_id("http://example.com/a/b")
self.assertEqual(result, self.local_user)
# test using origin id
2021-03-08 16:49:10 +00:00
result = models.Edition.find_existing_by_remote_id("http://book.com/book")
self.assertEqual(result, book)
# test subclass match
2021-03-08 16:49:10 +00:00
result = models.Status.find_existing_by_remote_id("https://comment.net")
def test_find_existing(self, *_):
2021-04-26 16:15:42 +00:00
"""match a blob of data to a model"""
2021-08-02 23:05:40 +00:00
book = models.Edition.objects.create(
title="Test edition",
openlibrary_key="OL1234",
)
2021-03-08 16:49:10 +00:00
result = models.Edition.find_existing({"openlibraryKey": "OL1234"})
self.assertEqual(result, book)
def test_get_recipients_public_object(self, *_):
2021-04-26 16:15:42 +00:00
"""determines the recipients for an object's broadcast"""
2021-03-08 16:49:10 +00:00
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, *_):
2021-04-26 16:15:42 +00:00
"""determines the recipients for a user's object broadcast"""
2021-03-08 16:49:10 +00:00
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, *_):
2021-04-26 16:15:42 +00:00
"""determines the recipients for a user's object broadcast"""
2021-03-08 16:49:10 +00:00
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, *_):
2021-04-26 16:15:42 +00:00
"""determines the recipients for a user's object broadcast"""
2021-03-08 16:49:10 +00:00
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",
)
2021-03-08 16:49:10 +00:00
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)
2021-04-22 16:37:24 +00:00
self.assertTrue(another_remote_user.inbox in recipients)
self.assertTrue(self.remote_user.inbox in recipients)
def test_get_recipients_direct(self, *_):
2021-04-26 16:15:42 +00:00
"""determines the recipients for a user's object broadcast"""
2021-03-08 16:49:10 +00:00
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",
)
2021-03-08 16:49:10 +00:00
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, *_):
2021-04-26 16:15:42 +00:00
"""should combine users with the same shared_inbox"""
2021-03-08 16:49:10 +00:00
self.remote_user.shared_inbox = "http://example.com/inbox"
2021-08-03 23:21:29 +00:00
self.remote_user.save(broadcast=False, update_fields=["shared_inbox"])
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",
)
2021-03-08 16:49:10 +00:00
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)
2021-03-08 16:49:10 +00:00
self.assertEqual(recipients[0], "http://example.com/inbox")
def test_get_recipients_software(self, *_):
2021-04-26 16:15:42 +00:00
"""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,
)
2021-03-08 16:49:10 +00:00
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)
2021-03-08 16:49:10 +00:00
recipients = ActivitypubMixin.get_recipients(mock_self, software="bookwyrm")
self.assertEqual(len(recipients), 1)
self.assertEqual(recipients[0], self.remote_user.inbox)
2021-03-08 16:49:10 +00:00
recipients = ActivitypubMixin.get_recipients(mock_self, software="other")
self.assertEqual(len(recipients), 1)
self.assertEqual(recipients[0], another_remote_user.inbox)
2021-02-06 20:06:45 +00:00
# ObjectMixin
def test_object_save_create(self, *_):
2021-04-26 16:15:42 +00:00
"""should save uneventufully when broadcast is disabled"""
2021-03-08 16:49:10 +00:00
class Success(Exception):
2021-04-26 16:15:42 +00:00
"""this means we got to the right method"""
class ObjectModel(ObjectMixin, base_model.BookWyrmModel):
2021-04-26 16:15:42 +00:00
"""real simple mock model because BookWyrmModel is abstract"""
2021-03-08 16:49:10 +00:00
user = models.fields.ForeignKey("User", on_delete=db.models.CASCADE)
def save(self, *args, **kwargs):
2021-03-08 16:49:10 +00:00
with patch("django.db.models.Model.save"):
super().save(*args, **kwargs)
2021-03-08 16:49:10 +00:00
def broadcast(
self, activity, sender, **kwargs
): # pylint: disable=arguments-differ
2021-04-26 16:15:42 +00:00
"""do something"""
raise Success()
2021-03-08 16:49:10 +00:00
def to_create_activity(self, user): # pylint: disable=arguments-differ
return {}
with self.assertRaises(Success):
ObjectModel(user=self.local_user).save()
2021-02-06 21:48:02 +00:00
ObjectModel(user=self.remote_user).save()
ObjectModel(user=self.local_user).save(broadcast=False)
ObjectModel(user=None).save()
def test_object_save_update(self, *_):
2021-04-26 16:15:42 +00:00
"""should save uneventufully when broadcast is disabled"""
2021-03-08 16:49:10 +00:00
2021-02-06 21:48:02 +00:00
class Success(Exception):
2021-04-26 16:15:42 +00:00
"""this means we got to the right method"""
2021-02-06 21:48:02 +00:00
class UpdateObjectModel(ObjectMixin, base_model.BookWyrmModel):
2021-04-26 16:15:42 +00:00
"""real simple mock model because BookWyrmModel is abstract"""
2021-03-08 16:49:10 +00:00
user = models.fields.ForeignKey("User", on_delete=db.models.CASCADE)
2021-02-06 21:48:02 +00:00
last_edited_by = models.fields.ForeignKey(
2021-03-08 16:49:10 +00:00
"User", on_delete=db.models.CASCADE
)
2021-02-06 21:48:02 +00:00
def save(self, *args, **kwargs):
2021-03-08 16:49:10 +00:00
with patch("django.db.models.Model.save"):
2021-02-06 21:48:02 +00:00
super().save(*args, **kwargs)
2021-03-08 16:49:10 +00:00
2021-02-06 21:48:02 +00:00
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, *_):
2021-04-26 16:15:42 +00:00
"""should create delete activities when objects are deleted by flag"""
2021-03-08 16:49:10 +00:00
class ActivitySuccess(Exception):
2021-04-26 16:15:42 +00:00
"""this means we got to the right method"""
class DeletableObjectModel(ObjectMixin, base_model.BookWyrmModel):
2021-04-26 16:15:42 +00:00
"""real simple mock model because BookWyrmModel is abstract"""
2021-03-08 16:49:10 +00:00
user = models.fields.ForeignKey("User", on_delete=db.models.CASCADE)
deleted = models.fields.BooleanField()
2021-03-08 16:49:10 +00:00
def save(self, *args, **kwargs):
2021-03-08 16:49:10 +00:00
with patch("django.db.models.Model.save"):
super().save(*args, **kwargs)
2021-03-08 16:49:10 +00:00
def to_delete_activity(self, user):
raise ActivitySuccess()
with self.assertRaises(ActivitySuccess):
2021-03-08 16:49:10 +00:00
DeletableObjectModel(id=1, user=self.local_user, deleted=True).save()
def test_to_delete_activity(self, *_):
2021-04-26 16:15:42 +00:00
"""wrapper for Delete activity"""
2021-03-08 16:49:10 +00:00
MockSelf = namedtuple("Self", ("remote_id", "to_activity"))
mock_self = MockSelf(
2021-03-08 16:49:10 +00:00
"https://example.com/status/1", lambda *args: self.object_mock
)
2021-03-08 16:49:10 +00:00
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(
2021-03-08 16:49:10 +00:00
activity["cc"], ["https://www.w3.org/ns/activitystreams#Public"]
)
def test_to_update_activity(self, *_):
2021-04-26 16:15:42 +00:00
"""ditto above but for Update"""
2021-03-08 16:49:10 +00:00
MockSelf = namedtuple("Self", ("remote_id", "to_activity"))
mock_self = MockSelf(
2021-03-08 16:49:10 +00:00
"https://example.com/status/1", lambda *args: self.object_mock
)
2021-03-08 16:49:10 +00:00
activity = ObjectMixin.to_update_activity(mock_self, self.local_user)
self.assertIsNotNone(
2021-03-08 16:49:10 +00:00
re.match(r"^https:\/\/example\.com\/status\/1#update\/.*", activity["id"])
)
2021-03-08 16:49:10 +00:00
self.assertEqual(activity["actor"], self.local_user.remote_id)
self.assertEqual(activity["type"], "Update")
self.assertEqual(
2021-03-08 16:49:10 +00:00
activity["to"], ["https://www.w3.org/ns/activitystreams#Public"]
)
self.assertIsInstance(activity["object"], dict)
def test_to_undo_activity(self, *_):
2021-04-26 16:15:42 +00:00
"""and again, for Undo"""
2021-03-08 16:49:10 +00:00
MockSelf = namedtuple("Self", ("remote_id", "to_activity", "user"))
mock_self = MockSelf(
2021-03-08 16:49:10 +00:00
"https://example.com/status/1",
2021-02-17 16:35:17 +00:00
lambda *args: self.object_mock,
self.local_user,
)
activity = ActivityMixin.to_undo_activity(mock_self)
2021-03-08 16:49:10 +00:00
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)
def test_to_ordered_collection_page(self, *_):
"""make sure the paged results of an ordered collection work"""
self.assertEqual(PAGE_LENGTH, 15)
for number in range(0, 2 * PAGE_LENGTH):
models.Status.objects.create(
user=self.local_user,
content="test status {:d}".format(number),
)
page_1 = to_ordered_collection_page(
models.Status.objects.all(), "http://fish.com/", page=1
)
self.assertEqual(page_1.partOf, "http://fish.com/")
self.assertEqual(page_1.id, "http://fish.com/?page=1")
self.assertEqual(page_1.next, "http://fish.com/?page=2")
self.assertEqual(page_1.orderedItems[0]["content"], "test status 29")
self.assertEqual(page_1.orderedItems[1]["content"], "test status 28")
page_2 = to_ordered_collection_page(
models.Status.objects.all(), "http://fish.com/", page=2
)
self.assertEqual(page_2.partOf, "http://fish.com/")
self.assertEqual(page_2.id, "http://fish.com/?page=2")
self.assertEqual(page_2.orderedItems[0]["content"], "test status 14")
self.assertEqual(page_2.orderedItems[-1]["content"], "test status 0")
def test_to_ordered_collection(self, *_):
"""convert a queryset into an ordered collection object"""
self.assertEqual(PAGE_LENGTH, 15)
for number in range(0, 2 * PAGE_LENGTH):
models.Status.objects.create(
user=self.local_user,
content="test status {:d}".format(number),
)
MockSelf = namedtuple("Self", ("remote_id"))
mock_self = MockSelf("")
collection = OrderedCollectionMixin.to_ordered_collection(
mock_self, models.Status.objects.all(), remote_id="http://fish.com/"
)
self.assertEqual(collection.totalItems, 30)
self.assertEqual(collection.first, "http://fish.com/?page=1")
self.assertEqual(collection.last, "http://fish.com/?page=2")
page_2 = OrderedCollectionMixin.to_ordered_collection(
mock_self, models.Status.objects.all(), remote_id="http://fish.com/", page=2
)
self.assertEqual(page_2.partOf, "http://fish.com/")
self.assertEqual(page_2.id, "http://fish.com/?page=2")
self.assertEqual(page_2.orderedItems[0]["content"], "test status 14")
self.assertEqual(page_2.orderedItems[-1]["content"], "test status 0")