org-social/org-social-relay
1
0
Fork 0
mirror of https://github.com/tanrax/org-social-relay synced 2026-07-25 02:34:58 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
org-social-relay/app/bridge/test_views.py
T
andros 349a5e1376 Add ActivityPub and RSS bridges serving virtual social.org feeds 
Expose external accounts as virtual Org Social feeds that any client
can follow with a plain #+FOLLOW: line:

- /bridge/activitypub/@{user}@{instance}/ bridges a Mastodon or any
  ActivityPub account (WebFinger, actor and paginated outbox; only
  public top-level notes, with CW, hashtags, language and attachments)
- /bridge/rss/?url={feed} bridges any RSS/Atom feed (entry title as
  *** sub-heading, converted body and link to the original article)

Registration is implicit on first GET. Bridged data is stored in the
existing Profile/Post tables so /profile/, /search/ and the rest of
the API work on bridged feeds. Active bridges are refreshed every 15
minutes; bridges unrequested for 90 days are cleaned up.

Remote HTML is converted to Org text, escaping headline-like lines so
external content cannot inject posts. Fetches enforce the Webmention
SSRF protections, a 10s timeout and a 5MB size cap. Bridged posts
never queue Webmentions nor publish notifications.
2026-07-17 15:26:41 +02:00

384 lines
14 KiB
Python
Raw Permalink Blame History

from datetime import timedelta
from unittest.mock import patch
from django.test import TestCase
from django.utils import timezone
from app.bridge.fetching import BridgeError
from app.bridge.models import (
BridgedActivityPubAccount,
BridgedRssFeed,
activitypub_feed_url,
rss_feed_url,
)
from app.feeds.models import Profile
AP_FEED_PATH = "/bridge/activitypub/@alice@m.example/"
RSS_SOURCE_URL = "https://blog.example/feed.xml"
def _ap_data(posts=None):
return {
"actor_url": "https://m.example/users/alice",
"outbox_url": "https://m.example/users/alice/outbox",
"metadata": {
"title": "Alice",
"nick": "alice",
"description": "Bio",
"avatar": "https://m.example/avatar.png",
"link": "https://m.example/@alice",
},
"posts": posts
if posts is not None
else [
{
"id": "2025-05-01T10:00:00+00:00",
"content": "Hello from the fediverse",
"tags": "",
"language": "",
}
],
}
def _rss_data():
return {
"metadata": {
"title": "My Blog",
"nick": "My_Blog",
"description": "Notes",
"avatar": "",
"link": "https://blog.example",
"language": "en",
},
"posts": [
{
"id": "2025-05-01T10:00:00+00:00",
"content": "*** First article\n\nBody\n\n[[https://blog.example/first]]",
"tags": "",
"language": "",
}
],
}
class ActivityPubBridgeFeedViewTest(TestCase):
"""Test cases for the ActivityPub virtual feed endpoint."""
def test_first_get_registers_account_and_serves_org_file(self):
# Given: A reachable ActivityPub account not yet bridged
with patch(
"app.bridge.activitypub.fetch_account", return_value=_ap_data()
) as mock_fetch:
# When: The virtual feed is requested for the first time
response = self.client.get(AP_FEED_PATH)
# Then: The org file is served as plain text
self.assertEqual(response.status_code, 200)
self.assertTrue(response["Content-Type"].startswith("text/plain"))
content = response.content.decode()
self.assertIn("#+TITLE: Alice", content)
self.assertIn("#+NICK: alice", content)
self.assertIn("** 2025-05-01T10:00:00+00:00", content)
self.assertIn("Hello from the fediverse", content)
# And: The bridge and its profile are stored
mock_fetch.assert_called_once_with("alice", "m.example")
bridge = BridgedActivityPubAccount.objects.get(handle="alice@m.example")
self.assertEqual(bridge.profile.feed, activitypub_feed_url("alice@m.example"))
def test_second_get_is_served_from_database_without_network(self):
# Given: An already bridged account
with patch(
"app.bridge.activitypub.fetch_account", return_value=_ap_data()
) as mock_fetch:
self.client.get(AP_FEED_PATH)
# When: The feed is requested again shortly after
response = self.client.get(AP_FEED_PATH)
# Then: The response is correct and no second fetch happened
self.assertEqual(response.status_code, 200)
self.assertIn("Hello from the fediverse", response.content.decode())
self.assertEqual(mock_fetch.call_count, 1)
def test_non_canonical_handles_redirect_to_canonical_url(self):
# Given: Handle spellings with capitals or without the @ prefix
for path in (
"/bridge/activitypub/@Alice@M.Example/",
"/bridge/activitypub/alice@m.example/",
):
# When: The non-canonical URL is requested
response = self.client.get(path)
# Then: A permanent redirect points to the canonical URL
self.assertEqual(response.status_code, 301, path)
self.assertEqual(response["Location"], AP_FEED_PATH)
def test_invalid_handle_returns_400(self):
# Given: A handle without an instance part
# When: The feed is requested
response = self.client.get("/bridge/activitypub/@alice/")
# Then: The request is rejected
self.assertEqual(response.status_code, 400)
self.assertEqual(response.json()["type"], "Error")
def test_unreachable_account_returns_502(self):
# Given: An account whose instance cannot be reached
with patch(
"app.bridge.activitypub.fetch_account",
side_effect=BridgeError("connection refused"),
):
# When: The feed is requested
response = self.client.get(AP_FEED_PATH)
# Then: A bad gateway error with the envelope format is returned
self.assertEqual(response.status_code, 502)
body = response.json()
self.assertEqual(body["type"], "Error")
self.assertIsNone(body["data"])
# And: Nothing was stored
self.assertEqual(BridgedActivityPubAccount.objects.count(), 0)
def test_stale_bridge_is_refreshed_inline_on_access(self):
# Given: A bridged account whose data is older than a day
with patch("app.bridge.activitypub.fetch_account", return_value=_ap_data()):
self.client.get(AP_FEED_PATH)
bridge = BridgedActivityPubAccount.objects.get(handle="alice@m.example")
bridge.last_refreshed_at = timezone.now() - timedelta(days=2)
bridge.save()
new_posts = _ap_data(
posts=[
{
"id": "2025-05-05T10:00:00+00:00",
"content": "Fresh post",
"tags": "",
"language": "",
}
]
)
# When: The feed is requested again
with patch(
"app.bridge.activitypub.fetch_account", return_value=new_posts
) as mock_fetch:
response = self.client.get(AP_FEED_PATH)
# Then: The origin was fetched again and the new post is served
mock_fetch.assert_called_once()
self.assertIn("Fresh post", response.content.decode())
bridge.refresh_from_db()
self.assertGreater(
bridge.last_refreshed_at, timezone.now() - timedelta(minutes=1)
)
def test_failed_inline_refresh_serves_stored_copy(self):
# Given: A stale bridge whose origin is now unreachable
with patch("app.bridge.activitypub.fetch_account", return_value=_ap_data()):
self.client.get(AP_FEED_PATH)
bridge = BridgedActivityPubAccount.objects.get(handle="alice@m.example")
bridge.last_refreshed_at = timezone.now() - timedelta(days=2)
bridge.save()
# When: The feed is requested while the origin fails
with patch(
"app.bridge.activitypub.fetch_account",
side_effect=BridgeError("down"),
):
response = self.client.get(AP_FEED_PATH)
# Then: The stored copy is served anyway
self.assertEqual(response.status_code, 200)
self.assertIn("Hello from the fediverse", response.content.decode())
def test_bridged_profile_is_available_through_profile_endpoint(self):
# Given: A bridged account
with patch("app.bridge.activitypub.fetch_account", return_value=_ap_data()):
self.client.get(AP_FEED_PATH)
# When: The existing /profile/ endpoint is queried with the bridge URL
feed_url = activitypub_feed_url("alice@m.example")
response = self.client.get("/profile/", {"feed": feed_url})
# Then: The bridged profile is returned like any other profile
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["type"], "Success")
class ActivityPubBridgeListViewTest(TestCase):
"""Test cases for the bridged accounts listing."""
def test_lists_bridged_accounts_with_feed_urls(self):
# Given: A bridged account
profile = Profile.objects.create(
feed=activitypub_feed_url("alice@m.example"),
title="Alice",
nick="alice",
version="v1",
)
BridgedActivityPubAccount.objects.create(
handle="alice@m.example",
actor_url="https://m.example/users/alice",
profile=profile,
last_accessed_at=timezone.now(),
)
# When: The list is requested
response = self.client.get("/bridge/activitypub/")
# Then: The account and its virtual feed URL are listed
self.assertEqual(response.status_code, 200)
body = response.json()
self.assertEqual(body["type"], "Success")
self.assertEqual(
body["data"],
[
{
"handle": "@alice@m.example",
"feed": activitypub_feed_url("alice@m.example"),
}
],
)
class RssBridgeViewTest(TestCase):
"""Test cases for the RSS bridge endpoint."""
def test_first_get_registers_feed_and_serves_org_file(self):
# Given: A reachable RSS feed not yet bridged
with patch(
"app.bridge.rss.fetch_rss_feed", return_value=_rss_data()
) as mock_fetch:
# When: The virtual feed is requested for the first time
response = self.client.get("/bridge/rss/", {"url": RSS_SOURCE_URL})
# Then: The org file is served as plain text
self.assertEqual(response.status_code, 200)
self.assertTrue(response["Content-Type"].startswith("text/plain"))
content = response.content.decode()
self.assertIn("#+TITLE: My Blog", content)
self.assertIn("#+NICK: My_Blog", content)
self.assertIn("*** First article", content)
# And: The bridge is stored
mock_fetch.assert_called_once_with(RSS_SOURCE_URL)
self.assertTrue(
BridgedRssFeed.objects.filter(source_url=RSS_SOURCE_URL).exists()
)
def test_second_get_is_served_from_database_without_network(self):
# Given: An already bridged RSS feed
with patch(
"app.bridge.rss.fetch_rss_feed", return_value=_rss_data()
) as mock_fetch:
self.client.get("/bridge/rss/", {"url": RSS_SOURCE_URL})
# When: The feed is requested again shortly after
response = self.client.get("/bridge/rss/", {"url": RSS_SOURCE_URL})
# Then: The response is correct and no second fetch happened
self.assertEqual(response.status_code, 200)
self.assertEqual(mock_fetch.call_count, 1)
def test_without_url_parameter_lists_bridged_feeds(self):
# Given: A bridged RSS feed
profile = Profile.objects.create(
feed=rss_feed_url(RSS_SOURCE_URL),
title="My Blog",
nick="My_Blog",
version="v1",
)
BridgedRssFeed.objects.create(
source_url=RSS_SOURCE_URL,
profile=profile,
last_accessed_at=timezone.now(),
)
# When: The endpoint is requested without url
response = self.client.get("/bridge/rss/")
# Then: The bridged feeds are listed
self.assertEqual(response.status_code, 200)
body = response.json()
self.assertEqual(
body["data"],
[{"url": RSS_SOURCE_URL, "feed": rss_feed_url(RSS_SOURCE_URL)}],
)
def test_invalid_url_parameter_returns_400(self):
# Given: Invalid url parameters
too_long = "https://example.com/" + "a" * 500
for bad_url in ("ftp://example.com/feed", "not-a-url", too_long):
# When: The endpoint is requested
response = self.client.get("/bridge/rss/", {"url": bad_url})
# Then: The request is rejected
self.assertEqual(response.status_code, 400, bad_url)
self.assertEqual(response.json()["type"], "Error")
def test_unreachable_feed_returns_502(self):
# Given: A feed URL that cannot be fetched
with patch(
"app.bridge.rss.fetch_rss_feed",
side_effect=BridgeError("timeout"),
):
# When: The endpoint is requested
response = self.client.get("/bridge/rss/", {"url": RSS_SOURCE_URL})
# Then: A bad gateway error is returned and nothing is stored
self.assertEqual(response.status_code, 502)
self.assertEqual(BridgedRssFeed.objects.count(), 0)
class BridgeIndexViewTest(TestCase):
"""Test cases for the bridge index endpoint."""
def test_index_exposes_bridge_links(self):
# When: The bridge index is requested
response = self.client.get("/bridge/")
# Then: Both bridge types are discoverable
self.assertEqual(response.status_code, 200)
links = response.json()["_links"]
self.assertIn("activitypub-feed", links)
self.assertIn("rss-feed", links)
def test_root_endpoint_links_to_bridges(self):
# When: The relay root is requested
response = self.client.get("/")
# Then: The bridge endpoints are listed
links = response.json()["_links"]
self.assertIn("bridge", links)
self.assertIn("bridge-activitypub", links)
self.assertIn("bridge-rss", links)
class LastAccessedTrackingTest(TestCase):
"""Test cases for access tracking used by refresh and cleanup."""
def test_old_last_accessed_is_updated_on_get(self):
# Given: A bridge last accessed two days ago
profile = Profile.objects.create(
feed=activitypub_feed_url("alice@m.example"),
title="Alice",
nick="alice",
version="v1",
)
old_access = timezone.now() - timedelta(days=2)
bridge = BridgedActivityPubAccount.objects.create(
handle="alice@m.example",
actor_url="https://m.example/users/alice",
profile=profile,
last_accessed_at=old_access,
last_refreshed_at=timezone.now(),
)
# When: The feed is requested
self.client.get(AP_FEED_PATH)
# Then: last_accessed_at moves forward
bridge.refresh_from_db()
self.assertGreater(bridge.last_accessed_at, old_access)
Reference in New Issue View Git Blame Copy Permalink