Compare commits

...

1 commit

Author SHA1 Message Date
b3009fd102 first test structure 2024-10-17 21:06:19 +02:00
4 changed files with 95 additions and 24 deletions

View file

@ -20,6 +20,7 @@ ACTIVITYPUB_AUTHORS = {
'alice': { 'alice': {
'name': 'Alice', 'name': 'Alice',
'movedTo': 'https://fedi.example/users/alice', 'movedTo': 'https://fedi.example/users/alice',
'movedToName': '@alice@fedi.example',
'alsoKnownAs': ['https://fedi.example/users/alice'], 'alsoKnownAs': ['https://fedi.example/users/alice'],
'summary': 'Hi, I\'m Alice! Please follow me at @alice@fedi.example.', 'summary': 'Hi, I\'m Alice! Please follow me at @alice@fedi.example.',
'attachment': [ 'attachment': [

View file

@ -1,28 +1,30 @@
import datetime import datetime
import logging
import json import json
import logging
import os import os
import urllib.parse import urllib.parse
from typing import Any, Dict
import pelican.writers import pelican.writers
from pelican import signals from pelican import signals
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
__version__ = '0.1.2' __version__ = '0.1.2'
pagination = 25 PAGINATION = 25
ENCODING = 'UTF-8'
def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Writer): def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Writer):
now = datetime.datetime.utcnow() now = datetime.datetime.utcnow()
author = generator.settings['AUTHOR'] author = generator.settings['AUTHOR']
domain = urllib.parse.urlparse(generator.settings['SITEURL']).netloc domain: str = urllib.parse.urlparse(generator.settings['SITEURL']).netloc
wkhmpath = os.path.join(writer.output_path, '.well-known/host-meta') wkhmpath = os.path.join(writer.output_path, '.well-known/host-meta')
wknipath = os.path.join(writer.output_path, '.well-known/nodeinfo') wknipath = os.path.join(writer.output_path, '.well-known/nodeinfo')
@ -72,7 +74,7 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
} }
} }
wfurl = os.path.join(generator.settings['SITEURL'], '.well-known/webfinger?resource={uri}') wfurl = os.path.join(generator.settings['SITEURL'], '.well-known/webfinger?resource={uri}')
hostmeta = f'<?xml version="1.0" encoding="UTF-8"?><XRD xmlns="http://docs.oasis-open.org/ns/xri/xrd-1.0"><Link rel="lrdd" template="{wfurl}" type="application/xrd+xml" /></XRD>' hostmeta = f'<?xml version="1.0" encoding="{ENCODING}"?><XRD xmlns="http://docs.oasis-open.org/ns/xri/xrd-1.0"><Link rel="lrdd" template="{wfurl}" type="application/xrd+xml" /></XRD>'
webfinger = { webfinger = {
'subject': f'acct:{author}@{domain}', 'subject': f'acct:{author}@{domain}',
'aliases': [ 'aliases': [
@ -92,13 +94,13 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
} }
] ]
} }
with open(wkhmpath, 'w') as hf: with open(wkhmpath, 'w', encoding=ENCODING) as hf:
hf.write(hostmeta) hf.write(hostmeta)
with open(wknipath, 'w') as nf: with open(wknipath, 'w', encoding=ENCODING) as nf:
json.dump(wknodeinfo, nf) json.dump(wknodeinfo, nf)
with open(nipath, 'w') as nf: with open(nipath, 'w', encoding=ENCODING) as nf:
json.dump(nodeinfo, nf) json.dump(nodeinfo, nf)
with open(wfpath, 'w') as wf: with open(wfpath, 'w', encoding=ENCODING) as wf:
json.dump(webfinger, wf) json.dump(webfinger, wf)
for t in generator.tags: for t in generator.tags:
@ -118,7 +120,7 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
'totalItems': len(articles), 'totalItems': len(articles),
'orderedItems': articles 'orderedItems': articles
} }
with open(path, 'w') as f: with open(path, 'w', encoding=ENCODING) as f:
json.dump(tag, f) json.dump(tag, f)
articlemap = {} articlemap = {}
@ -142,13 +144,13 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
replyto = os.path.join(generator.settings['SITEURL'], 'activitypub/posts', sa.slug) replyto = os.path.join(generator.settings['SITEURL'], 'activitypub/posts', sa.slug)
break break
cc = [os.path.join(generator.settings['SITEURL'], 'activitypub/collections/followers', article.author.slug)] cc = [os.path.join(generator.settings['SITEURL'], 'activitypub/collections/followers', article.author.slug)]
aa = generator.settings.get('ACTIVITYPUB_AUTHORS', {}).get(author.name, {}) aa: Dict[str, Dict[str, Any]] = generator.settings.get('ACTIVITYPUB_AUTHORS', {}).get(author.name, {})
if 'movedTo' in aa: if 'movedTo' in aa:
cc.append(aa['movedTo']) cc.append(aa['movedTo'])
tags.append({ tags.append({
'type': 'Mention', 'type': 'Mention',
'href': aa['movedTo'], 'href': aa['movedTo'],
'name': aa['_movedTo_name'] 'name': aa['movedToName']
}) })
cmap = {} cmap = {}
tmap = {} tmap = {}
@ -173,7 +175,7 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
'attachment': [], 'attachment': [],
'tag': tags 'tag': tags
} }
with open(apath, 'w') as f: with open(apath, 'w', encoding=ENCODING) as f:
json.dump(articlemap[article.slug], f) json.dump(articlemap[article.slug], f)
for author, articles in generator.authors: for author, articles in generator.authors:
@ -255,7 +257,7 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
'totalItems': 0, 'totalItems': 0,
'orderedItems': [] 'orderedItems': []
} }
maxpage = len(creates) // pagination maxpage = len(creates) // PAGINATION
outbox = { outbox = {
'@context': ['https://www.w3.org/ns/activitystreams'], '@context': ['https://www.w3.org/ns/activitystreams'],
'type': 'OrderedCollection', 'type': 'OrderedCollection',
@ -278,8 +280,8 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
'totalItems': 0, 'totalItems': 0,
'orderedItems': [] 'orderedItems': []
} }
for i in range(0, len(creates), pagination): for i in range(0, len(creates), PAGINATION):
ipage = i // pagination ipage = i // PAGINATION
outpageurl = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage)) outpageurl = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage))
outpagepath = os.path.join(writer.output_path, 'activitypub/collections/outbox_page', author.slug, str(ipage)) outpagepath = os.path.join(writer.output_path, 'activitypub/collections/outbox_page', author.slug, str(ipage))
page = { page = {
@ -288,13 +290,13 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
'id': outpageurl, 'id': outpageurl,
'totalItems': len(creates), 'totalItems': len(creates),
'partOf': outboxurl, 'partOf': outboxurl,
'orderedItems': creates[i:i+pagination] 'orderedItems': creates[i:i+PAGINATION]
} }
if ipage > 0: if ipage > 0:
page['prev'] = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage-1)) page['prev'] = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage-1))
if ipage < maxpage: if ipage < maxpage:
page['next'] = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage+1)) page['next'] = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage+1))
with open(outpagepath, 'w') as f: with open(outpagepath, 'w', encoding=ENCODING) as f:
json.dump(page, f) json.dump(page, f)
author_webfinger = { author_webfinger = {
'subject': f'acct:{author.name}@{domain}', 'subject': f'acct:{author.name}@{domain}',
@ -315,17 +317,17 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri
} }
] ]
} }
with open(os.path.join(awfpath, author.name), 'w') as f: with open(os.path.join(awfpath, author.name), 'w', encoding=ENCODING) as f:
json.dump(author_webfinger, f) json.dump(author_webfinger, f)
with open(os.path.join(authorpath, author.name), 'w') as f: with open(os.path.join(authorpath, author.name), 'w', encoding=ENCODING) as f:
json.dump(a, f) json.dump(a, f)
with open(inboxpath, 'w') as f: with open(inboxpath, 'w', encoding=ENCODING) as f:
json.dump(inbox, f) json.dump(inbox, f)
with open(outboxpath, 'w') as f: with open(outboxpath, 'w', encoding=ENCODING) as f:
json.dump(outbox, f) json.dump(outbox, f)
with open(followingpath, 'w') as f: with open(followingpath, 'w', encoding=ENCODING) as f:
json.dump(following, f) json.dump(following, f)
with open(followerspath, 'w') as f: with open(followerspath, 'w', encoding=ENCODING) as f:
json.dump(followers, f) json.dump(followers, f)

View file

@ -0,0 +1,62 @@
# SPDX-FileCopyrightText: 2024 Tobias Schmidl
#
# SPDX-License-Identifier: MIT
"""Unit tests for the ActivityPub plugin"""
import datetime
import unittest
from unittest.mock import MagicMock, patch
import os
import json
from activitypub import ap_article, ENCODING
class TestApArticle(unittest.TestCase):
"""Tests the ap_article function"""
@patch('os.makedirs')
@patch('builtins.open')
@patch('json.dump')
def test_ap_article(self, mock_json_dump, mock_open, mock_makedirs):
# Mock the generator and writer
generator = MagicMock()
writer = MagicMock()
# Mock settings
generator.settings = {
'AUTHOR': 'test_author',
'SITEURL': 'http://example.com',
'SITENAME': 'Test Site',
'ACTIVITYPUB_AUTHORS': {}
}
# Mock authors and articles
generator.authors = [('test_author', MagicMock(slug='test_author'))]
generator.articles = [MagicMock(slug='test_article', date=datetime.datetime.utcnow(), title='Test Article', content='Test Content', metadata={'tags': ['test_tag']}, author=MagicMock(slug='test_author'), url='test_article.html', translations=[])]
generator.tags = [MagicMock(slug='test_tag', name='test_tag')]
# Call the function
ap_article(generator, writer)
# Check if directories were created
mock_makedirs.assert_any_call(os.path.join(writer.output_path, '.well-known'), exist_ok=True)
mock_makedirs.assert_any_call(os.path.join(writer.output_path, 'activitypub/users'), exist_ok=True)
mock_makedirs.assert_any_call(os.path.join(writer.output_path, 'activitypub/posts'), exist_ok=True)
mock_makedirs.assert_any_call(os.path.join(writer.output_path, 'activitypub/tags'), exist_ok=True)
mock_makedirs.assert_any_call(os.path.join(writer.output_path, 'activitypub/collections/inbox'), exist_ok=True)
mock_makedirs.assert_any_call(os.path.join(writer.output_path, 'activitypub/collections/outbox'), exist_ok=True)
mock_makedirs.assert_any_call(os.path.join(writer.output_path, 'activitypub/collections/outbox_page', 'test_author'), exist_ok=True)
mock_makedirs.assert_any_call(os.path.join(writer.output_path, 'activitypub/collections/following'), exist_ok=True)
mock_makedirs.assert_any_call(os.path.join(writer.output_path, 'activitypub/collections/followers'), exist_ok=True)
# Check if files were written
mock_open.assert_any_call(os.path.join(writer.output_path, '.well-known/host-meta'), 'w', encoding=ENCODING)
mock_open.assert_any_call(os.path.join(writer.output_path, '.well-known/nodeinfo'), 'w', encoding=ENCODING)
mock_open.assert_any_call(os.path.join(writer.output_path, 'activitypub/nodeinfo'), 'w', encoding=ENCODING)
mock_open.assert_any_call(os.path.join(writer.output_path, '.well-known/webfinger'), 'w', encoding=ENCODING)
# Check if JSON data was dumped
self.assertTrue(mock_json_dump.called)
if __name__ == '__main__':
unittest.main()

View file

@ -40,3 +40,9 @@ build-backend = "setuptools.build_meta"
[tool.setuptools.dynamic] [tool.setuptools.dynamic]
version = {attr = "pelican.plugins.activitypub.activitypub.__version__"} version = {attr = "pelican.plugins.activitypub.activitypub.__version__"}
[project.optional-dependencies]
Test = [
"pytest>=8.3.0",
"isort"
]