# coding=utf-8
"""Unit tests for webmention.py."""
import copy
from unittest import mock
from urllib.parse import urlencode
import feedparser
from granary import as2, atom, microformats2
from httpsig.sign import HeaderSigner
from oauth_dropins.webutil import appengine_config, util
from oauth_dropins.webutil.appengine_config import tasks_client
from oauth_dropins.webutil.appengine_info import APP_ID
from oauth_dropins.webutil.testutil import requests_response
from oauth_dropins.webutil.util import json_dumps, json_loads
import requests
import activitypub
from common import (
CONNEG_HEADERS_AS2_HTML,
CONTENT_TYPE_HTML,
redirect_unwrap,
)
from models import Follower, Object, Target, User
import webmention
from webmention import TASKS_LOCATION
from . import testutil
ACTOR_HTML = """\
Ms. ☕ Baz
"""
ACTOR_MF2 = {
'type': ['h-card'],
'properties': {
'url': ['https://orig'],
'name': ['Ms. ☕ Baz'],
},
}
ACTOR_AS1_UNWRAPPED = {
'objectType': 'person',
'displayName': 'Ms. ☕ Baz',
'url': 'https://orig',
'urls': [{'value': 'https://orig', 'displayName': 'Ms. ☕ Baz'}],
}
ACTOR_AS2 = {
'type': 'Person',
'id': 'http://localhost/orig',
'url': 'http://localhost/r/https://orig',
'name': 'Ms. ☕ Baz',
'preferredUsername': 'orig',
}
ACTOR_AS2_FULL = {
**ACTOR_AS2,
'@context': [
'https://www.w3.org/ns/activitystreams',
'https://w3id.org/security/v1',
],
'preferredUsername': 'orig',
'attachment': [{
'name': 'Ms. ☕ Baz',
'type': 'PropertyValue',
'value': 'orig',
}],
'inbox': 'http://localhost/orig/inbox',
'outbox': 'http://localhost/orig/outbox',
'following': 'http://localhost/orig/following',
'followers': 'http://localhost/orig/followers',
'endpoints': {
'sharedInbox': 'http://localhost/inbox',
},
}
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',
'to': [as2.PUBLIC_AUDIENCE],
'cc': [
'https://orig/author',
'https://orig/recipient',
as2.PUBLIC_AUDIENCE,
'https://orig/bystander',
],
'actor': ACTOR_AS2,
}
@mock.patch('requests.post')
@mock.patch('requests.get')
class WebmentionTest(testutil.TestCase):
def setUp(self):
super().setUp()
self.user_orig = User.get_or_create('orig')
self.user_a = User.get_or_create('a')
self.orig_html_as2 = requests_response("""\
""", url='https://orig/post', content_type=CONTENT_TYPE_HTML)
self.orig_html_atom = requests_response("""\
""", url='https://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=atom.CONTENT_TYPE)
self.orig_as2_data = {
'@context': ['https://www.w3.org/ns/activitystreams'],
'type': 'Article',
'id': 'tag:orig,2017:as2',
'content': 'Lots of ☕ words...',
'actor': {'url': 'https://orig/author'},
'to': ['https://orig/recipient', as2.PUBLIC_AUDIENCE],
'cc': ['https://orig/bystander', as2.PUBLIC_AUDIENCE],
}
self.orig_as2 = requests_response(
self.orig_as2_data, url='https://orig/as2',
content_type=as2.CONTENT_TYPE + '; 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.reply_as1 = microformats2.json_to_object(self.reply_mf2['items'][0])
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_as1 = microformats2.json_to_object(self.repost_mf2['items'][0])
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=as2.CONTENT_TYPE)
self.as2_create = {
'@context': 'https://www.w3.org/ns/activitystreams',
'type': 'Create',
'id': 'http://localhost/r/http://a/reply#bridgy-fed-create',
'actor': {
'id': 'http://localhost/a',
'url': 'http://localhost/r/https://a/',
'preferredUsername': 'a',
},
'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',
'to': [as2.PUBLIC_AUDIENCE],
'cc': [
'https://orig/author',
'https://orig/recipient',
as2.PUBLIC_AUDIENCE,
'https://orig/bystander',
],
'attributedTo': [ACTOR_AS2],
'tag': [{
'type': 'Mention',
'href': 'https://orig/author',
}],
},
}
self.as2_update = copy.deepcopy(self.as2_create)
self.as2_update['type'] = 'Update'
# we should generate this if it's not already in mf2 because Mastodon
# requires it for updates
self.as2_update['object']['updated'] = util.now().isoformat()
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_as1 = microformats2.json_to_object(self.follow_mf2['items'][0])
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': ACTOR_AS2,
'to': [as2.PUBLIC_AUDIENCE],
}
self.follow_fragment_html = """\
Ignored
Ms. ☕ Baz
"""
self.follow_fragment = requests_response(
self.follow_fragment_html, content_type=CONTENT_TYPE_HTML)
self.follow_fragment_mf2 = util.parse_mf2(
self.follow_fragment_html, url='http://a/follow', id='2')
self.follow_fragment_as1 = microformats2.json_to_object(
self.follow_fragment_mf2['items'][0])
self.follow_fragment_as2 = copy.deepcopy(self.follow_as2)
self.follow_fragment_as2.update({
'id': 'http://localhost/r/http://a/follow#2',
'url': 'http://localhost/r/http://a/follow#2',
})
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_as1 = microformats2.json_to_object(self.create_mf2['items'][0])
self.create_as2 = {
'@context': 'https://www.w3.org/ns/activitystreams',
'type': 'Create',
'id': 'http://localhost/r/https://orig/post#bridgy-fed-create',
'actor': {
'id': 'http://localhost/orig',
'url': 'http://localhost/r/https://orig/',
'preferredUsername': 'orig',
},
'object': {
'@context': 'https://www.w3.org/ns/activitystreams',
'type': 'Note',
'id': 'http://localhost/r/https://orig/post',
'url': 'http://localhost/r/https://orig/post',
'name': 'hello i am a post',
'content': 'hello i am a post',
'attributedTo': [ACTOR_AS2],
'to': [as2.PUBLIC_AUDIENCE],
},
}
self.update_as2 = copy.deepcopy(self.create_as2)
self.update_as2['type'] = 'Update'
self.update_as2['object']['updated'] = util.now().isoformat()
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]
self.author = requests_response(ACTOR_HTML, url='https://orig/',
content_type=CONTENT_TYPE_HTML)
def assert_deliveries(self, mock_post, inboxes, data):
self.assertEqual(len(inboxes), len(mock_post.call_args_list))
calls = {call[0][0]: call for call in mock_post.call_args_list}
for inbox in inboxes:
with self.subTest(inbox=inbox):
got = json_loads(calls[inbox][1]['data'])
got.get('object', {}).pop('publicKey', None)
self.assertEqual(data, got)
def test_bad_source_url(self, mock_get, mock_post):
got = self.client.post('/webmention', data=b'')
self.assertEqual(400, got.status_code)
mock_get.side_effect = ValueError('foo bar')
got = self.client.post('/webmention', data={'source': 'bad'})
self.assertEqual(400, got.status_code)
self.assertEqual(0, Object.query().count())
def test_source_fetch_fails(self, mock_get, mock_post):
mock_get.side_effect = (
requests_response(self.reply_html, status=405,
content_type=CONTENT_TYPE_HTML),
)
got = self.client.post('/webmention', data={'source': 'http://a/post'})
self.assertEqual(502, got.status_code)
self.assertEqual(0, Object.query().count())
def test_no_source_entry(self, mock_get, mock_post):
mock_get.return_value = requests_response("""
nothing to see here except link
""", content_type=CONTENT_TYPE_HTML)
got = self.client.post( '/webmention', data={
'source': 'http://a/post',
'target': 'https://fed.brid.gy/',
})
self.assertEqual(400, got.status_code)
self.assertEqual(0, Object.query().count())
mock_get.assert_has_calls((self.req('http://a/post'),))
def test_no_targets(self, mock_get, mock_post):
mock_get.return_value = requests_response("""
no one to send to!
""", content_type=CONTENT_TYPE_HTML)
got = self.client.post('/_ah/queue/webmention', data={
'source': 'http://a/post',
'target': 'https://fed.brid.gy/',
})
self.assertEqual(200, got.status_code)
self.assertEqual(0, Object.query().count())
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('https://orig/post', 'bad'),
content_type=CONTENT_TYPE_HTML),
ValueError('foo bar'),
)
got = self.client.post('/webmention', data={'source': 'http://a/post'})
self.assertEqual(400, got.status_code)
self.assertEqual(0, Object.query().count())
def test_target_fetch_fails(self, mock_get, mock_post):
mock_get.side_effect = (
requests_response(self.reply_html.replace('https://orig/post', 'bad'),
content_type=CONTENT_TYPE_HTML),
requests.Timeout('foo bar'))
got = self.client.post('/webmention', data={'source': 'http://a/post'})
self.assertEqual(502, got.status_code)
def test_target_fetch_has_no_content_type(self, mock_get, mock_post):
html = self.reply_html.replace(
' ',
"