Basic logic to extract provider data from URLs

retribute_api/providers.py

+import re
+
+
+class Registry(object):
+    def __init__(self):
+        self._data = {}
+
+    def register(self, obj):
+        if not obj.id:
+            raise ValueError("{} has no id".format(obj.id))
+        self._data[obj.id] = obj()
+
+    def __iter__(self):
+        yield from self._data.items()
+
+
+registry = Registry()
+
+
+class Provider(object):
+    id = None
+    domains = []
+    username_regex = None
+
+
+class BasicUsernameProvider(Provider):
+    username_regex = r"^\/([\w\.]+)/?$"
+
+    def match_from_url(self, parsed_url):
+        if not self.domains and not self.username_regex:
+            return
+
+        if parsed_url.hostname not in self.domains:
+            return
+
+        compiled = re.compile(self.username_regex)
+        result = compiled.match(parsed_url.path)
+        if not result:
+            return
+        username = result.groups()[0]
+        return {"provider": self.id, "id": username}
+
+
+@registry.register
+class Patreon(BasicUsernameProvider):
+    id = "patreon"
+    domains = ["patreon.com"]
+
+
+@registry.register
+class Liberapay(BasicUsernameProvider):
+    id = "liberapay"
+    domains = ["liberapay.com"]
+
+
+@registry.register
+class KoFi(BasicUsernameProvider):
+    id = "ko-fi"
+    domains = ["ko-fi.com"]

retribute_api/search/means.py

+import urllib.parse
+
+from retribute_api import providers
+
+
+def extract_from_url(url):
+    parsed = urllib.parse.urlparse(url)
+    for id, provider in providers.registry:
+        result = provider.match_from_url(parsed)
+        if result:
+            return result
+    return None

retribute_api/search/webfinger.py

+from rest_framework import serializers
+
+
+async def lookup(name, session):
+    username, domain = name.split("@")
+    response = await session.get(
+        "https://{}/.well-known/webfinger".format(domain),
+        params={"resource": "acct:{}".format(name)},
+    )
+    response.raise_for_status()
+    return await response.json()
+
+
+class AccountLinkSerializer(serializers.Serializer):
+    rel = serializers.CharField()
+    href = serializers.URLField()
+    type = serializers.CharField()
+
+
+class WebfingerSerializer(serializers.Serializer):
+    links = serializers.ListField(child=AccountLinkSerializer(), min_length=1)
+
+
+def get_links(payload):
+    serializer = WebfingerSerializer(data=payload)
+    serializer.is_valid(raise_exception=True)
+    links = {}
+    for link in serializer.validated_data["links"]:
+        if link["rel"] != "self":
+            continue
+
+        if link["type"] == "application/activity+json":
+            links["activitypub"] = link["href"]
+
+    return links

setup.cfg

@@ -65,6 +65,7 @@ universal = 1
 [tool:pytest]
 testpaths = tests
 env =
+    DJANGO_SETTINGS_MODULE=config.settings.base
     SECRET_KEY=test
     EMAIL_CONFIG=consolemail://
     DEBUG_TOOLBAR_ENABLED=False

tests/conftest.py

+import pytest
+import aiohttp
+from aioresponses import aioresponses
+import asynctest
+
+
+pytest_plugins = "aiohttp.pytest_plugin"
+
+
+@pytest.fixture
+def responses():
+    with aioresponses() as m:
+        yield m
+
+
+@pytest.fixture
+async def session(loop):
+    async with aiohttp.ClientSession() as session:
+        yield session
+
+
+@pytest.fixture
+async def coroutine_mock():
+    return asynctest.CoroutineMock

tests/search/test_means.py

+import pytest
+from retribute_api.search import means
+
+
+@pytest.mark.parametrize(
+    "input, expected",
+    [
+        ("https://patreon.com/username", {"provider": "patreon", "id": "username"}),
+        ("https://liberapay.com/username", {"provider": "liberapay", "id": "username"}),
+        ("https://ko-fi.com/username", {"provider": "ko-fi", "id": "username"}),
+        ("https://patreon.com/username/nope", None),
+    ],
+)
+def test_extract_from_url(input, expected):
+    assert means.extract_from_url(input) == expected

tests/search/test_webfinger.py

+from retribute_api.search import webfinger
+
+
+async def test_wellknown_success(responses, session):
+    name = "user@domain.test"
+    webfinger_response = {
+        "subject": "acct:user@domain.test",
+        "aliases": ["https://domain.test/@user", "https://domain.test/users/user"],
+        "links": [
+            {
+                "rel": "self",
+                "type": "application/activity+json",
+                "href": "https://domain.test/users/user",
+            }
+        ],
+    }
+    responses.get(
+        "https://domain.test/.well-known/webfinger?resource=acct:{}".format(name),
+        payload=webfinger_response,
+    )
+
+    response = await webfinger.lookup(name, session)
+    assert response == webfinger_response
+
+
+def test_get_links(settings):
+    payload = {
+        "links": [
+            {
+                "rel": "self",
+                "type": "application/activity+json",
+                "href": "https://domain.test/users/user",
+            }
+        ]
+    }
+
+    expected = {"activitypub": "https://domain.test/users/user"}
+
+    assert webfinger.get_links(payload) == expected

tests/test_dummy.py

-def test_something():
-    assert 1 == 2