# coding=utf-8 """Unit tests for webmention.py. TODO: test error handling """ import copy from unittest import mock from urllib.parse import urlencode from django_salmon import magicsigs, utils import feedparser from granary import atom, microformats2 from httpsig.sign import HeaderSigner from oauth_dropins.webutil import util from oauth_dropins.webutil.testutil import requests_response from oauth_dropins.webutil.util import json_dumps, json_loads import requests import activitypub from app import application from common import ( AS2_PUBLIC_AUDIENCE, CONNEG_HEADERS_AS2, CONNEG_HEADERS_AS2_HTML, CONTENT_TYPE_AS2, CONTENT_TYPE_ATOM, CONTENT_TYPE_HTML, CONTENT_TYPE_MAGIC_ENVELOPE, HEADERS, ) from models import Follower, MagicKey, Response import webmention from . import testutil REPOST_HTML = """\ reposted! Ms. ☕ Baz """ REPOST_AS2 = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Announce', 'id': 'http://localhost/r/http://a/repost', 'url': 'http://localhost/r/http://a/repost', 'name': 'reposted!', 'object': 'tag:orig,2017:as2', 'cc': [ AS2_PUBLIC_AUDIENCE, 'http://orig/author', 'http://orig/recipient', 'http://orig/bystander', ], 'actor': { 'type': 'Person', 'id': 'http://localhost/orig', 'url': 'http://localhost/r/http://orig', 'name': 'Ms. ☕ Baz', 'preferredUsername': 'orig', }, } @mock.patch('requests.post') @mock.patch('requests.get') class WebmentionTest(testutil.TestCase): def setUp(self): super(WebmentionTest, self).setUp() self.key = MagicKey.get_or_create('a') self.orig_html_as2 = requests_response("""\ """, url='http://orig/post', content_type=CONTENT_TYPE_HTML) self.orig_html_atom = requests_response("""\ """, url='http://orig/post', content_type=CONTENT_TYPE_HTML) self.orig_atom = requests_response("""\ tag:fed.brid.gy,2017-08-22:orig-post baz ☕ baj """, content_type=CONTENT_TYPE_ATOM) self.orig_as2_data = { '@context': ['https://www.w3.org/ns/activitystreams'], 'type': 'Article', 'id': 'tag:orig,2017:as2', 'content': 'Lots of ☕ words...', 'actor': {'url': 'http://orig/author'}, 'to': ['http://orig/recipient'], 'cc': ['http://orig/bystander', AS2_PUBLIC_AUDIENCE], } self.orig_as2 = requests_response( self.orig_as2_data, url='http://orig/as2', content_type=CONTENT_TYPE_AS2 + '; charset=utf-8') self.reply_html = """\

""" self.reply = requests_response( self.reply_html, content_type=CONTENT_TYPE_HTML) self.reply_mf2 = util.parse_mf2(self.reply_html, url='http://a/reply') self.repost_html = REPOST_HTML self.repost = requests_response( self.repost_html, content_type=CONTENT_TYPE_HTML) self.repost_mf2 = util.parse_mf2(self.repost_html, url='http://a/repost') self.repost_as2 = REPOST_AS2 self.like_html = """\ Ms. ☕ Baz """ self.like = requests_response( self.like_html, content_type=CONTENT_TYPE_HTML) self.like_mf2 = util.parse_mf2(self.like_html, url='http://a/like') self.actor = requests_response({ 'objectType' : 'person', 'displayName': 'Mrs. ☕ Foo', 'url': 'https://foo.com/about-me', 'inbox': 'https://foo.com/inbox', }, content_type=CONTENT_TYPE_AS2) self.as2_create = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Create', 'object': { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Note', 'id': 'http://localhost/r/http://a/reply', 'url': 'http://localhost/r/http://a/reply', 'name': 'foo ☕ bar', 'content': """\ foo ☕ bar """, 'inReplyTo': 'tag:orig,2017:as2', 'cc': [ AS2_PUBLIC_AUDIENCE, 'http://orig/author', 'http://orig/recipient', 'http://orig/bystander', ], 'attributedTo': [{ 'type': 'Person', 'id': 'http://localhost/orig', 'url': 'http://localhost/r/http://orig', 'preferredUsername': 'orig', 'name': 'Ms. ☕ Baz', }], 'tag': [{ 'type': 'Mention', 'href': 'http://orig/author', }], }, } self.as2_update = copy.deepcopy(self.as2_create) self.as2_update['type'] = 'Update' self.follow_html = """\ Ms. ☕ Baz """ self.follow = requests_response( self.follow_html, content_type=CONTENT_TYPE_HTML) self.follow_mf2 = util.parse_mf2(self.follow_html, url='http://a/follow') self.follow_as2 = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Follow', 'id': 'http://localhost/r/http://a/follow', 'url': 'http://localhost/r/http://a/follow', 'object': 'http://followee', 'actor': { 'id': 'http://localhost/orig', 'name': 'Ms. ☕ Baz', 'preferredUsername': 'orig', 'type': 'Person', 'url': 'http://localhost/r/https://orig', }, 'cc': ['https://www.w3.org/ns/activitystreams#Public'], } self.create_html = """\

hello i am a post

Ms. ☕ Baz """ self.create = requests_response( self.create_html, content_type=CONTENT_TYPE_HTML) self.create_mf2 = util.parse_mf2(self.create_html, url='http://a/create') self.create_as2 = { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Create', 'object': { '@context': 'https://www.w3.org/ns/activitystreams', 'type': 'Note', 'id': 'http://localhost/r/http://orig/post', 'url': 'http://localhost/r/http://orig/post', 'name': 'hello i am a post', 'content': 'hello i am a post', 'attributedTo': [{ 'type': 'Person', 'id': 'http://localhost/orig', 'url': 'http://localhost/r/https://orig', 'name': 'Ms. ☕ Baz', 'preferredUsername': 'orig', }], 'cc': ['https://www.w3.org/ns/activitystreams#Public'], }, } self.not_fediverse = requests_response("""\ foo """, url='http://not/fediverse', content_type=CONTENT_TYPE_HTML) self.activitypub_gets = [self.reply, self.not_fediverse, self.orig_as2, self.actor] def verify_salmon(self, mock_post): args, kwargs = mock_post.call_args self.assertEqual(('http://orig/salmon',), args) self.assertEqual(CONTENT_TYPE_MAGIC_ENVELOPE, kwargs['headers']['Content-Type']) env = utils.parse_magic_envelope(kwargs['data']) assert magicsigs.verify(None, env['data'], env['sig'].encode(), key=self.key) return env['data'] def test_bad_source_url(self, mock_get, mock_post): got = application.get_response('/webmention', method='POST', body=b'') self.assertEqual(400, got.status_int) mock_get.side_effect = ValueError('foo bar') got = application.get_response('/webmention', method='POST', body=urlencode({'source': 'bad'}).encode()) self.assertEqual(400, got.status_int) def test_no_source_entry(self, mock_get, mock_post): mock_get.return_value = requests_response("""

nothing to see here except link

no one to send to!

""", content_type=CONTENT_TYPE_HTML) got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/post', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_has_calls((self.req('http://a/post'),)) def test_bad_target_url(self, mock_get, mock_post): mock_get.side_effect = ( requests_response(self.reply_html.replace('http://orig/post', 'bad'), content_type=CONTENT_TYPE_HTML), ValueError('foo bar')) got = application.get_response( '/webmention', method='POST', body=urlencode({'source': 'http://a/post'}).encode()) self.assertEqual(400, got.status_int) def test_target_fetch_fails(self, mock_get, mock_post): mock_get.side_effect = ( requests_response(self.reply_html.replace('http://orig/post', 'bad'), content_type=CONTENT_TYPE_HTML), requests.Timeout('foo bar')) got = application.get_response( '/webmention', method='POST', body=urlencode({'source': 'http://a/post'}).encode()) self.assertEqual(502, got.status_int) def test_target_fetch_has_no_content_type(self, mock_get, mock_post): mock_get.side_effect = ( requests_response(self.reply_html), requests_response(self.reply_html, content_type='None') ) got = application.get_response( '/webmention', method='POST', body=urlencode({'source': 'http://a/post'}).encode()) self.assertEqual(502, got.status_int) def test_no_backlink(self, mock_get, mock_post): mock_get.return_value = requests_response( self.reply_html.replace('', ''), content_type=CONTENT_TYPE_HTML) got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/post', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(400, got.status_int) mock_get.assert_has_calls((self.req('http://a/post'),)) def test_activitypub_create_reply(self, mock_get, mock_post): mock_get.side_effect = self.activitypub_gets mock_post.return_value = requests_response('abc xyz', status=203) got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(203, got.status_int) mock_get.assert_has_calls(( self.req('http://a/reply'), self.req('http://not/fediverse', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/post', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/author', headers=CONNEG_HEADERS_AS2_HTML), )) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) self.assertEqual(self.as2_create, kwargs['json']) headers = kwargs['headers'] self.assertEqual(CONTENT_TYPE_AS2, headers['Content-Type']) rsa_key = kwargs['auth'].header_signer._rsa._key self.assertEqual(self.key.private_pem(), rsa_key.exportKey()) resp = Response.get_by_id('http://a/reply http://orig/as2') self.assertEqual('out', resp.direction) self.assertEqual('activitypub', resp.protocol) self.assertEqual('complete', resp.status) self.assertEqual(self.reply_mf2, json_loads(resp.source_mf2)) # TODO: if i do this, maybe switch to separate HttpRequest model and # foreign key # self.assertEqual([self.as2_create], resp.request_statuses) # self.assertEqual([self.as2_create], resp.requests) # self.assertEqual(['abc xyz'], resp.responses) def test_activitypub_update_reply(self, mock_get, mock_post): Response(id='http://a/reply http://orig/as2', status='complete').put() mock_get.side_effect = self.activitypub_gets mock_post.return_value = requests_response('abc xyz') got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(200, got.status_int) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) self.assertEqual(self.as2_update, kwargs['json']) def test_activitypub_create_reply_attributed_to_id_only(self, mock_get, mock_post): """Based on PeerTube's AS2. https://github.com/snarfed/bridgy-fed/issues/40 """ del self.orig_as2_data['actor'] self.orig_as2_data['attributedTo'] = [{ 'type': 'Person', 'id': 'http://orig/author', }] orig_as2_resp = requests_response( self.orig_as2_data, content_type=CONTENT_TYPE_AS2 + '; charset=utf-8') mock_get.side_effect = [self.reply, self.not_fediverse, orig_as2_resp, self.actor] mock_post.return_value = requests_response('abc xyz', status=203) got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(203, got.status_int) mock_get.assert_has_calls(( self.req('http://a/reply'), self.req('http://not/fediverse', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/post', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/author', headers=CONNEG_HEADERS_AS2_HTML), )) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) self.assertEqual(self.as2_create, kwargs['json']) def test_activitypub_update_reply(self, mock_get, mock_post): Response(id='http://a/reply http://orig/as2', status='complete').put() mock_get.side_effect = self.activitypub_gets mock_post.return_value = requests_response('abc xyz') got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(200, got.status_int) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) self.assertEqual(self.as2_update, kwargs['json']) def test_activitypub_create_repost(self, mock_get, mock_post): mock_get.side_effect = [self.repost, self.orig_as2, self.actor] mock_post.return_value = requests_response('abc xyz') got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/repost', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_has_calls(( self.req('http://a/repost'), self.req('http://orig/post', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/author', headers=CONNEG_HEADERS_AS2_HTML), )) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) self.assertEqual(self.repost_as2, kwargs['json']) headers = kwargs['headers'] self.assertEqual(CONTENT_TYPE_AS2, headers['Content-Type']) rsa_key = kwargs['auth'].header_signer._rsa._key self.assertEqual(self.key.private_pem(), rsa_key.exportKey()) resp = Response.get_by_id('http://a/repost http://orig/as2') self.assertEqual('out', resp.direction) self.assertEqual('activitypub', resp.protocol) self.assertEqual('complete', resp.status) self.assertEqual(self.repost_mf2, json_loads(resp.source_mf2)) def test_activitypub_link_rel_alternate_as2(self, mock_get, mock_post): mock_get.side_effect = [self.reply, self.not_fediverse, self.orig_html_as2, self.orig_as2, self.actor] mock_post.return_value = requests_response('abc xyz') got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_has_calls(( self.req('http://a/reply'), self.req('http://not/fediverse', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/post', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/as2', headers=CONNEG_HEADERS_AS2), self.req('http://orig/author', headers=CONNEG_HEADERS_AS2_HTML), )) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) self.assertEqual(self.as2_create, kwargs['json']) def test_activitypub_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 """, content_type=CONTENT_TYPE_HTML) mock_get.side_effect = [missing_url, self.orig_as2, self.actor] mock_post.return_value = requests_response('abc xyz', status=203) got = application.get_response('/webmention', method='POST', body=urlencode({ 'source': 'http://a/repost', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(203, got.status_int) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) self.assert_equals(self.repost_as2, kwargs['json']) def test_activitypub_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! """, content_type=CONTENT_TYPE_HTML) author = requests_response("""\ Ms. ☕ Baz

""", url='http://orig', content_type=CONTENT_TYPE_HTML) mock_get.side_effect = [repost, author, self.orig_as2, self.actor] mock_post.return_value = requests_response('abc xyz', status=201) got = application.get_response('/webmention', method='POST', body=urlencode({ 'source': 'http://a/repost', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(201, got.status_int) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) repost_as2 = copy.deepcopy(self.repost_as2) repost_as2['actor']['image'] = repost_as2['actor']['icon'] = \ {'type': 'Image', 'url': 'http://orig/pic'}, self.assert_equals(repost_as2, kwargs['json']) def test_activitypub_create_post(self, mock_get, mock_post): mock_get.side_effect = [self.create, self.actor] mock_post.return_value = requests_response('abc xyz') Follower.get_or_create('orig', 'https://mastodon/aaa') Follower.get_or_create('orig', 'https://mastodon/bbb', last_follow=json_dumps({'actor': { 'publicInbox': 'https://public/inbox', 'inbox': 'https://unused', }})) Follower.get_or_create('orig', 'https://mastodon/ccc', last_follow=json_dumps({'actor': { 'endpoints': { 'sharedInbox': 'https://shared/inbox', }, }})) Follower.get_or_create('orig', 'https://mastodon/ddd', last_follow=json_dumps({'actor': { 'inbox': 'https://inbox', }})) Follower.get_or_create('orig', 'https://mastodon/eee', status='inactive', last_follow=json_dumps({'actor': { 'inbox': 'https://unused/2', }})) got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://orig/post', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_has_calls(( self.req('http://orig/post'), )) inboxes = ('https://public/inbox', 'https://shared/inbox', 'https://inbox') self.assertEqual(len(inboxes), len(mock_post.call_args_list)) for call, inbox in zip(mock_post.call_args_list, inboxes): self.assertEqual((inbox,), call[0]) self.assertEqual(self.create_as2, call[1]['json']) for inbox in inboxes: resp = Response.get_by_id('http://orig/post %s' % inbox) self.assertEqual('out', resp.direction, inbox) self.assertEqual('activitypub', resp.protocol, inbox) self.assertEqual('complete', resp.status, inbox) self.assertEqual(self.create_mf2, json_loads(resp.source_mf2), inbox) def test_activitypub_create_with_image(self, mock_get, mock_post): create_html = self.create_html.replace( '', '

\n') mock_get.side_effect = [ requests_response(create_html, content_type=CONTENT_TYPE_HTML), self.actor, ] mock_post.return_value = requests_response('abc xyz ') Follower.get_or_create( 'orig', 'https://mastodon/aaa', last_follow=json_dumps({'actor': {'inbox': 'https://inbox'}})) got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://orig/post', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(200, got.status_int) self.assertEqual(('https://inbox',), mock_post.call_args[0]) create = copy.deepcopy(self.create_as2) create['object'].update({ 'image': [{'url': 'http://im/age', 'type': 'Image'}], 'attachment': [{'url': 'http://im/age', 'type': 'Image'}], }) self.assertEqual(create, mock_post.call_args[1]['json']) def test_activitypub_follow(self, mock_get, mock_post): mock_get.side_effect = [self.follow, self.actor] mock_post.return_value = requests_response('abc xyz') got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/follow', 'target': 'https://fed.brid.gy/', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_has_calls(( self.req('http://a/follow'), self.req('http://followee/', headers=CONNEG_HEADERS_AS2_HTML), )) args, kwargs = mock_post.call_args self.assertEqual(('https://foo.com/inbox',), args) self.assertEqual(self.follow_as2, kwargs['json']) headers = kwargs['headers'] self.assertEqual(CONTENT_TYPE_AS2, headers['Content-Type']) rsa_key = kwargs['auth'].header_signer._rsa._key self.assertEqual(self.key.private_pem(), rsa_key.exportKey()) resp = Response.get_by_id('http://a/follow http://followee/') self.assertEqual('out', resp.direction) self.assertEqual('activitypub', resp.protocol) self.assertEqual('complete', resp.status) self.assertEqual(self.follow_mf2, json_loads(resp.source_mf2)) def test_salmon_reply(self, mock_get, mock_post): mock_get.side_effect = [self.reply, self.not_fediverse, self.orig_html_atom, self.orig_atom] got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'http://orig/post', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_has_calls(( self.req('http://a/reply'), self.req('http://not/fediverse', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/post', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/atom'), )) data = self.verify_salmon(mock_post) parsed = feedparser.parse(data) entry = parsed.entries[0] self.assertEqual('http://a/reply', entry['id']) self.assertIn({ 'rel': 'alternate', 'href': 'http://a/reply', 'type': 'text/html', }, entry['links']) self.assertEqual({ 'type': 'text/html', 'href': 'http://orig/post', 'ref': 'tag:fed.brid.gy,2017-08-22:orig-post', }, entry['thr_in-reply-to']) self.assertEqual("""\

foo ☕ bar

""", entry.content[0]['value']) resp = Response.get_by_id('http://a/reply http://orig/post') self.assertEqual('out', resp.direction) self.assertEqual('ostatus', resp.protocol) self.assertEqual('complete', resp.status) self.assertEqual(self.reply_mf2, json_loads(resp.source_mf2)) def test_salmon_like(self, mock_get, mock_post): mock_get.side_effect = [self.like, self.orig_html_atom, self.orig_atom] got = application.get_response( '/webmention', method='POST', body=urlencode({ 'source': 'http://a/like', 'target': 'http://orig/post', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_has_calls(( self.req('http://a/like'), self.req('http://orig/post', headers=CONNEG_HEADERS_AS2_HTML), self.req('http://orig/atom'), )) data = self.verify_salmon(mock_post) parsed = feedparser.parse(data) entry = parsed.entries[0] self.assertEqual('http://a/like', entry['id']) self.assertIn({ 'rel': 'alternate', 'href': 'http://a/like', 'type': 'text/html', }, entry['links']) self.assertEqual('http://orig/post', entry['activity_object']) resp = Response.get_by_id('http://a/like http://orig/post') self.assertEqual('out', resp.direction) self.assertEqual('ostatus', resp.protocol) self.assertEqual('complete', resp.status) self.assertEqual(self.like_mf2, json_loads(resp.source_mf2)) def test_salmon_get_salmon_from_webfinger(self, mock_get, mock_post): orig_atom = requests_response("""\ ryan ryan@orig tag:fed.brid.gy,2017-08-22:orig-post """) webfinger = requests_response({ 'subject': 'acct:ryan@orig', 'links': [{ 'rel': 'salmon', 'href': 'http://orig/@ryan/salmon', }], }) mock_get.side_effect = [self.reply, self.not_fediverse, self.orig_html_atom, orig_atom, webfinger] got = application.get_response('/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'http://orig/post', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_any_call( 'http://orig/.well-known/webfinger?resource=acct:ryan@orig', headers=HEADERS, stream=True, timeout=util.HTTP_TIMEOUT, verify=False) self.assertEqual(('http://orig/@ryan/salmon',), mock_post.call_args[0]) def test_salmon_no_target_atom(self, mock_get, mock_post): orig_no_atom = requests_response("""\ foo """, 'http://orig/url') mock_get.side_effect = [self.reply, self.not_fediverse, orig_no_atom] got = application.get_response('/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'http://orig/post', }).encode()) self.assertEqual(400, got.status_int) self.assertIn('Target post http://orig/url has no Atom link', got.body.decode()) self.assertIsNone(Response.get_by_id('http://a/reply http://orig/post')) def test_salmon_relative_atom_href(self, mock_get, mock_post): orig_relative = requests_response("""\ """, 'http://orig/url') mock_get.side_effect = [self.reply, self.not_fediverse, orig_relative, self.orig_atom] got = application.get_response('/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'http://orig/post', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_any_call('http://orig/atom/1', headers=HEADERS, stream=True, timeout=util.HTTP_TIMEOUT) data = self.verify_salmon(mock_post) def test_salmon_relative_atom_href_with_base(self, mock_get, mock_post): orig_base = requests_response("""\ """, 'http://orig/url') mock_get.side_effect = [self.reply, self.not_fediverse, orig_base, self.orig_atom] got = application.get_response('/webmention', method='POST', body=urlencode({ 'source': 'http://a/reply', 'target': 'http://orig/post', }).encode()) self.assertEqual(200, got.status_int) mock_get.assert_any_call('http://orig/base/atom/1', headers=HEADERS, stream=True, timeout=util.HTTP_TIMEOUT) data = self.verify_salmon(mock_post)