"""Unit tests for convert.py.
"""
import copy
from unittest.mock import patch
from granary import as2
from granary.tests.test_as1 import ACTOR, COMMENT, DELETE_OF_ID, UPDATE
from models import Object
from oauth_dropins.webutil.testutil import requests_response
from oauth_dropins.webutil.util import json_loads, parse_mf2
# import first so that Fake is defined before URL routes are registered
from . import testutil
from .testutil import Fake, OtherFake
from activitypub import ActivityPub
from common import CONTENT_TYPE_HTML
from web import Web
COMMENT_AS2 = {
**as2.from_as1(COMMENT),
'type': 'Note',
'id': 'https://web.brid.gy/r/https://fake.com/123456',
'url': 'https://web.brid.gy/r/https://fake.com/123456',
'name': 'A ☕ reply',
'contentMap': {'en': COMMENT['content']},
'inReplyTo': 'https://web.brid.gy/r/https://fake.com/123',
}
HTML = """\
tag:fake.com:123456
2012-12-05T00:58:26+00:00
fake.com/123456
A ☕ reply
"""
HTML_NO_ID = HTML.replace('tag:fake.com:123456 ', '')
AUTHOR_HTML = """\
tag:fake.com:123456
2012-12-05T00:58:26+00:00
Bob
fake.com/123456
A ☕ reply
"""
class ConvertTest(testutil.TestCase):
def test_unknown_source(self):
resp = self.client.get('/convert/web/http://foo',
base_url='https://nope.brid.gy/')
self.assertEqual(404, resp.status_code)
def test_unknown_dest(self):
resp = self.client.get('/convert/nope/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(404, resp.status_code)
def test_missing_url(self):
resp = self.client.get('/convert/web/',
base_url='https://ap.brid.gy/')
self.assertEqual(404, resp.status_code)
def test_fake_to_other(self):
self.store_object(id='fake:post', our_as1={'foo': 'bar'})
resp = self.client.get('/convert/other/fake:post',
base_url='https://fa.brid.gy/')
self.assertEqual(200, resp.status_code)
self.assertEqual(OtherFake.CONTENT_TYPE, resp.content_type)
self.assertEqual({
'id': 'other:o:fa:fake:post',
'foo': 'bar',
}, json_loads(resp.get_data()))
def test_fake_to_activitypub(self):
self.make_user('fake:alice', cls=Fake)
self.store_object(id='fake:post', our_as1={
'actor': 'fake:alice',
'foo': 'bar',
})
resp = self.client.get('/convert/ap/fake:post',
base_url='https://fa.brid.gy/')
self.assertEqual(200, resp.status_code)
self.assertEqual(ActivityPub.CONTENT_TYPE, resp.content_type)
self.assertEqual({
'@context': 'https://www.w3.org/ns/activitystreams',
'id': 'https://fa.brid.gy/convert/ap/fake:post',
'actor': 'https://fa.brid.gy/ap/fake:alice',
'foo': 'bar',
'to': ['https://www.w3.org/ns/activitystreams#Public'],
}, json_loads(resp.get_data()))
def test_activitypub_to_web_object(self):
url = 'https://user.com/bar?baz=baj&biff'
Object(id=url, our_as1=COMMENT).put()
resp = self.client.get('/convert/web/https://user.com/bar?baz=baj&biff',
base_url='https://ap.brid.gy/')
self.assertEqual(200, resp.status_code)
self.assert_multiline_equals(HTML, resp.get_data(as_text=True),
ignore_blanks=True)
def test_activitypub_to_web_object_empty(self):
Object(id='http://foo').put()
resp = self.client.get('/convert/web/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(404, resp.status_code)
@patch('requests.get')
def test_activitypub_to_web_fetch(self, mock_get):
mock_get.return_value = self.as2_resp(as2.from_as1(COMMENT))
url = 'https://user.com/bar?baz=baj&biff'
resp = self.client.get(f'/convert/web/{url}',
base_url='https://ap.brid.gy/')
self.assertEqual(200, resp.status_code, resp.get_data(as_text=True))
self.assertEqual(CONTENT_TYPE_HTML, resp.content_type)
self.assert_multiline_equals(HTML, resp.get_data(as_text=True),
ignore_blanks=True)
mock_get.assert_has_calls((self.as2_req(url),))
@patch('requests.get')
def test_activitypub_to_web_fetch_fails(self, mock_get):
mock_get.side_effect = [requests_response('', status=405)]
resp = self.client.get('/convert/web/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(502, resp.status_code)
mock_get.assert_has_calls((self.as2_req('http://foo'),))
def test_activitypub_to_web_with_author(self):
Object(id='http://fo/o', our_as1={**COMMENT, 'author': 'http://ba/r'},
source_protocol='activitypub').put()
Object(id='http://ba/r', our_as1=ACTOR,
source_protocol='activitypub').put()
resp = self.client.get('/convert/web/http://fo/o',
base_url='https://ap.brid.gy/')
self.assertEqual(200, resp.status_code)
self.assert_multiline_equals(AUTHOR_HTML, resp.get_data(as_text=True),
ignore_blanks=True)
def test_activitypub_to_web_no_url(self):
comment = copy.deepcopy(COMMENT)
del comment['url']
Object(id='http://foo', our_as1=comment).put()
resp = self.client.get('/convert/web/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(200, resp.status_code)
expected = HTML.replace(
'\nfake.com/123456 ', '')
self.assert_multiline_equals(expected, resp.get_data(as_text=True),
ignore_blanks=True)
def test_activitypub_to_web_deleted_object(self):
Object(id='http://foo', as2={'content': 'foo'}, deleted=True).put()
resp = self.client.get('/convert/web/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(410, resp.status_code)
def test_activitypub_to_web_delete_activity(self):
Object(id='http://foo', our_as1=DELETE_OF_ID).put()
resp = self.client.get('/convert/web/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(410, resp.status_code)
def test_activitypub_to_web_update_inner_obj_exists_redirect(self):
# UPDATE's object field is a full object
Object(id='http://foo', our_as1=UPDATE).put()
Object(id=UPDATE['object']['id'], as2={'content': 'foo'}).put()
resp = self.client.get('/convert/web/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(301, resp.status_code)
self.assertEqual(f'/convert/web/tag:fake.com:123456',
resp.headers['Location'])
def test_activitypub_to_web_delete_inner_obj_exists_redirect(self):
# DELETE_OF_ID's object field is a bare string id
Object(id='http://foo', our_as1=DELETE_OF_ID).put()
Object(id=DELETE_OF_ID['object'], as2={'content': 'foo'}).put()
resp = self.client.get('/convert/web/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(301, resp.status_code)
self.assertEqual(f'/convert/web/tag:fake.com:123456',
resp.headers['Location'])
def test_activitypub_to_web_update_no_inner_obj_serve_as_is(self):
# Update's object field is a full object
Object(id='http://foo', our_as1=UPDATE).put()
resp = self.client.get('/convert/web/http://foo',
base_url='https://ap.brid.gy/')
self.assertEqual(200, resp.status_code)
self.assert_multiline_in("""\
A ☕ reply
A ☕ reply
""", url='https://nope.com/')
mock_get.side_effect = [
# post protocol inference
requests_response(''),
hcard,
hcard,
hcard,
]
resp = self.client.get(f'/convert/ap/https://nope.com/post',
base_url='https://web.brid.gy/')
self.assertEqual(200, resp.status_code)
self.assert_equals({
'type': 'Note',
'id': 'https://web.brid.gy/r/https://nope.com/post',
'attributedTo': 'https://web.brid.gy/nope.com',
}, resp.json, ignore=['to'])
@patch('requests.get', return_value=requests_response(HTML_NO_ID))
def test_web_to_activitypub_url_decode(self, mock_get):
"""https://github.com/snarfed/bridgy-fed/issues/581"""
self.make_user('user.com', cls=Web)
self.store_object(id='http://user.com/a#b',
mf2=parse_mf2(HTML_NO_ID)['items'][0])
resp = self.client.get(f'/convert/ap/http://user.com/a%23b',
base_url='https://web.brid.gy/')
self.assertEqual(200, resp.status_code)
self.assert_equals(COMMENT_AS2, resp.json, ignore=['to'])
def test_fed_subdomain(self):
url = 'https://user.com/post'
Object(id=url, our_as1=COMMENT).put()
resp = self.client.get('/convert/web/https://user.com/post',
base_url='https://fed.brid.gy/')
self.assertEqual(200, resp.status_code)
self.assert_multiline_equals(HTML, resp.get_data(as_text=True),
ignore_blanks=True)
def test_missing_protocols(self):
resp = self.client.get('/convert/https:/user.com/post')
self.assertEqual(404, resp.status_code)
resp = self.client.get('/convert/https:/user.com/post',
base_url='https://fed.brid.gy/')
self.assertEqual(404, resp.status_code)