kopia lustrzana https://github.com/snarfed/bridgy-fed
1025 wiersze
37 KiB
Python
1025 wiersze
37 KiB
Python
"""Unit tests for protocol.py."""
|
|
from unittest.mock import patch
|
|
|
|
from flask import g
|
|
from granary import as2
|
|
from oauth_dropins.webutil.flask_util import NoContent
|
|
from oauth_dropins.webutil.testutil import requests_response
|
|
import requests
|
|
|
|
# import first so that Fake is defined before URL routes are registered
|
|
from .testutil import Fake, TestCase
|
|
|
|
from activitypub import ActivityPub
|
|
from app import app
|
|
from models import Follower, Object, PROTOCOLS, User
|
|
import protocol
|
|
from protocol import Protocol
|
|
from ui import UIProtocol
|
|
from web import Web
|
|
from werkzeug.exceptions import BadRequest
|
|
|
|
from .test_activitypub import ACTOR
|
|
from .test_web import ACTOR_HTML
|
|
|
|
|
|
class ProtocolTest(TestCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.user = self.make_user('foo.com', has_hcard=True)
|
|
g.user = None
|
|
|
|
def tearDown(self):
|
|
PROTOCOLS.pop('greedy', None)
|
|
super().tearDown()
|
|
|
|
def test_protocols_global(self):
|
|
self.assertEqual(Fake, PROTOCOLS['fake'])
|
|
self.assertEqual(Web, PROTOCOLS['web'])
|
|
self.assertEqual(Web, PROTOCOLS['webmention'])
|
|
|
|
def test_for_domain_for_request(self):
|
|
for domain, expected in [
|
|
('fake.brid.gy', Fake),
|
|
('ap.brid.gy', ActivityPub),
|
|
('activitypub.brid.gy', ActivityPub),
|
|
('web.brid.gy', Web),
|
|
(None, None),
|
|
('', None),
|
|
('brid.gy', None),
|
|
('www.brid.gy', None),
|
|
('fed.brid.gy', None),
|
|
('fake.fed.brid.gy', None),
|
|
('fake', None),
|
|
('fake.com', None),
|
|
]:
|
|
with self.subTest(domain=domain, expected=expected):
|
|
self.assertEqual(expected, Protocol.for_domain(domain))
|
|
with app.test_request_context('/foo', base_url=f'https://{domain}/'):
|
|
self.assertEqual(expected, Protocol.for_request())
|
|
|
|
def test_for_domain_for_request_fed(self):
|
|
for url, expected in [
|
|
('https://fed.brid.gy/', Fake),
|
|
('http://localhost/foo', Fake),
|
|
('https://ap.brid.gy/bar', ActivityPub),
|
|
('https://baz/biff', None),
|
|
]:
|
|
with self.subTest(url=url, expected=expected):
|
|
self.assertEqual(expected, Protocol.for_domain(url, fed=Fake))
|
|
with app.test_request_context('/foo', base_url=url):
|
|
self.assertEqual(expected, Protocol.for_request(fed=Fake))
|
|
|
|
def test_subdomain_url(self):
|
|
self.assertEqual('https://fa.brid.gy/', Fake.subdomain_url())
|
|
self.assertEqual('https://fa.brid.gy/foo?bar', Fake.subdomain_url('foo?bar'))
|
|
self.assertEqual('https://fed.brid.gy/', UIProtocol.subdomain_url())
|
|
|
|
def test_for_id(self):
|
|
for id, expected in [
|
|
(None, None),
|
|
('', None),
|
|
('foo://bar', None),
|
|
('fake:foo', Fake),
|
|
# TODO
|
|
# ('at://foo', ATProto),
|
|
('https://ap.brid.gy/foo/bar', ActivityPub),
|
|
('https://web.brid.gy/foo/bar', Web),
|
|
]:
|
|
self.assertEqual(expected, Protocol.for_id(id))
|
|
|
|
def test_for_id_true_overrides_none(self):
|
|
class Greedy(Protocol, User):
|
|
@classmethod
|
|
def owns_id(cls, id):
|
|
return True
|
|
|
|
self.assertEqual(Greedy, Protocol.for_id('http://foo'))
|
|
self.assertEqual(Greedy, Protocol.for_id('https://bar/baz'))
|
|
|
|
def test_for_id_object(self):
|
|
self.store_object(id='http://ui/obj', source_protocol='ui')
|
|
self.assertEqual(UIProtocol, Protocol.for_id('http://ui/obj'))
|
|
|
|
def test_for_id_object_missing_source_protocol(self):
|
|
self.store_object(id='http://bad/obj')
|
|
self.assertIsNone(Protocol.for_id('http://bad/obj'))
|
|
|
|
@patch('requests.get')
|
|
def test_for_id_activitypub_fetch(self, mock_get):
|
|
mock_get.return_value = self.as2_resp(ACTOR)
|
|
self.assertEqual(ActivityPub, Protocol.for_id('http://ap/actor'))
|
|
self.assertIn(self.as2_req('http://ap/actor'), mock_get.mock_calls)
|
|
|
|
@patch('requests.get')
|
|
def test_for_id_web_fetch(self, mock_get):
|
|
mock_get.return_value = requests_response(ACTOR_HTML)
|
|
self.assertEqual(Web, Protocol.for_id('http://web.site/'))
|
|
self.assertIn(self.req('http://web.site/'), mock_get.mock_calls)
|
|
|
|
@patch('requests.get')
|
|
def test_for_id_web_fetch_no_mf2(self, mock_get):
|
|
mock_get.return_value = requests_response('<html></html>')
|
|
self.assertIsNone(Protocol.for_id('http://web.site/'))
|
|
self.assertIn(self.req('http://web.site/'), mock_get.mock_calls)
|
|
|
|
def test_load(self):
|
|
Fake.fetchable['foo'] = {'x': 'y'}
|
|
|
|
loaded = Fake.load('foo')
|
|
self.assert_equals({'x': 'y'}, loaded.our_as1)
|
|
self.assertFalse(loaded.changed)
|
|
self.assertTrue(loaded.new)
|
|
|
|
self.assertIsNotNone(Object.get_by_id('foo'))
|
|
self.assertEqual(['foo'], Fake.fetched)
|
|
|
|
def test_load_existing(self):
|
|
self.store_object(id='foo', our_as1={'x': 'y'})
|
|
|
|
loaded = Fake.load('foo')
|
|
self.assert_equals({'x': 'y'}, loaded.our_as1)
|
|
self.assertFalse(loaded.changed)
|
|
self.assertFalse(loaded.new)
|
|
|
|
self.assertEqual([], Fake.fetched)
|
|
|
|
def test_load_existing_empty_deleted(self):
|
|
stored = self.store_object(id='foo', deleted=True)
|
|
|
|
loaded = Fake.load('foo')
|
|
self.assert_entities_equal(stored, loaded)
|
|
self.assertFalse(loaded.changed)
|
|
self.assertFalse(loaded.new)
|
|
|
|
self.assertEqual([], Fake.fetched)
|
|
|
|
def test_load_cached(self):
|
|
obj = Object(id='foo', our_as1={'x': 'y'})
|
|
protocol.objects_cache['foo'] = obj
|
|
loaded = Fake.load('foo')
|
|
self.assert_entities_equal(obj, loaded)
|
|
|
|
# check that it's a separate copy of the entity in the cache
|
|
# https://github.com/snarfed/bridgy-fed/issues/558#issuecomment-1603203927
|
|
loaded.our_as1 = {'a': 'b'}
|
|
self.assertEqual({'x': 'y'}, Protocol.load('foo').our_as1)
|
|
|
|
def test_load_remote_true_existing_empty(self):
|
|
Fake.fetchable['foo'] = {'x': 'y'}
|
|
Object(id='foo').put()
|
|
|
|
loaded = Fake.load('foo', remote=True)
|
|
self.assertEqual({'x': 'y'}, loaded.as1)
|
|
self.assertTrue(loaded.changed)
|
|
self.assertFalse(loaded.new)
|
|
self.assertEqual(['foo'], Fake.fetched)
|
|
|
|
def test_load_remote_true_new_empty(self):
|
|
Fake.fetchable['foo'] = None
|
|
self.store_object(id='foo', our_as1={'x': 'y'})
|
|
|
|
loaded = Fake.load('foo', remote=True)
|
|
self.assertIsNone(loaded.as1)
|
|
self.assertTrue(loaded.changed)
|
|
self.assertFalse(loaded.new)
|
|
self.assertEqual(['foo'], Fake.fetched)
|
|
|
|
def test_load_remote_true_unchanged(self):
|
|
obj = self.store_object(id='foo', our_as1={'x': 'stored'},
|
|
source_protocol='fake')
|
|
Fake.fetchable['foo'] = {'x': 'stored'}
|
|
|
|
loaded = Fake.load('foo', remote=True)
|
|
self.assert_entities_equal(obj, loaded,
|
|
ignore=['expire', 'created', 'updated'])
|
|
self.assertFalse(loaded.changed)
|
|
self.assertFalse(loaded.new)
|
|
self.assertEqual(['foo'], Fake.fetched)
|
|
|
|
def test_load_remote_true_local_false(self):
|
|
Fake.fetchable['foo'] = our_as1={'x': 'y'}
|
|
|
|
loaded = Fake.load('foo', local=False, remote=True)
|
|
self.assertEqual({'x': 'y'}, loaded.as1)
|
|
self.assertIsNone(loaded.changed)
|
|
self.assertIsNone(loaded.new)
|
|
self.assertEqual(['foo'], Fake.fetched)
|
|
|
|
def test_load_remote_true_changed(self):
|
|
self.store_object(id='foo', our_as1={'content': 'stored'})
|
|
Fake.fetchable['foo'] = {'content': 'new'}
|
|
|
|
loaded = Fake.load('foo', remote=True)
|
|
self.assert_equals({'content': 'new'}, loaded.our_as1)
|
|
self.assertTrue(loaded.changed)
|
|
self.assertFalse(loaded.new)
|
|
self.assertEqual(['foo'], Fake.fetched)
|
|
|
|
def test_load_remote_false(self):
|
|
self.assertIsNone(Fake.load('nope', remote=False))
|
|
self.assertEqual([], Fake.fetched)
|
|
|
|
obj = self.store_object(id='foo', our_as1={'content': 'stored'})
|
|
self.assert_entities_equal(obj, Fake.load('foo', remote=False))
|
|
self.assertEqual([], Fake.fetched)
|
|
|
|
def test_Protocol_load_remote_false_existing_object_empty(self):
|
|
obj = self.store_object(id='foo')
|
|
self.assert_entities_equal(obj, Protocol.load('foo', remote=False))
|
|
|
|
def test_local_false_missing(self):
|
|
with self.assertRaises(requests.HTTPError) as e:
|
|
Fake.load('foo', local=False)
|
|
self.assertEqual(410, e.response.status_code)
|
|
|
|
self.assertEqual(['foo'], Fake.fetched)
|
|
|
|
def test_local_false_existing(self):
|
|
self.store_object(id='foo', our_as1={'content': 'stored'}, source_protocol='ui')
|
|
|
|
Fake.fetchable['foo'] = {'foo': 'bar'}
|
|
Fake.load('foo', local=False)
|
|
self.assert_object('foo', source_protocol='fake', our_as1={'foo': 'bar'})
|
|
self.assertEqual(['foo'], Fake.fetched)
|
|
|
|
def test_remote_false_local_false_assert(self):
|
|
with self.assertRaises(AssertionError):
|
|
Fake.load('nope', local=False, remote=False)
|
|
|
|
|
|
class ProtocolReceiveTest(TestCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
g.user = self.make_user('fake:user', cls=Fake, obj_id='fake:user')
|
|
self.alice = self.make_user('fake:alice', cls=Fake, obj_id='fake:alice')
|
|
self.bob = self.make_user('fake:bob', cls=Fake, obj_id='fake:bob')
|
|
|
|
def assert_object(self, id, **props):
|
|
props.setdefault('source_protocol', 'fake')
|
|
props.setdefault('delivered_protocol', 'fake')
|
|
|
|
ignore = []
|
|
for field in 'as2', 'bsky', 'mf2':
|
|
if 'our_as1' in props and field not in props:
|
|
ignore.append(field)
|
|
|
|
return super().assert_object(id, ignore=ignore, **props)
|
|
|
|
def make_followers(self):
|
|
Follower.get_or_create(to=g.user, from_=self.alice)
|
|
Follower.get_or_create(to=g.user, from_=self.bob)
|
|
Follower.get_or_create(to=g.user, from_=Fake(id='fake:eve'),
|
|
status='inactive')
|
|
|
|
def test_create_post(self):
|
|
self.make_followers()
|
|
|
|
post_as1 = {
|
|
'id': 'fake:post',
|
|
'objectType': 'note',
|
|
}
|
|
create_as1 = {
|
|
'id': 'fake:create',
|
|
'objectType': 'activity',
|
|
'verb': 'post',
|
|
'actor': 'fake:user',
|
|
'object': post_as1,
|
|
}
|
|
self.assertEqual('OK', Fake.receive(create_as1))
|
|
|
|
self.assert_object('fake:post',
|
|
our_as1=post_as1,
|
|
type='note',
|
|
)
|
|
obj = self.assert_object('fake:create',
|
|
status='complete',
|
|
our_as1=create_as1,
|
|
delivered=['shared:target'],
|
|
type='post',
|
|
labels=['user', 'activity', 'feed'],
|
|
users=[g.user.key, self.alice.key, self.bob.key],
|
|
)
|
|
|
|
self.assertEqual([(obj, 'shared:target')], Fake.sent)
|
|
|
|
def test_create_post_bare_object(self):
|
|
self.make_followers()
|
|
|
|
post_as1 = {
|
|
'id': 'fake:post',
|
|
'objectType': 'note',
|
|
'author': 'fake:user',
|
|
}
|
|
self.assertEqual('OK', Fake.receive(post_as1))
|
|
|
|
self.assert_object('fake:post',
|
|
our_as1=post_as1,
|
|
type='note',
|
|
)
|
|
|
|
obj = self.assert_object('fake:post#bridgy-fed-create',
|
|
status='complete',
|
|
our_as1={
|
|
'objectType': 'activity',
|
|
'verb': 'post',
|
|
'id': 'fake:post#bridgy-fed-create',
|
|
'actor': 'fake:user',
|
|
'object': post_as1,
|
|
'published': '2022-01-02T03:04:05+00:00',
|
|
},
|
|
delivered=['shared:target'],
|
|
type='post',
|
|
labels=['user', 'activity', 'feed'],
|
|
users=[g.user.key, self.alice.key, self.bob.key],
|
|
)
|
|
|
|
self.assertEqual([(obj, 'shared:target')], Fake.sent)
|
|
|
|
def test_update_post(self):
|
|
self.make_followers()
|
|
|
|
post_as1 = {
|
|
'id': 'fake:post',
|
|
'objectType': 'note',
|
|
}
|
|
self.store_object(id='fake:post', our_as1=post_as1)
|
|
|
|
update_as1 = {
|
|
'id': 'fake:update',
|
|
'objectType': 'activity',
|
|
'verb': 'update',
|
|
'actor': 'fake:user',
|
|
'object': post_as1,
|
|
}
|
|
self.assertEqual('OK', Fake.receive(update_as1))
|
|
|
|
self.assert_object('fake:post',
|
|
our_as1=post_as1,
|
|
type='note',
|
|
)
|
|
obj = self.assert_object('fake:update',
|
|
status='complete',
|
|
our_as1=update_as1,
|
|
delivered=['shared:target'],
|
|
type='update',
|
|
labels=['user', 'activity'],
|
|
users=[g.user.key],
|
|
)
|
|
|
|
self.assertEqual([(obj, 'shared:target')], Fake.sent)
|
|
|
|
def test_update_post_bare_object(self):
|
|
self.make_followers()
|
|
|
|
post_as1 = {
|
|
'id': 'fake:post',
|
|
'objectType': 'note',
|
|
'content': 'first',
|
|
}
|
|
self.store_object(id='fake:post', our_as1=post_as1)
|
|
existing = Object.get_by_id('fake:post')
|
|
|
|
post_as1['content'] = 'second'
|
|
self.assertEqual('OK', Fake.receive(post_as1))
|
|
|
|
post_as1['updated'] = '2022-01-02T03:04:05+00:00'
|
|
self.assert_object('fake:post',
|
|
our_as1=post_as1,
|
|
type='note',
|
|
)
|
|
|
|
update_id = 'fake:post#bridgy-fed-update-2022-01-02T03:04:05+00:00'
|
|
obj = self.assert_object(update_id,
|
|
status='complete',
|
|
our_as1={
|
|
'objectType': 'activity',
|
|
'verb': 'update',
|
|
'id': update_id,
|
|
'actor': 'fake:user',
|
|
'object': post_as1,
|
|
},
|
|
delivered=['shared:target'],
|
|
type='update',
|
|
labels=['user', 'activity'],
|
|
users=[g.user.key],
|
|
)
|
|
|
|
self.assertEqual([(obj, 'shared:target')], Fake.sent)
|
|
|
|
def test_create_reply(self):
|
|
self.make_followers()
|
|
|
|
Fake.fetchable['fake:post'] = {
|
|
'objectType': 'note',
|
|
'author': 'fake:bob',
|
|
}
|
|
reply_as1 = {
|
|
'id': 'fake:reply',
|
|
'objectType': 'note',
|
|
'inReplyTo': 'fake:post',
|
|
'author': 'fake:alice',
|
|
}
|
|
create_as1 = {
|
|
'id': 'fake:create',
|
|
'objectType': 'activity',
|
|
'verb': 'post',
|
|
'actor': 'fake:user',
|
|
'object': reply_as1,
|
|
}
|
|
self.assertEqual('OK', Fake.receive(create_as1))
|
|
|
|
self.assert_object('fake:reply',
|
|
our_as1=reply_as1,
|
|
type='note',
|
|
)
|
|
obj = self.assert_object('fake:create',
|
|
status='complete',
|
|
our_as1=create_as1,
|
|
delivered=['fake:post:target'],
|
|
type='post',
|
|
labels=['user', 'activity', 'notification'],
|
|
users=[g.user.key, self.bob.key, self.alice.key],
|
|
)
|
|
|
|
self.assertEqual([(obj, 'fake:post:target')], Fake.sent)
|
|
|
|
def test_create_reply_bare_object(self):
|
|
self.make_followers()
|
|
|
|
reply_as1 = {
|
|
'id': 'fake:reply',
|
|
'objectType': 'note',
|
|
'inReplyTo': 'fake:post',
|
|
'author': 'fake:alice',
|
|
}
|
|
Fake.fetchable['fake:post'] = {
|
|
'objectType': 'note',
|
|
'id': 'fake:post',
|
|
'author': 'fake:bob',
|
|
}
|
|
self.assertEqual('OK', Fake.receive(reply_as1))
|
|
|
|
self.assert_object('fake:reply',
|
|
our_as1=reply_as1,
|
|
type='note',
|
|
)
|
|
|
|
create_as1 = {
|
|
'id': 'fake:reply#bridgy-fed-create',
|
|
'objectType': 'activity',
|
|
'verb': 'post',
|
|
'actor': 'fake:alice',
|
|
'object': reply_as1,
|
|
'published': '2022-01-02T03:04:05+00:00',
|
|
}
|
|
obj = self.assert_object('fake:reply#bridgy-fed-create',
|
|
status='complete',
|
|
our_as1=create_as1,
|
|
delivered=['fake:post:target'],
|
|
type='post',
|
|
labels=['user', 'activity', 'notification'],
|
|
users=[self.alice.key, self.bob.key],
|
|
)
|
|
|
|
self.assertEqual([(obj, 'fake:post:target')], Fake.sent)
|
|
|
|
def test_update_reply(self):
|
|
self.make_followers()
|
|
|
|
Fake.fetchable['fake:post'] = {
|
|
'objectType': 'note',
|
|
'author': 'fake:bob',
|
|
}
|
|
reply_as1 = {
|
|
'id': 'fake:reply',
|
|
'objectType': 'note',
|
|
'inReplyTo': 'fake:post',
|
|
'author': 'fake:alice',
|
|
}
|
|
self.store_object(id='fake:reply', our_as1=reply_as1)
|
|
|
|
update_as1 = {
|
|
'id': 'fake:update',
|
|
'objectType': 'activity',
|
|
'verb': 'update',
|
|
'actor': 'fake:user',
|
|
'object': reply_as1,
|
|
}
|
|
self.assertEqual('OK', Fake.receive(update_as1))
|
|
|
|
self.assert_object('fake:reply',
|
|
our_as1=reply_as1,
|
|
type='note',
|
|
)
|
|
obj = self.assert_object('fake:update',
|
|
status='complete',
|
|
our_as1=update_as1,
|
|
delivered=['fake:post:target'],
|
|
type='update',
|
|
labels=['user', 'activity', 'notification'],
|
|
users=[g.user.key, self.alice.key, self.bob.key],
|
|
)
|
|
self.assertEqual([(obj, 'fake:post:target')], Fake.sent)
|
|
|
|
# TODO: revisit? remove?
|
|
# def test_reply_not_feed_not_notification(self):
|
|
# Follower.get_or_create(to=g.user, from_=self.alice)
|
|
# g.user = None
|
|
|
|
# Fake.fetchable['fake:post'] = {
|
|
# 'objectType': 'note',
|
|
# 'id': 'fake:post',
|
|
# 'author': 'fake:eve', # we have no user for this id
|
|
# }
|
|
# reply_as1 = {
|
|
# 'objectType': 'comment',
|
|
# 'id': 'fake:reply',
|
|
# 'author': 'fake:bob',
|
|
# 'content': 'A ☕ reply',
|
|
# 'inReplyTo': 'fake:post',
|
|
# }
|
|
# create_as1 = {
|
|
# 'objectType': 'post',
|
|
# 'id': 'fake:create',
|
|
# 'object': reply_as1,
|
|
# }
|
|
# Fake.receive(create_as1)
|
|
|
|
# self.assert_object('fake:reply',
|
|
# our_as1=reply_as1,
|
|
# type='comment',
|
|
# )
|
|
# self.assert_object('fake:create',
|
|
# our_as1=create_as1,
|
|
# type='post',
|
|
# users=[Fake(id='fake:eve').key],
|
|
# # not feed since it's a reply
|
|
# # not notification since it doesn't involve the user
|
|
# labels=['notification', 'user'],
|
|
# delivered=['fake:post:target'],
|
|
# status='complete',
|
|
# )
|
|
|
|
# self.assertEqual([(obj, 'fake:post:target')], Fake.sent)
|
|
|
|
def test_repost(self):
|
|
self.make_followers()
|
|
|
|
Fake.fetchable['fake:post'] = {
|
|
'objectType': 'note',
|
|
'author': 'fake:bob',
|
|
}
|
|
repost_as1 = {
|
|
'id': 'fake:repost',
|
|
'objectType': 'activity',
|
|
'verb': 'share',
|
|
'actor': 'fake:user',
|
|
'object': 'fake:post',
|
|
}
|
|
self.assertEqual('OK', Fake.receive(repost_as1))
|
|
|
|
obj = self.assert_object('fake:repost',
|
|
status='complete',
|
|
our_as1={
|
|
**repost_as1,
|
|
'object': {
|
|
'id': 'fake:post',
|
|
'objectType': 'note',
|
|
'author': 'fake:bob',
|
|
},
|
|
},
|
|
delivered=['fake:post:target', 'shared:target'],
|
|
type='share',
|
|
labels=['user', 'activity', 'notification', 'feed'],
|
|
users=[g.user.key, self.alice.key, self.bob.key],
|
|
)
|
|
self.assertEqual([
|
|
(obj, 'fake:post:target'),
|
|
(obj, 'shared:target'),
|
|
], Fake.sent)
|
|
|
|
def test_repost_twitter_blocklisted(self):
|
|
self._test_repost_blocklisted_error('https://twitter.com/foo')
|
|
|
|
def test_repost_bridgy_fed_blocklisted(self):
|
|
self._test_repost_blocklisted_error('https://fed.brid.gy/foo')
|
|
|
|
def _test_repost_blocklisted_error(self, orig_url):
|
|
"""Reposts of non-fediverse (ie blocklisted) sites aren't yet supported."""
|
|
repost_as1 = {
|
|
'id': 'fake:repost',
|
|
'objectType': 'activity',
|
|
'verb': 'share',
|
|
'actor': 'fake:user',
|
|
'object': orig_url,
|
|
}
|
|
with self.assertRaises(NoContent):
|
|
Fake.receive(repost_as1)
|
|
|
|
obj = self.assert_object('fake:repost',
|
|
status='ignored',
|
|
our_as1=repost_as1,
|
|
delivered=[],
|
|
type='share',
|
|
labels=['user', 'activity', 'feed'],
|
|
users=[g.user.key],
|
|
)
|
|
self.assertEqual([], Fake.sent)
|
|
|
|
def test_like(self):
|
|
Fake.fetchable['fake:post'] = {
|
|
'objectType': 'note',
|
|
}
|
|
|
|
like_as1 = {
|
|
'id': 'fake:like',
|
|
'objectType': 'activity',
|
|
'verb': 'like',
|
|
'actor': 'fake:user',
|
|
'object': 'fake:post',
|
|
}
|
|
self.assertEqual('OK', Fake.receive(like_as1))
|
|
|
|
like_obj = self.assert_object('fake:like',
|
|
users=[g.user.key],
|
|
status='complete',
|
|
our_as1=like_as1,
|
|
delivered=['fake:post:target'],
|
|
type='like',
|
|
labels=['user', 'activity'],
|
|
object_ids=['fake:post'])
|
|
|
|
self.assertEqual([(like_obj, 'fake:post:target')], Fake.sent)
|
|
|
|
def test_delete(self):
|
|
self.make_followers()
|
|
|
|
post_as1 = {
|
|
'id': 'fake:post',
|
|
'objectType': 'note',
|
|
'author': 'fake:user',
|
|
}
|
|
self.store_object(id='fake:post', our_as1=post_as1)
|
|
|
|
delete_as1 = {
|
|
'id': 'fake:delete',
|
|
'objectType': 'activity',
|
|
'verb': 'delete',
|
|
'object': 'fake:post',
|
|
}
|
|
self.assertEqual('OK', Fake.receive(delete_as1))
|
|
|
|
self.assert_object('fake:post',
|
|
our_as1=post_as1,
|
|
deleted=True,
|
|
source_protocol=None,
|
|
)
|
|
|
|
obj = self.assert_object('fake:delete',
|
|
status='complete',
|
|
our_as1=delete_as1,
|
|
delivered=['shared:target'],
|
|
type='delete',
|
|
labels=['user', 'activity', 'feed'],
|
|
users=[g.user.key, self.alice.key, self.bob.key],
|
|
)
|
|
self.assertEqual([(obj, 'shared:target')], Fake.sent)
|
|
|
|
def test_delete_no_followers_no_stored_object(self):
|
|
delete_as1 = {
|
|
'id': 'fake:delete',
|
|
'objectType': 'activity',
|
|
'verb': 'delete',
|
|
'object': 'fake:post',
|
|
}
|
|
with self.assertRaises(NoContent):
|
|
self.assertEqual('OK', Fake.receive(delete_as1))
|
|
|
|
self.assert_object('fake:post',
|
|
deleted=True,
|
|
source_protocol=None,
|
|
)
|
|
|
|
self.assert_object('fake:delete',
|
|
status='ignored',
|
|
our_as1=delete_as1,
|
|
delivered=[],
|
|
type='delete',
|
|
labels=['user', 'activity', 'feed'],
|
|
users=[g.user.key],
|
|
)
|
|
self.assertEqual([], Fake.sent)
|
|
|
|
def test_delete_actor(self):
|
|
follower = Follower.get_or_create(to=g.user, from_=self.alice)
|
|
followee = Follower.get_or_create(to=self.alice, from_=self.bob)
|
|
other = Follower.get_or_create(to=g.user, from_=self.bob)
|
|
self.assertEqual(3, Follower.query().count())
|
|
|
|
self.assertEqual('OK', Fake.receive({
|
|
'objectType': 'activity',
|
|
'verb': 'delete',
|
|
'id': 'fake:delete',
|
|
'object': 'fake:alice',
|
|
}))
|
|
|
|
self.assertEqual(3, Follower.query().count())
|
|
self.assertEqual('inactive', follower.key.get().status)
|
|
self.assertEqual('inactive', followee.key.get().status)
|
|
self.assertEqual('active', other.key.get().status)
|
|
|
|
self.assert_object('fake:alice',
|
|
deleted=True,
|
|
source_protocol=None,
|
|
)
|
|
|
|
# def test_send_error(self):
|
|
# mock_get.side_effect = [FOLLOW, ACTOR]
|
|
# mock_post.return_value = requests_response(
|
|
# 'abc xyz', status=405, url='https://mas.to/inbox')
|
|
|
|
# got = self.client.post('/_ah/queue/webmention', data={
|
|
# 'source': 'https://user.com/follow',
|
|
# 'target': 'https://fed.brid.gy/',
|
|
# })
|
|
# body = got.get_data(as_text=True)
|
|
# self.assertEqual(502, got.status_code, body)
|
|
# self.assertIn(
|
|
# '405 Client Error: None for url: https://mas.to/inbox ; abc xyz',
|
|
# body)
|
|
|
|
# mock_get.assert_has_calls((
|
|
# self.req('https://user.com/follow'),
|
|
# self.as2_req('https://mas.to/mrs-foo'),
|
|
# ))
|
|
|
|
# self.assert_deliveries(mock_post, ['https://mas.to/inbox'], FOLLOW_AS2)
|
|
|
|
# self.assert_object('https://user.com/follow',
|
|
# users=[g.user.key],
|
|
# status='failed',
|
|
# mf2=FOLLOW_MF2,
|
|
# as1=FOLLOW_AS1,
|
|
# failed=['https://mas.to/inbox'],
|
|
# type='follow',
|
|
# object_ids=['https://mas.to/mrs-foo'],
|
|
# labels=['user', 'activity'],
|
|
# )
|
|
|
|
def test_update_profile(self):
|
|
Follower.get_or_create(to=g.user, from_=self.alice)
|
|
Follower.get_or_create(to=g.user, from_=self.bob)
|
|
|
|
id = 'fake:user#update-2022-01-02T03:04:05+00:00'
|
|
update_as1 = {
|
|
'objectType': 'activity',
|
|
'verb': 'update',
|
|
'id': id,
|
|
'actor': 'fake:user',
|
|
'object': {
|
|
'objectType': 'person',
|
|
'id': 'fake:user',
|
|
'displayName': 'Ms. ☕ Baz',
|
|
'urls': [{'displayName': 'Ms. ☕ Baz', 'value': 'https://user.com/'}],
|
|
'updated': '2022-01-02T03:04:05+00:00',
|
|
},
|
|
}
|
|
|
|
Fake.receive(update_as1)
|
|
|
|
# profile object
|
|
self.assert_object('fake:user',
|
|
our_as1=update_as1['object'],
|
|
type='person',
|
|
)
|
|
|
|
# update activity
|
|
update_as1['actor'] = update_as1['object']
|
|
update_obj = self.assert_object(
|
|
id,
|
|
users=[g.user.key],
|
|
status='complete',
|
|
our_as1=update_as1,
|
|
delivered=['shared:target'],
|
|
type='update',
|
|
object_ids=['fake:user'],
|
|
labels=['user', 'activity'],
|
|
)
|
|
|
|
self.assertEqual([(update_obj, 'shared:target')], Fake.sent)
|
|
|
|
# def test_mention_object(self, *mocks):
|
|
# self._test_mention(
|
|
# MENTION_OBJECT,
|
|
# {
|
|
# 'type': 'note', # not mention (?)
|
|
# 'labels': ['notification'],
|
|
# },
|
|
# *mocks,
|
|
# )
|
|
|
|
# def test_mention_create_activity(self, *mocks):
|
|
# self._test_mention(
|
|
# MENTION,
|
|
# {
|
|
# 'type': 'post', # not mention (?)
|
|
# 'object_ids': [MENTION_OBJECT['id']],
|
|
# 'labels': ['notification', 'activity'],
|
|
# },
|
|
# *mocks,
|
|
# )
|
|
|
|
# # redirect unwrap
|
|
# expected_as2 = copy.deepcopy(MENTION_OBJECT)
|
|
# expected_as2['tag'][1]['href'] = 'https://tar.get/'
|
|
# self.assert_object(MENTION_OBJECT['id'],
|
|
# as2=expected_as2,
|
|
# type='note')
|
|
|
|
# def _test_mention(self, mention, expected_props):
|
|
# self.make_user('tar.get')
|
|
|
|
# mock_get.side_effect = [
|
|
# self.as2_resp(ACTOR),
|
|
# requests_response(test_web.NOTE_HTML),
|
|
# requests_response(test_web.NOTE_HTML),
|
|
# WEBMENTION_DISCOVERY,
|
|
# ]
|
|
# mock_post.return_value = requests_response()
|
|
|
|
# got = self.post('/user.com/inbox', json=mention)
|
|
# self.assertEqual(200, got.status_code, got.get_data(as_text=True))
|
|
# self.assert_req(mock_get, 'https://tar.get/')
|
|
# convert_id = mention['id'].replace('://', ':/')
|
|
# self.assert_req(
|
|
# mock_post,
|
|
# 'https://tar.get/webmention',
|
|
# headers={'Accept': '*/*'},
|
|
# allow_redirects=False,
|
|
# data={
|
|
# 'source': f'http://localhost/convert/activitypub/web/{convert_id}',
|
|
# 'target': 'https://tar.get/',
|
|
# },
|
|
# )
|
|
|
|
# expected_as2 = common.redirect_unwrap(mention)
|
|
# self.assert_object(mention['id'],
|
|
# users=[Web(id='tar.get').key],
|
|
# status='complete',
|
|
# as2=expected_as2,
|
|
# delivered=['https://tar.get/'],
|
|
# **expected_props)
|
|
|
|
def test_follow(self):
|
|
self._test_follow()
|
|
|
|
def test_follow_no_g_user(self):
|
|
"""No user from request, eg delivered to our ActivityPub shared inbox."""
|
|
g.user = None
|
|
self.test_follow()
|
|
|
|
def test_follow_existing_inactive(self):
|
|
follower = Follower.get_or_create(to=g.user, from_=self.alice,
|
|
status='inactive')
|
|
self._test_follow()
|
|
|
|
def test_follow_actor_object_composite_objects(self):
|
|
self._test_follow(actor={'id': 'fake:alice', 'objectType': 'person'},
|
|
object={'id': 'fake:user', 'objectType': 'person'})
|
|
|
|
def _test_follow(self, **extra):
|
|
Fake.fetchable['fake:alice'] = {}
|
|
|
|
follow_as1 = {
|
|
'id': 'fake:follow',
|
|
'objectType': 'activity',
|
|
'verb': 'follow',
|
|
'actor': 'fake:alice',
|
|
'object': 'fake:user',
|
|
**extra,
|
|
}
|
|
self.assertEqual('OK', Fake.receive(follow_as1))
|
|
|
|
user = Fake.get_by_id('fake:user')
|
|
follow_obj = self.assert_object('fake:follow',
|
|
our_as1=follow_as1,
|
|
status='complete',
|
|
users=[self.alice.key, user.key],
|
|
labels=['activity', 'user', 'notification'],
|
|
delivered=['fake:user:target'],
|
|
)
|
|
|
|
accept_id = 'http://localhost/fa/fake:user/followers#accept-fake:follow'
|
|
accept_as1 = {
|
|
'id': accept_id,
|
|
'objectType': 'activity',
|
|
'verb': 'accept',
|
|
'actor': 'fake:user',
|
|
'object': follow_as1,
|
|
}
|
|
accept_obj = self.assert_object(accept_id,
|
|
our_as1=accept_as1,
|
|
type='accept',
|
|
labels=['activity'],
|
|
status='complete',
|
|
delivered=['fake:alice:target'],
|
|
users=[],
|
|
source_protocol=None,
|
|
)
|
|
|
|
self.assertEqual([
|
|
(accept_obj, 'fake:alice:target'),
|
|
(follow_obj, 'fake:user:target'),
|
|
], Fake.sent)
|
|
|
|
self.assert_entities_equal(
|
|
Follower(to=user.key, from_=self.alice.key, status='active',
|
|
follow=follow_obj.key),
|
|
Follower.query().fetch(),
|
|
ignore=['created', 'updated'],
|
|
)
|
|
|
|
def test_follow_no_actor(self):
|
|
with self.assertRaises(BadRequest):
|
|
Fake.receive({
|
|
'id': 'fake:follow',
|
|
'objectType': 'activity',
|
|
'verb': 'follow',
|
|
'object': 'fake:user',
|
|
})
|
|
|
|
self.assertEqual([], Follower.query().fetch())
|
|
|
|
def test_follow_no_object(self):
|
|
with self.assertRaises(BadRequest):
|
|
Fake.receive({
|
|
'id': 'fake:follow',
|
|
'objectType': 'activity',
|
|
'verb': 'follow',
|
|
'actor': 'fake:alice',
|
|
})
|
|
|
|
self.assertEqual([], Follower.query().fetch())
|
|
|
|
def test_undo_follow(self):
|
|
follower = Follower.get_or_create(to=g.user, from_=self.alice)
|
|
Fake.fetchable['fake:alice'] = {}
|
|
|
|
self.assertEqual('OK', Fake.receive({
|
|
'id': 'fake:undo-follow',
|
|
'objectType': 'activity',
|
|
'verb': 'stop-following',
|
|
'actor': 'fake:alice',
|
|
'object': 'fake:user',
|
|
}))
|
|
|
|
self.assertEqual('inactive', follower.key.get().status)
|
|
|
|
def test_undo_follow_doesnt_exist(self):
|
|
self.assertEqual('OK', Fake.receive({
|
|
'id': 'fake:undo-follow',
|
|
'objectType': 'activity',
|
|
'verb': 'stop-following',
|
|
'actor': 'fake:alice',
|
|
'object': 'fake:user',
|
|
}))
|
|
# it's a noop
|
|
self.assertEqual(0, Follower.query().count())
|
|
|
|
def test_undo_follow_inactive(self):
|
|
follower = Follower.get_or_create(to=g.user, from_=self.alice,
|
|
status='inactive')
|
|
Fake.fetchable['fake:alice'] = {}
|
|
|
|
self.assertEqual('OK', Fake.receive({
|
|
'id': 'fake:undo-follow',
|
|
'objectType': 'activity',
|
|
'verb': 'stop-following',
|
|
'actor': 'fake:alice',
|
|
'object': 'fake:user',
|
|
}))
|
|
self.assertEqual('inactive', follower.key.get().status)
|
|
|
|
def test_receive_from_bridgy_fed_fails(self):
|
|
with self.assertRaises(BadRequest):
|
|
Fake.receive({
|
|
'id': 'https://fed.brid.gy/r/foo',
|
|
})
|
|
|
|
self.assertIsNone(Object.get_by_id('https://fed.brid.gy/r/foo'))
|
|
|
|
with self.assertRaises(BadRequest):
|
|
Fake.receive({
|
|
'id': 'fake:foo',
|
|
'actor': 'https://ap.brid.gy/user.com',
|
|
})
|
|
|
|
self.assertIsNone(Object.get_by_id('foo'))
|
|
self.assertIsNone(Object.get_by_id('https://ap.brid.gy/user.com'))
|
|
|
|
# def test_skip_same_domain_target(self):
|
|
# TODO
|