Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
AP users: add convert.py and /convert/... endpoint
- Loading branch information
Showing
3 changed files
with
100 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,49 @@ | ||
"""Serves /convert/... URLs to convert data from one protocol to another. | ||
URL pattern is /convert/SOURCE/DEST , where SOURCE and DEST are the LABEL | ||
constants from the :class:`Protocol` subclasses. | ||
Currently only supports /convert/activitypub/webmention/... | ||
""" | ||
import logging | ||
import re | ||
import urllib.parse | ||
|
||
from flask import request | ||
from oauth_dropins.webutil import flask_util, util | ||
from oauth_dropins.webutil.flask_util import error | ||
|
||
from activitypub import ActivityPub | ||
from common import CACHE_TIME | ||
from flask_app import app, cache | ||
from protocol import protocols | ||
from webmention import Webmention | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
SOURCES = frozenset(( | ||
ActivityPub.LABEL, | ||
)) | ||
DESTS = frozenset(( | ||
Webmention.LABEL, | ||
)) | ||
|
||
|
||
@app.get(f'/convert/<any({",".join(SOURCES)}):src>/<any({",".join(DESTS)}):dest>/<path:url>') | ||
@flask_util.cached(cache, CACHE_TIME, headers=['Accept']) | ||
def convert(src, dest, url): | ||
"""Converts data from one protocol to another and serves it. | ||
Fetches the source data if it's not already stored. | ||
""" | ||
if request.args: | ||
url += '?' + urllib.parse.urlencode(request.args) | ||
# some browsers collapse repeated /s in the path down to a single slash. | ||
# if that happened to this URL, expand it back to two /s. | ||
url = re.sub(r'^(https?:/)([^/])', r'\1/\2', url) | ||
|
||
if not util.is_web(url): | ||
error(f'Expected fully qualified URL; got {url}') | ||
|
||
obj = protocols[src].load(url) | ||
return protocols[dest].serve(obj) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,50 @@ | ||
"""Unit tests for convert.py. | ||
""" | ||
from unittest.mock import patch | ||
|
||
from oauth_dropins.webutil.testutil import requests_response | ||
import requests | ||
|
||
from common import CONTENT_TYPE_HTML | ||
|
||
from .test_redirect import ( | ||
REPOST_AS2, | ||
REPOST_HTML, | ||
) | ||
from . import testutil | ||
|
||
|
||
@patch('requests.get') | ||
class ConvertTest(testutil.TestCase): | ||
|
||
def test_unknown_source(self, _): | ||
got = self.client.get('/convert/nope/webmention/http://foo') | ||
self.assertEqual(404, got.status_code) | ||
|
||
def test_unknown_dest(self, _): | ||
got = self.client.get('/convert/activitypub/nope/http://foo') | ||
self.assertEqual(404, got.status_code) | ||
|
||
def test_missing_url(self, _): | ||
got = self.client.get('/convert/activitypub/webmention/') | ||
self.assertEqual(404, got.status_code) | ||
|
||
def test_url_not_web(self, _): | ||
got = self.client.get('/convert/activitypub/webmention/git+ssh://foo/bar') | ||
self.assertEqual(400, got.status_code) | ||
|
||
def test_activitypub_to_web(self, mock_get): | ||
mock_get.return_value = self.as2_resp(REPOST_AS2) | ||
|
||
got = self.client.get('/convert/activitypub/webmention/https://user.com/bar?baz=baj&biff') | ||
self.assertEqual(200, got.status_code) | ||
self.assertEqual(CONTENT_TYPE_HTML, got.content_type) | ||
|
||
mock_get.assert_has_calls((self.as2_req('https://user.com/bar?baz=baj&biff='),)) | ||
|
||
def test_activitypub_to_web_fetch_fails(self, mock_get): | ||
mock_get.side_effect = [requests_response('', status=405)] | ||
|
||
got = self.client.get('/convert/activitypub/webmention/http://foo') | ||
self.assertEqual(502, got.status_code) | ||
mock_get.assert_has_calls((self.as2_req('http://foo'),)) |