first test structure
This commit is contained in:
parent
803abfaf97
commit
b3009fd102
4 changed files with 95 additions and 24 deletions
|
@ -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': [
|
||||||
|
|
|
@ -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)
|
||||||
|
|
||||||
|
|
||||||
|
|
62
pelican/plugins/activitypub/activitypub_test.py
Normal file
62
pelican/plugins/activitypub/activitypub_test.py
Normal 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()
|
|
@ -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"
|
||||||
|
]
|
Loading…
Reference in a new issue