Compare commits
	
		
			1 commit
		
	
	
		
			main
			...
			feature/te
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| b3009fd102 | 
					 4 changed files with 95 additions and 24 deletions
				
			
		|  | @ -20,6 +20,7 @@ ACTIVITYPUB_AUTHORS = { | |||
|     'alice': { | ||||
|         'name': 'Alice', | ||||
|         'movedTo': 'https://fedi.example/users/alice', | ||||
|         'movedToName': '@alice@fedi.example', | ||||
|         'alsoKnownAs': ['https://fedi.example/users/alice'], | ||||
|         'summary': 'Hi, I\'m Alice!  Please follow me at @alice@fedi.example.', | ||||
|         'attachment': [ | ||||
|  |  | |||
|  | @ -1,28 +1,30 @@ | |||
| 
 | ||||
| import datetime | ||||
| import logging | ||||
| import json | ||||
| import logging | ||||
| import os | ||||
| import urllib.parse | ||||
| from typing import Any, Dict | ||||
| 
 | ||||
| import pelican.writers | ||||
| 
 | ||||
| from pelican import signals | ||||
| 
 | ||||
| 
 | ||||
| log = logging.getLogger(__name__) | ||||
| 
 | ||||
| 
 | ||||
| __version__ = '0.1.2' | ||||
| 
 | ||||
| pagination = 25 | ||||
| PAGINATION = 25 | ||||
| 
 | ||||
| ENCODING = 'UTF-8' | ||||
| 
 | ||||
| def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Writer): | ||||
| 
 | ||||
|     now = datetime.datetime.utcnow() | ||||
| 
 | ||||
|     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') | ||||
|     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}') | ||||
|     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 = { | ||||
|         'subject': f'acct:{author}@{domain}', | ||||
|         '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) | ||||
|     with open(wknipath, 'w') as nf: | ||||
|     with open(wknipath, 'w', encoding=ENCODING) as nf: | ||||
|         json.dump(wknodeinfo, nf) | ||||
|     with open(nipath, 'w') as nf: | ||||
|     with open(nipath, 'w', encoding=ENCODING) as nf: | ||||
|         json.dump(nodeinfo, nf) | ||||
|     with open(wfpath, 'w') as wf: | ||||
|     with open(wfpath, 'w', encoding=ENCODING) as wf: | ||||
|         json.dump(webfinger, wf) | ||||
| 
 | ||||
|     for t in generator.tags: | ||||
|  | @ -118,7 +120,7 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri | |||
|             'totalItems': len(articles), | ||||
|             'orderedItems': articles | ||||
|         } | ||||
|         with open(path, 'w') as f: | ||||
|         with open(path, 'w', encoding=ENCODING) as f: | ||||
|             json.dump(tag, f) | ||||
| 
 | ||||
|     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) | ||||
|                     break | ||||
|         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: | ||||
|             cc.append(aa['movedTo']) | ||||
|             tags.append({ | ||||
|                 'type': 'Mention', | ||||
|                 'href': aa['movedTo'], | ||||
|                 'name': aa['_movedTo_name'] | ||||
|                 'name': aa['movedToName'] | ||||
|             }) | ||||
|         cmap = {} | ||||
|         tmap = {} | ||||
|  | @ -173,7 +175,7 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri | |||
|             'attachment': [], | ||||
|             'tag': tags | ||||
|         } | ||||
|         with open(apath, 'w') as f: | ||||
|         with open(apath, 'w', encoding=ENCODING) as f: | ||||
|             json.dump(articlemap[article.slug], f) | ||||
| 
 | ||||
|     for author, articles in generator.authors: | ||||
|  | @ -255,7 +257,7 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri | |||
|             'totalItems': 0, | ||||
|             'orderedItems': [] | ||||
|         } | ||||
|         maxpage = len(creates) // pagination | ||||
|         maxpage = len(creates) // PAGINATION | ||||
|         outbox = { | ||||
|             '@context': ['https://www.w3.org/ns/activitystreams'], | ||||
|             'type': 'OrderedCollection', | ||||
|  | @ -278,8 +280,8 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri | |||
|             'totalItems': 0, | ||||
|             'orderedItems': [] | ||||
|         } | ||||
|         for i in range(0, len(creates), pagination): | ||||
|             ipage = i // pagination | ||||
|         for i in range(0, len(creates), PAGINATION): | ||||
|             ipage = i // PAGINATION | ||||
|             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)) | ||||
|             page = { | ||||
|  | @ -288,13 +290,13 @@ def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Wri | |||
|                 'id': outpageurl, | ||||
|                 'totalItems': len(creates), | ||||
|                 'partOf': outboxurl, | ||||
|                 'orderedItems': creates[i:i+pagination] | ||||
|                 'orderedItems': creates[i:i+PAGINATION] | ||||
|             } | ||||
|             if ipage > 0: | ||||
|                 page['prev'] = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage-1)) | ||||
|             if ipage < maxpage: | ||||
|                 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) | ||||
|         author_webfinger = { | ||||
|             '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) | ||||
|         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) | ||||
|         with open(inboxpath, 'w') as f: | ||||
|         with open(inboxpath, 'w', encoding=ENCODING) as f: | ||||
|             json.dump(inbox, f) | ||||
|         with open(outboxpath, 'w') as f: | ||||
|         with open(outboxpath, 'w', encoding=ENCODING) as f: | ||||
|             json.dump(outbox, f) | ||||
|         with open(followingpath, 'w') as f: | ||||
|         with open(followingpath, 'w', encoding=ENCODING) as f: | ||||
|             json.dump(following, f) | ||||
|         with open(followerspath, 'w') as f: | ||||
|         with open(followerspath, 'w', encoding=ENCODING) as 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] | ||||
| version = {attr = "pelican.plugins.activitypub.activitypub.__version__"} | ||||
| 
 | ||||
| [project.optional-dependencies] | ||||
| Test = [ | ||||
|     "pytest>=8.3.0", | ||||
|     "isort" | ||||
| ] | ||||
		Loading…
	
	Add table
		Add a link
		
	
		Reference in a new issue