# coding=utf-8 """Unit tests for webmention.py.""" import copy from unittest.mock import patch from flask import g, get_flashed_messages from google.cloud import ndb from granary import as1, as2, microformats2 from oauth_dropins.webutil import util from oauth_dropins.webutil.appengine_info import APP_ID from oauth_dropins.webutil.testutil import NOW, requests_response from oauth_dropins.webutil.util import json_dumps, json_loads import requests from werkzeug.exceptions import BadGateway, BadRequest # import first so that Fake is defined before URL routes are registered from . import testutil from activitypub import ActivityPub, postprocess_as2 from common import CONTENT_TYPE_HTML from models import Follower, Object from web import TASKS_LOCATION, Web from . import test_activitypub from .testutil import TestCase FULL_REDIR = requests_response( status=302, redirected_url='http://localhost/.well-known/webfinger?resource=acct:user.com@user.com') ACTOR_HTML = """\
Ms. ☕ Baz """ ACTOR_HTML_RESP = requests_response(ACTOR_HTML, url='https://user.com/') ACTOR_MF2 = { 'type': ['h-card'], 'properties': { 'url': ['https://user.com/'], 'name': ['Ms. ☕ Baz'], }, } ACTOR_MF2_REL_URLS = { **ACTOR_MF2, 'rel-urls': {'https://user.com/': {'rels': ['me'], 'text': 'Ms. ☕ Baz'}} } ACTOR_AS1_UNWRAPPED = { 'objectType': 'person', 'id': 'https://user.com/', 'url': 'https://user.com/', 'displayName': 'Ms. ☕ Baz', } ACTOR_AS2 = { 'type': 'Person', 'id': 'http://localhost/user.com', 'url': 'http://localhost/r/https://user.com/', 'name': 'Ms. ☕ Baz', 'preferredUsername': 'user.com', 'inbox': 'http://localhost/user.com/inbox', 'outbox': 'http://localhost/user.com/outbox', } ACTOR_AS2_USER = { 'type': 'Person', 'id': 'https://user.com/', 'url': 'https://user.com/', 'name': 'Ms. ☕ Baz', 'attachment': [{ 'name': 'Ms. ☕ Baz', 'type': 'PropertyValue', 'value': 'https://user.com/', }], } ACTOR_AS2_FULL = { **ACTOR_AS2, '@context': [ 'https://www.w3.org/ns/activitystreams', 'https://w3id.org/security/v1', ], 'attachment': [{ 'name': 'Web site', 'type': 'PropertyValue', 'value': 'https://user.com/', }], 'inbox': 'http://localhost/user.com/inbox', 'outbox': 'http://localhost/user.com/outbox', 'following': 'http://localhost/user.com/following', 'followers': 'http://localhost/user.com/followers', 'endpoints': { 'sharedInbox': 'http://localhost/inbox', }, } REPOST_HTML = """\ reposted! Ms. ☕ Baz """ REPOST = requests_response(REPOST_HTML, url='https://user.com/repost') REPOST_MF2 = util.parse_mf2(REPOST_HTML)['items'][0] REPOST_AS2 = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Announce', 'id': 'http://localhost/r/https://user.com/repost', 'url': 'http://localhost/r/https://user.com/repost', 'name': 'reposted!', 'object': 'https://mas.to/toot/id', 'to': [as2.PUBLIC_AUDIENCE], 'cc': [ 'https://mas.to/author', 'https://mas.to/bystander', 'https://mas.to/recipient', as2.PUBLIC_AUDIENCE, ], 'actor': 'http://localhost/user.com', } REPOST_HCITE_HTML = """\ reposted! Ms. ☕ Baz """ REPOST_HCITE = requests_response(REPOST_HTML, url='https://user.com/repost') WEBMENTION_REL_LINK = requests_response( '') WEBMENTION_NO_REL_LINK = requests_response('') DELETE_AS1 = { 'objectType': 'activity', 'verb': 'delete', 'id': 'https://user.com/post#bridgy-fed-delete', 'actor': 'http://localhost/user.com', 'object': 'https://user.com/post', } DELETE_AS2 = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Delete', 'id': 'http://localhost/r/https://user.com/post#bridgy-fed-delete', 'actor': 'http://localhost/user.com', 'object': 'http://localhost/r/https://user.com/post', 'to': [as2.PUBLIC_AUDIENCE], } TOOT_HTML = requests_response("""\ """, url='https://mas.to/toot') TOOT_AS2_DATA = { '@context': ['https://www.w3.org/ns/activitystreams'], 'type': 'Article', 'id': 'https://mas.to/toot/id', 'content': 'Lots of ☕ words...', 'actor': {'url': 'https://mas.to/author'}, 'to': ['https://mas.to/recipient', as2.PUBLIC_AUDIENCE], 'cc': ['https://mas.to/bystander', as2.PUBLIC_AUDIENCE], } TOOT_AS2 = requests_response( TOOT_AS2_DATA, url='https://mas.to/toot/id', content_type=as2.CONTENT_TYPE + '; charset=utf-8') REPLY_HTML = """\ """ REPLY = requests_response(REPLY_HTML, url='https://user.com/reply') REPLY_MF2 = util.parse_mf2(REPLY_HTML)['items'][0] REPLY_AS1 = microformats2.json_to_object(REPLY_MF2) REPLY_AS1['id'] = 'https://user.com/reply' REPLY_AS1['author']['id'] = 'https://user.com/' CREATE_REPLY_AS1 = { 'objectType': 'activity', 'verb': 'post', 'id': 'https://user.com/reply#bridgy-fed-create', 'actor': ACTOR_AS1_UNWRAPPED, 'object': REPLY_AS1, 'published': '2022-01-02T03:04:05+00:00', } REPLY_AS2 = as2.from_as1(REPLY_AS1) LIKE_HTML = """\ Ms. ☕ Baz """ LIKE = requests_response(LIKE_HTML, url='https://user.com/like') LIKE_MF2 = util.parse_mf2(LIKE_HTML)['items'][0] ACTOR = TestCase.as2_resp({ 'type': 'Person', 'name': 'Mrs. ☕ Foo', 'id': 'https://mas.to/mrs-foo', 'inbox': 'https://mas.to/inbox', }) AS2_CREATE = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Create', 'id': 'http://localhost/r/https://user.com/reply#bridgy-fed-create', 'actor': 'http://localhost/user.com', 'published': '2022-01-02T03:04:05+00:00', 'object': { 'type': 'Note', 'id': 'http://localhost/r/https://user.com/reply', 'url': 'http://localhost/r/https://user.com/reply', 'name': 'foo ☕ bar', 'content': """\ foo ☕ bar """, 'inReplyTo': 'https://mas.to/toot/id', 'to': [as2.PUBLIC_AUDIENCE], 'cc': [ 'https://mas.to/author', 'https://mas.to/bystander', 'https://mas.to/recipient', as2.PUBLIC_AUDIENCE, ], 'attributedTo': ACTOR_AS2, 'tag': [{ 'type': 'Mention', 'href': 'https://mas.to/author', }], }, 'to': [as2.PUBLIC_AUDIENCE], } AS2_UPDATE = copy.deepcopy(AS2_CREATE) AS2_UPDATE.update({ 'id': 'http://localhost/r/https://user.com/reply#bridgy-fed-update-2022-01-02T03:04:05+00:00', 'type': 'Update', }) del AS2_UPDATE['published'] # we should generate this if it's not already in mf2 because Mastodon # requires it for updates AS2_UPDATE['object']['updated'] = NOW.isoformat() FOLLOW_HTML = """\ Ms. ☕ Baz """ FOLLOW = requests_response(FOLLOW_HTML, url='https://user.com/follow') FOLLOW_MF2 = util.parse_mf2(FOLLOW_HTML)['items'][0] FOLLOW_AS1 = microformats2.json_to_object(FOLLOW_MF2) FOLLOW_AS2 = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Follow', 'id': 'http://localhost/r/https://user.com/follow', 'url': 'http://localhost/r/https://user.com/follow', 'object': 'https://mas.to/mrs-foo', 'actor': 'http://localhost/user.com', 'to': [as2.PUBLIC_AUDIENCE], } FOLLOW_FRAGMENT_HTML = """\hello i am a post
Ms. ☕ Baz
""" NOTE = requests_response(NOTE_HTML, url='https://user.com/post') NOTE_MF2 = util.parse_mf2(NOTE_HTML)['items'][0] NOTE_AS1 = microformats2.json_to_object(NOTE_MF2) NOTE_AS1.update({ 'author': { **NOTE_AS1['author'], 'id': 'https://user.com/', }, 'id': 'https://user.com/post', }) NOTE_AS2 = { 'type': 'Note', 'id': 'http://localhost/r/https://user.com/post', 'url': 'http://localhost/r/https://user.com/post', 'attributedTo': ACTOR_AS2, 'name': 'hello i am a post', 'content': 'hello i am a post', 'to': [as2.PUBLIC_AUDIENCE], } CREATE_AS1 = { 'objectType': 'activity', 'verb': 'post', 'id': 'https://user.com/post#bridgy-fed-create', 'actor': ACTOR_AS1_UNWRAPPED, 'object': NOTE_AS1, 'published': '2022-01-02T03:04:05+00:00', } CREATE_AS2 = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Create', 'id': 'http://localhost/r/https://user.com/post#bridgy-fed-create', 'actor': 'http://localhost/user.com', 'object': NOTE_AS2, 'published': '2022-01-02T03:04:05+00:00', 'to': [as2.PUBLIC_AUDIENCE], } UPDATE_AS2 = copy.deepcopy(CREATE_AS2) UPDATE_AS2.update({ 'type': 'Update', 'id': 'http://localhost/r/https://user.com/post#bridgy-fed-update-2022-01-02T03:04:05+00:00', }) del UPDATE_AS2['published'] UPDATE_AS2['object']['updated'] = NOW.isoformat() NOT_FEDIVERSE = requests_response("""\ foo """, url='http://not/fediverse') ACTIVITYPUB_GETS = [ REPLY, NOT_FEDIVERSE, # AP NOT_FEDIVERSE, # Web TOOT_AS2, # AP ACTOR, ] @patch('requests.post') @patch('requests.get') class WebTest(TestCase): def setUp(self): super().setUp() obj = Object(id='https://user.com/', mf2=ACTOR_MF2, source_protocol='web') obj.put() g.user = self.make_user('user.com', has_redirects=True, obj=obj) self.mrs_foo = ndb.Key(ActivityPub, 'https://mas.to/mrs-foo') def assert_deliveries(self, mock_post, inboxes, data, ignore=()): self.assertEqual(len(inboxes), len(mock_post.call_args_list), mock_post.call_args_list) calls = {} # maps inbox URL to JSON data for args, kwargs in mock_post.call_args_list: self.assertEqual(as2.CONTENT_TYPE, kwargs['headers']['Content-Type']) rsa_key = kwargs['auth'].header_signer._rsa._key self.assertEqual(g.user.private_pem(), rsa_key.exportKey()) calls[args[0]] = json_loads(kwargs['data']) for inbox in inboxes: got = calls[inbox] as1.get_object(got).pop('publicKey', None) self.assert_equals(data, got, inbox, ignore=ignore) def assert_object(self, id, **props): return super().assert_object(id, delivered_protocol='activitypub', **props) def make_followers(self): self.followers = [] for id, kwargs, actor in [ ('https://mastodon/aaa', {}, None), ('https://mastodon/bbb', {}, { 'publicInbox': 'https://public/inbox', 'inbox': 'https://unused', }), ('https://mastodon/ccc', {}, { 'endpoints': { 'sharedInbox': 'https://shared/inbox', }, }), ('https://mastodon/ddd', {}, { 'inbox': 'https://inbox', }), ('https://mastodon/ggg', {'status': 'inactive'}, { 'inbox': 'https://unused/2', }), ('https://mastodon/hhh', {}, { # dupe of ddd; should be de-duped 'inbox': 'https://inbox', }), ]: from_ = self.make_user(id, cls=ActivityPub, obj_as2=actor) f = Follower.get_or_create(to=g.user, from_=from_, **kwargs) if f.status != 'inactive': self.followers.append(from_.key) def test_put_validates_domain_id(self, *_): for bad in ( 'AbC.cOm', 'foo', '@user.com', '@user.com@user.com', 'acct:user.com', 'acct:@user.com@user.com', 'acc:me@user.com', 'fed.brid.gy', 'ap.brid.gy', 'localhost', ): with self.assertRaises(AssertionError): Web(id=bad).put() def test_get_or_create_lower_cases_domain(self, *_): user = Web.get_or_create('AbC.oRg') self.assertEqual('abc.org', user.key.id()) self.assert_entities_equal(user, Web.get_by_id('abc.org')) self.assertIsNone(Web.get_by_id('AbC.oRg')) def test_get_or_create_unicode_domain(self, *_): user = Web.get_or_create('☃.net') self.assertEqual('☃.net', user.key.id()) self.assert_entities_equal(user, Web.get_by_id('☃.net')) def test_get_or_create_scripts_leading_trailing_dots(self, *_): user = Web.get_or_create('..foo.bar.') self.assertEqual('foo.bar', user.key.id()) self.assert_entities_equal(user, Web.get_by_id('foo.bar')) self.assertIsNone(Web.get_by_id('..foo.bar.')) def test_bad_source_url(self, *mocks): orig_count = Object.query().count() for data in b'', {'source': 'bad'}, {'source': 'https://'}: got = self.client.post('/webmention', data=data) self.assertEqual(400, got.status_code) self.assertEqual(orig_count, Object.query().count()) def test_username(self, *mocks): self.assertEqual('user.com', g.user.username()) g.user.obj = Object(id='a', as2={ 'type': 'Person', 'name': 'foo', 'url': ['bar'], 'preferredUsername': 'baz', }) g.user.direct = True self.assertEqual('user.com', g.user.username()) # bad acct: URI, util.parse_acct_uri raises ValueError # https://console.cloud.google.com/errors/detail/CPLmrpzFs4qTUA;time=P30D?project=bridgy-federated g.user.obj.as2['url'].append('acct:@user.com') self.assertEqual('user.com', g.user.username()) g.user.obj.as2['url'].append('acct:alice@foo.com') self.assertEqual('user.com', g.user.username()) g.user.obj.as2['url'].append('acct:alice@user.com') self.assertEqual('alice', g.user.username()) g.user.direct = False self.assertEqual('user.com', g.user.username()) @patch('oauth_dropins.webutil.appengine_config.tasks_client.create_task') def test_make_task(self, mock_create_task, mock_get, mock_post): mock_get.side_effect = [NOTE, ACTOR] params = { 'source': 'https://user.com/post', 'target': 'https://fed.brid.gy/', } got = self.client.post('/webmention', data=params) self.assertEqual(202, got.status_code) self.assert_task(mock_create_task, 'webmention', '/_ah/queue/webmention', **params) def test_no_user(self, mock_get, mock_post): orig_count = Object.query().count() got = self.client.post('/webmention', data={'source': 'https://nope.com/post'}) self.assertEqual(400, got.status_code) self.assertEqual(orig_count, Object.query().count()) def test_source_fetch_fails(self, mock_get, mock_post): orig_count = Object.query().count() mock_get.side_effect = ( requests_response(REPLY_HTML, status=405), ) got = self.client.post('/_ah/queue/webmention', data={'source': 'https://user.com/post'}) self.assertEqual(502, got.status_code) self.assertEqual(orig_count, Object.query().count()) def test_no_source_entry(self, mock_get, mock_post): orig_count = Object.query().count() mock_get.return_value = requests_response("""nothing to see here except link
""", url='https://user.com/post') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/post', 'target': 'https://fed.brid.gy/', }) self.assertEqual(304, got.status_code) self.assertEqual(orig_count, Object.query().count()) mock_get.assert_has_calls((self.req('https://user.com/post'),)) def test_source_homepage_no_mf2(self, mock_get, mock_post): mock_get.return_value = requests_response("""nothing to see here except link
""", url='https://user.com/') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/', 'target': 'https://fed.brid.gy/', }) self.assertEqual(304, got.status_code) mock_get.assert_has_calls((self.req('https://user.com/'),)) def test_no_targets(self, mock_get, mock_post): mock_get.return_value = requests_response("""no one to send to!
""", url='https://user.com/post') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/post', 'target': 'https://fed.brid.gy/', }) self.assertEqual(204, got.status_code) mock_get.assert_has_calls((self.req('https://user.com/post'),)) def test_bad_target_url(self, mock_get, mock_post): mock_get.side_effect = ( requests_response( REPLY_HTML.replace('https://mas.to/toot', 'bad:nope'), content_type=CONTENT_TYPE_HTML, url='https://user.com/reply'), ValueError('foo bar'), # AS2 fetch ValueError('foo bar'), # HTML fetch ) got = self.client.post('/_ah/queue/webmention', data={'source': 'https://user.com/reply'}) self.assertEqual(204, got.status_code) self.assert_object('https://user.com/reply', source_protocol='web', type='comment', labels=[], ignore=['our_as1'], ) self.assert_object('https://user.com/reply#bridgy-fed-create', source_protocol='web', our_as1=CREATE_REPLY_AS1, type='post', labels=['activity', 'user'], ignore=['our_as1'], status='ignored', users=[g.user.key], ) def test_target_fetch_fails(self, mock_get, mock_post): mock_get.side_effect = [ requests_response( REPLY_HTML.replace('https://mas.to/toot', 'bad:nope'), url='https://user.com/post'), # http://not/fediverse AP protocol discovery requests.Timeout('foo bar'), # http://not/fediverse web protocol discovery requests.Timeout('foo bar'), ] got = self.client.post('/_ah/queue/webmention', data={'source': 'https://user.com/reply'}) self.assertEqual(204, got.status_code) def test_target_fetch_has_no_content_type(self, mock_get, mock_post): Object(id='http://not/fediverse', mf2=NOTE_MF2, source_protocol='web').put() no_content_type = requests_response(REPLY_HTML, content_type='') mock_get.side_effect = ( requests_response(REPLY_HTML, url='https://user.com/reply'), # requests: no_content_type, # https://mas.to/toot AP protocol discovery no_content_type, # https://mas.to/toot Web protocol discovery no_content_type, # https://user.com/ webmention discovery no_content_type, # http://not/fediverse webmention discovery ) got = self.client.post('/_ah/queue/webmention', data={'source': 'https://user.com/reply'}) self.assertEqual(204, got.status_code) mock_post.assert_not_called() def test_missing_backlink(self, mock_get, mock_post): orig_count = Object.query().count() mock_get.return_value = requests_response( REPLY_HTML.replace('', ''), url='https://user.com/reply') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/reply', 'target': 'https://fed.brid.gy/', }) self.assertEqual(304, got.status_code) self.assertEqual(orig_count, Object.query().count()) mock_get.assert_has_calls((self.req('https://user.com/reply'),)) def test_backlink_without_trailing_slash(self, mock_get, mock_post): mock_get.return_value = requests_response( REPLY_HTML.replace('', ''), content_type=CONTENT_TYPE_HTML, url='https://user.com/reply') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/reply', 'target': 'https://fed.brid.gy/', }) self.assertEqual(204, got.status_code) def test_create_reply(self, mock_get, mock_post): mock_get.side_effect = ACTIVITYPUB_GETS mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/reply', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) mock_get.assert_has_calls(( self.req('https://user.com/reply'), self.as2_req('http://not/fediverse'), self.req('http://not/fediverse'), self.as2_req('https://mas.to/toot'), self.as2_req('https://mas.to/author'), )) self.assert_deliveries(mock_post, ['https://mas.to/inbox'], AS2_CREATE) self.assert_object('https://user.com/reply', source_protocol='web', our_as1=REPLY_AS1, type='comment', ) author = ndb.Key(ActivityPub, 'https://mas.to/author') self.assert_object('https://user.com/reply#bridgy-fed-create', users=[g.user.key], notify=[author], source_protocol='web', status='complete', our_as1=CREATE_REPLY_AS1, delivered=['https://mas.to/inbox'], type='post', ) def test_update_reply(self, mock_get, mock_post): self.make_followers() mf2 = { 'properties': { 'content': ['other'], }, } Object(id='https://user.com/reply', status='complete', mf2=mf2).put() mock_get.side_effect = ACTIVITYPUB_GETS mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/reply', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) self.assertEqual(1, mock_post.call_count) args, kwargs = mock_post.call_args self.assertEqual(('https://mas.to/inbox',), args) self.assert_equals(AS2_UPDATE, json_loads(kwargs['data'])) def test_redo_repost_isnt_update(self, mock_get, mock_post): """Like and Announce shouldn't use Update, they should just resend as is.""" Object(id='https://user.com/repost', mf2={}, status='complete').put() mock_get.side_effect = [REPOST, TOOT_AS2, ACTOR] mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/repost', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) self.assert_deliveries(mock_post, ['https://mas.to/inbox'], REPOST_AS2, ignore=['cc']) def test_skip_update_if_content_unchanged(self, mock_get, mock_post): """https://github.com/snarfed/bridgy-fed/issues/78""" self.store_object(id='https://user.com/reply', mf2=REPLY_MF2) self.store_object(id='https://user.com/reply#bridgy-fed-create', status='complete') mock_get.side_effect = ACTIVITYPUB_GETS got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/reply', 'target': 'https://fed.brid.gy/', }) self.assertEqual(204, got.status_code) mock_post.assert_not_called() def test_force_with_content_unchanged_sends_create(self, mock_get, mock_post): Object(id='https://user.com/reply', mf2=REPLY_MF2).put() mock_get.side_effect = ACTIVITYPUB_GETS mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/reply', 'target': 'https://fed.brid.gy/', 'force': '', }) self.assertEqual(200, got.status_code) args, kwargs = mock_post.call_args self.assertEqual(('https://mas.to/inbox',), args) self.assert_equals(AS2_CREATE, json_loads(kwargs['data'])) def test_create_reply_attributed_to_id_only(self, mock_get, mock_post): """Based on PeerTube's AS2. https://github.com/snarfed/bridgy-fed/issues/40 """ toot_as2_data = copy.deepcopy(TOOT_AS2_DATA) del toot_as2_data['actor'] toot_as2_data['attributedTo'] = { 'type': 'Person', 'id': 'https://mas.to/author', } mock_get.side_effect = [ REPLY, NOT_FEDIVERSE, # AP NOT_FEDIVERSE, # Web self.as2_resp(toot_as2_data), # AP ACTOR, ] mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/reply', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) mock_get.assert_has_calls(( self.req('https://user.com/reply'), self.as2_req('http://not/fediverse'), self.req('http://not/fediverse'), self.as2_req('https://mas.to/toot'), self.as2_req('https://mas.to/author'), )) args, kwargs = mock_post.call_args self.assertEqual(('https://mas.to/inbox',), args) self.assert_equals(AS2_CREATE, json_loads(kwargs['data'])) def test_repost(self, mock_get, mock_post): self._test_repost(REPOST_HTML, REPOST_AS2, mock_get, mock_post) def test_repost_composite_hcite(self, mock_get, mock_post): self._test_repost(REPOST_HCITE_HTML, REPOST_AS2, mock_get, mock_post) def _test_repost(self, html, expected_as2, mock_get, mock_post): self.make_followers() mock_get.side_effect = [ requests_response(html, url='https://user.com/repost'), TOOT_AS2, ACTOR, ] mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/repost', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) mock_get.assert_has_calls(( self.req('https://user.com/repost'), self.as2_req('https://mas.to/toot/id'), self.as2_req('https://mas.to/author'), )) inboxes = ('https://inbox/', 'https://public/inbox', 'https://shared/inbox', 'https://mas.to/inbox') self.assert_deliveries(mock_post, inboxes, expected_as2, ignore=['cc']) for args, kwargs in mock_get.call_args_list[1:]: with self.subTest(url=args[0]): rsa_key = kwargs['auth'].header_signer._rsa._key self.assertEqual(g.user.private_pem(), rsa_key.exportKey()) mf2 = util.parse_mf2(html)['items'][0] author_key = ndb.Key('ActivityPub', 'https://mas.to/author') self.assert_object('https://user.com/repost', users=[g.user.key], notify=[author_key], feed=self.followers, source_protocol='web', status='complete', mf2=mf2, delivered=inboxes, type='share', object_ids=['https://mas.to/toot/id'], labels=['user', 'activity', 'notification', 'feed'], ) def test_link_rel_alternate_as2(self, mock_get, mock_post): mock_get.side_effect = [ REPLY, NOT_FEDIVERSE, # AP NOT_FEDIVERSE, # Web TOOT_HTML, # AP TOOT_AS2, # AP via rel-alternate ACTOR, ] mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/reply', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) mock_get.assert_has_calls(( self.req('https://user.com/reply'), self.as2_req('http://not/fediverse'), self.req('http://not/fediverse'), self.as2_req('https://mas.to/toot'), self.as2_req('https://mas.to/toot/id', headers=as2.CONNEG_HEADERS), self.as2_req('https://mas.to/author'), )) args, kwargs = mock_post.call_args self.assertEqual(('https://mas.to/inbox',), args) self.assert_equals(AS2_CREATE, json_loads(kwargs['data'])) def test_like_stored_object_without_as2(self, mock_get, mock_post): Object(id='https://mas.to/toot', mf2=NOTE_MF2, source_protocol='ap').put() Object(id='https://user.com/', mf2=ACTOR_MF2).put() mock_get.side_effect = [ LIKE, ] got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/like', 'target': 'https://fed.brid.gy/', }) self.assertEqual(204, got.status_code) mock_get.assert_has_calls(( self.req('https://user.com/like'), )) mock_post.assert_not_called() self.assert_object('https://user.com/like', users=[g.user.key], source_protocol='web', mf2=LIKE_MF2, type='like', labels=['activity', 'user'], status='ignored', ) def test_post_type_discovery_multiple_types(self, mock_get, mock_post): self.make_followers() mock_get.return_value = requests_response( NOTE_HTML.replace('', """ """), url='https://user.com/post') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/multiple', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) inboxes = ['https://inbox/', 'https://public/inbox', 'https://shared/inbox'] self.assert_deliveries(mock_post, inboxes, { **NOTE_AS2, 'attributedTo': None, 'type': 'Create', 'actor': 'http://localhost/user.com', # TODO: this is an awkward wart left over from the multi-type mf2. # remove it eventually. 'object': { 'targetUrl': 'http://bob.com/post', 'to': ['https://www.w3.org/ns/activitystreams#Public'], }, }) def test_create_default_url_to_wm_source(self, mock_get, mock_post): """Source post has no u-url. AS2 id should default to webmention source.""" missing_url = requests_response("""\ reposted! Ms. ☕ Baz """, url='https://user.com/repost') mock_get.side_effect = [missing_url, TOOT_AS2, ACTOR] mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/repost', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) args, kwargs = mock_post.call_args self.assertEqual(('https://mas.to/inbox',), args) self.assert_equals(REPOST_AS2, json_loads(kwargs['data'])) def test_create_author_only_url(self, mock_get, mock_post): """Mf2 author property is just a URL. We should run full authorship. https://indieweb.org/authorship """ repost = requests_response("""\ reposted! """, url='https://user.com/repost') mock_get.side_effect = [repost, ACTOR, TOOT_AS2, ACTOR] mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/repost', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) args, kwargs = mock_post.call_args self.assertEqual(('https://mas.to/inbox',), args) self.assert_equals(REPOST_AS2, json_loads(kwargs['data'])) def test_create_no_author(self, mock_get, mock_post): """No mf2 author. We should default to the user's homepage.""" mock_get.side_effect = [ requests_response("""\ reposted! """, url='https://user.com/repost'), NOT_FEDIVERSE, TOOT_AS2, ACTOR, ] mock_post.return_value = requests_response('abc xyz') got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/repost', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) repost_mf2 = copy.deepcopy(REPOST_MF2) repost_mf2['properties']['author'] = ['https://user.com/'] self.assert_object('https://user.com/repost', users=[g.user.key], source_protocol='web', mf2=repost_mf2, # includes author https://user.com/ type='share', labels=['activity', 'user'], notify=[ndb.Key('ActivityPub', 'https://mas.to/author')], delivered=['https://mas.to/inbox'], status='complete', ) def test_create_non_domain_author(self, mock_get, mock_post): """Author is a page on the user's domain.""" self.make_followers() mock_get.side_effect = [ requests_response(NOTE_HTML.replace( '', '' ), url='https://user.com/post'), ACTOR_HTML_RESP, TOOT_AS2, ACTOR, ] got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/post', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) inboxes = ['https://inbox/', 'https://public/inbox', 'https://shared/inbox'] self.assert_object('https://user.com/post#bridgy-fed-create', users=[g.user.key], source_protocol='web', our_as1=CREATE_AS1, type='post', labels=['activity', 'user'], delivered=inboxes, status='complete', ) def test_create_post(self, mock_get, mock_post): mock_get.side_effect = [NOTE, ACTOR] mock_post.return_value = requests_response('abc xyz') self.make_followers() got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/post', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) mock_get.assert_has_calls(( self.req('https://user.com/post'), )) inboxes = ('https://inbox/', 'https://public/inbox', 'https://shared/inbox') self.assert_deliveries(mock_post, inboxes, CREATE_AS2) self.assert_object('https://user.com/post', our_as1=NOTE_AS1, feed=self.followers, type='note', source_protocol='web', ) self.assert_object('https://user.com/post#bridgy-fed-create', users=[g.user.key], source_protocol='web', status='complete', our_as1=CREATE_AS1, delivered=inboxes, type='post', ) def test_update_post(self, mock_get, mock_post): mock_get.side_effect = [NOTE, ACTOR] mock_post.return_value = requests_response('abc xyz') mf2 = copy.deepcopy(NOTE_MF2) mf2['properties']['content'] = 'different' Object(id='https://user.com/post', users=[g.user.key], mf2=mf2).put() self.make_followers() got = self.client.post('/_ah/queue/webmention', data={ 'source': 'https://user.com/post', 'target': 'https://fed.brid.gy/', }) self.assertEqual(200, got.status_code) mock_get.assert_has_calls(( self.req('https://user.com/post'), )) inboxes = ('https://inbox/', 'https://public/inbox', 'https://shared/inbox') self.assert_deliveries(mock_post, inboxes, UPDATE_AS2) update_as1 = { 'objectType': 'activity', 'verb': 'update', 'id': 'https://user.com/post#bridgy-fed-update-2022-01-02T03:04:05+00:00', 'actor': ACTOR_AS1_UNWRAPPED, 'object': { **NOTE_AS1, 'updated': '2022-01-02T03:04:05+00:00', }, } self.assert_object( f'https://user.com/post#bridgy-fed-update-2022-01-02T03:04:05+00:00', users=[g.user.key], source_protocol='web', status='complete', our_as1=update_as1, delivered=inboxes, type='update', labels=['user', 'activity'], ) def test_create_with_image(self, mock_get, mock_post): create_html = NOTE_HTML.replace( '', '\n