From 9b6ec3cb79f74b6deb8194b1b6993b60355acf15 Mon Sep 17 00:00:00 2001
From: supersonicwisd1 <supersonicwisd1>
Date: Mon, 9 Dec 2024 15:16:56 +0100
Subject: [PATCH] feat: media handling (image, video, and audio)
Signed-off-by: supersonicwisd1 <supersonicwisd1>
---
README.md => README-old.md | 0
pyproject.toml | 36 +-
requirements.txt | 9 +
setup.py | 41 ++-
src/pyfed/content/media.py | 211 +++++++++--
src/pyfed/content/video.py | 139 +++++++
src/pyfed/protocols/__init__.py | 27 --
src/pyfed/storage/backend.py | 177 +++++++++
src/pyfed/storage/factory.py | 42 +++
src/pyfed/storage/s3.py | 162 ++++++++
tests/conftest.py | 105 +++---
tests/pytest.ini | 8 +-
tests/unit_tests/models/test_activities.py | 265 ++++++++------
.../serializers/test_serialization.py | 345 +++++++++---------
14 files changed, 1134 insertions(+), 433 deletions(-)
rename README.md => README-old.md (100%)
create mode 100644 src/pyfed/content/video.py
delete mode 100644 src/pyfed/protocols/__init__.py
create mode 100644 src/pyfed/storage/backend.py
create mode 100644 src/pyfed/storage/factory.py
create mode 100644 src/pyfed/storage/s3.py
diff --git a/README.md b/README-old.md
similarity index 100%
rename from README.md
rename to README-old.md
diff --git a/pyproject.toml b/pyproject.toml
index dbc1678..73e67d4 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,34 +1,22 @@
[build-system]
-requires = ["setuptools>=45", "wheel"]
+requires = ["setuptools>=42", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "pyfed"
version = "0.1.0"
-description = "ActivityPub Federation Library"
+description = "A comprehensive Python library for ActivityPub federation"
requires-python = ">=3.8"
-dependencies = [
- "aiohttp>=3.8.0",
- "pydantic>=2.0.0",
- "sqlalchemy>=2.0.0",
- "cryptography>=3.4.0",
- "pytest>=7.0.0",
- "pytest-asyncio>=0.18.0",
- "aiosqlite>=0.17.0",
-]
+dynamic = ["readme", "dependencies"]
+
+[tool.setuptools.dynamic]
+dependencies = {file = ["requirements.txt"]}
+readme = {file = ["README.md"]}
[tool.pytest.ini_options]
-asyncio_mode = "auto"
+minversion = "6.0"
+addopts = "-ra -q --import-mode=importlib"
testpaths = ["tests"]
-python_files = ["test_*.py"]
-addopts = "-ra -q"
-markers = [
- "asyncio: mark test as async",
-]
-
-[project.optional-dependencies]
-test = [
- "pytest>=7.0.0",
- "pytest-asyncio>=0.18.0",
- "pytest-cov>=3.0.0",
-]
\ No newline at end of file
+pythonpath = ["src"]
+asyncio_mode = "auto"
+filterwarnings = ["ignore::DeprecationWarning"]
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index 0c06fd5..e926c32 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -29,6 +29,7 @@ iniconfig==2.0.0
Markdown==3.7
motor==3.6.0
multidict==6.1.0
+mutagen==1.47.0
packaging==24.1
pluggy==1.5.0
prometheus_client==0.21.0
@@ -36,6 +37,7 @@ propcache==0.2.0
pycparser==2.22
pydantic==2.9.2
pydantic_core==2.23.4
+pydub==0.25.1
PyJWT==2.9.0
pymongo==4.9.2
pytest==8.3.3
@@ -52,3 +54,10 @@ starlette==0.41.2
typing_extensions==4.12.2
uvicorn==0.32.1
yarl==1.15.2
+aioboto3==12.3.0
+blurhash==1.1.4
+python-magic==0.4.27
+Pillow==10.2.0
+audioread==3.0.1
+moviepy>=1.0.3
+ffmpeg-python>=0.2.0
diff --git a/setup.py b/setup.py
index 331dde7..50164fc 100644
--- a/setup.py
+++ b/setup.py
@@ -3,6 +3,12 @@ from setuptools import setup, find_packages
setup(
name="pyfed",
version="0.1.0",
+ description="A comprehensive Python library for ActivityPub federation",
+ long_description=open("README.md").read(),
+ long_description_content_type="text/markdown",
+ author="PyFed Team",
+ author_email="team@pyfed.org",
+ url="https://github.com/pyfed/pyfed",
package_dir={"": "src"},
packages=find_packages(where="src"),
python_requires=">=3.8",
@@ -16,6 +22,14 @@ setup(
"redis>=4.0.0",
"beautifulsoup4>=4.9.3",
"markdown>=3.3.4",
+ "Pillow>=10.2.0",
+ "python-magic>=0.4.27",
+ "mutagen>=1.47.0",
+ "pydub>=0.25.1",
+ "numpy>=1.24.0",
+ "blurhash>=1.1.4",
+ "aioboto3>=12.3.0",
+ "aiofiles>=24.1.0",
],
extras_require={
"fastapi": ["fastapi>=0.68.0", "uvicorn>=0.15.0"],
@@ -25,6 +39,31 @@ setup(
"pytest>=6.2.5",
"pytest-asyncio>=0.15.1",
"pytest-cov>=2.12.1",
+ "black>=22.3.0",
+ "isort>=5.10.1",
+ "flake8>=4.0.1",
+ "mypy>=0.950",
],
},
-)
\ No newline at end of file
+ classifiers=[
+ "Development Status :: 4 - Beta",
+ "Intended Audience :: Developers",
+ "License :: OSI Approved :: MIT License",
+ "Operating System :: OS Independent",
+ "Programming Language :: Python :: 3",
+ "Programming Language :: Python :: 3.8",
+ "Programming Language :: Python :: 3.9",
+ "Programming Language :: Python :: 3.10",
+ "Programming Language :: Python :: 3.11",
+ "Programming Language :: Python :: 3.12",
+ "Topic :: Internet :: WWW/HTTP",
+ "Topic :: Software Development :: Libraries :: Python Modules",
+ "Framework :: AsyncIO",
+ ],
+ keywords="activitypub federation fediverse social-network",
+ project_urls={
+ "Documentation": "https://pyfed.readthedocs.io/",
+ "Source": "https://github.com/pyfed/pyfed",
+ "Tracker": "https://github.com/pyfed/pyfed/issues",
+ },
+)
\ No newline at end of file
diff --git a/src/pyfed/content/media.py b/src/pyfed/content/media.py
index 78e8a27..cd87a2e 100644
--- a/src/pyfed/content/media.py
+++ b/src/pyfed/content/media.py
@@ -2,7 +2,7 @@
Media attachment handling implementation.
"""
-from typing import Dict, Any, Optional, List
+from typing import Dict, Any, Optional, List, Union, BinaryIO
import aiohttp
import mimetypes
import hashlib
@@ -10,9 +10,18 @@ from pathlib import Path
import magic
from PIL import Image
import asyncio
+import blurhash
+import io
+import mutagen
+from mutagen.easyid3 import EasyID3
+from pydub import AudioSegment
+import numpy as np
+import base64
+from .video import VideoProcessor
from ..utils.exceptions import MediaError
from ..utils.logging import get_logger
+from ..storage.backend import StorageBackend, LocalStorageBackend
logger = get_logger(__name__)
@@ -20,10 +29,10 @@ class MediaHandler:
"""Handle media attachments."""
def __init__(self,
- upload_path: str = "uploads",
- max_size: int = 10_000_000, # 10MB
+ storage_backend: Optional[StorageBackend] = None,
+ max_size: int = 50 * 1024 * 1024, # 50MB default
allowed_types: Optional[List[str]] = None):
- self.upload_path = Path(upload_path)
+ self.storage = storage_backend or LocalStorageBackend()
self.max_size = max_size
self.allowed_types = allowed_types or [
'image/jpeg',
@@ -31,9 +40,12 @@ class MediaHandler:
'image/gif',
'video/mp4',
'audio/mpeg',
- 'audio/ogg'
+ 'audio/mp3',
+ 'audio/ogg',
+ 'audio/wav',
+ 'audio/flac',
+ 'audio/aac'
]
- self.upload_path.mkdir(parents=True, exist_ok=True)
async def process_attachment(self,
url: str,
@@ -65,25 +77,30 @@ class MediaHandler:
ext = mimetypes.guess_extension(mime_type) or ''
filename = f"{file_hash}{ext}"
- # Save file
- file_path = self.upload_path / filename
- with open(file_path, 'wb') as f:
- f.write(content)
+ # Save file using storage backend
+ file_url = await self.storage.save(filename, content)
- # Generate thumbnails for images
- thumbnails = {}
+ # Process media based on type
+ media_info = {}
if mime_type.startswith('image/'):
- thumbnails = await self._generate_thumbnails(file_path)
+ image = Image.open(io.BytesIO(content))
+ media_info = {
+ 'width': image.width,
+ 'height': image.height,
+ 'blurhash': await self._generate_blurhash(image),
+ 'thumbnails': await self._generate_thumbnails(image, filename)
+ }
+ elif mime_type.startswith('audio/'):
+ media_info = await self._process_audio(content, filename)
+ elif mime_type.startswith('video/'):
+ media_info = await self._process_video(content, mime_type)
return {
"type": "Document",
"mediaType": mime_type,
- "url": f"/media/{filename}",
+ "url": file_url,
"name": description or filename,
- "blurhash": await self._generate_blurhash(file_path) if mime_type.startswith('image/') else None,
- "width": await self._get_image_width(file_path) if mime_type.startswith('image/') else None,
- "height": await self._get_image_height(file_path) if mime_type.startswith('image/') else None,
- "thumbnails": thumbnails
+ **media_info
}
except Exception as e:
@@ -101,25 +118,30 @@ class MediaHandler:
except Exception as e:
raise MediaError(f"Failed to download media: {e}")
- async def _generate_thumbnails(self, file_path: Path) -> Dict[str, Dict[str, Any]]:
+ async def _generate_thumbnails(self, image: Image.Image, original_filename: str) -> Dict[str, Dict[str, Any]]:
"""Generate image thumbnails."""
thumbnails = {}
sizes = [(320, 320), (640, 640)]
try:
- image = Image.open(file_path)
for width, height in sizes:
thumb = image.copy()
thumb.thumbnail((width, height))
- thumb_hash = hashlib.sha256(str(file_path).encode()).hexdigest()
+ # Convert to bytes
+ thumb_bytes = io.BytesIO()
+ thumb.save(thumb_bytes, format='JPEG', quality=85)
+ thumb_bytes.seek(0)
+
+ # Generate thumbnail filename
+ thumb_hash = hashlib.sha256(f"{original_filename}_{width}x{height}".encode()).hexdigest()
thumb_filename = f"thumb_{width}x{height}_{thumb_hash}.jpg"
- thumb_path = self.upload_path / thumb_filename
- thumb.save(thumb_path, "JPEG", quality=85)
+ # Save thumbnail using storage backend
+ thumb_url = await self.storage.save(thumb_filename, thumb_bytes.getvalue())
thumbnails[f"{width}x{height}"] = {
- "url": f"/media/{thumb_filename}",
+ "url": thumb_url,
"width": thumb.width,
"height": thumb.height
}
@@ -130,26 +152,137 @@ class MediaHandler:
logger.error(f"Failed to generate thumbnails: {e}")
return {}
- async def _generate_blurhash(self, file_path: Path) -> Optional[str]:
+ async def _generate_blurhash(self, image: Image.Image) -> str:
"""Generate blurhash for image."""
try:
- # Implementation for blurhash generation
- return None
- except Exception:
+ # Convert image to RGB if necessary
+ if image.mode != 'RGB':
+ image = image.convert('RGB')
+
+ # Calculate blurhash
+ return blurhash.encode(
+ image,
+ x_components=4,
+ y_components=3
+ )
+ except Exception as e:
+ logger.error(f"Failed to generate blurhash: {e}")
return None
- async def _get_image_width(self, file_path: Path) -> Optional[int]:
- """Get image width."""
+ async def _process_audio(self, content: bytes, filename: str) -> Dict[str, Any]:
+ """Process audio file and extract metadata."""
+ try:
+ # Save temporary file for audio processing
+ temp_file = io.BytesIO(content)
+ temp_file.name = filename # Required for mutagen
+
+ # Extract metadata
+ audio = mutagen.File(temp_file)
+ if audio is None:
+ return {}
+
+ # Get basic metadata
+ metadata = {
+ 'duration': int(audio.info.length),
+ 'bitrate': int(audio.info.bitrate),
+ 'channels': getattr(audio.info, 'channels', None),
+ 'sample_rate': getattr(audio.info, 'sample_rate', None)
+ }
+
+ # Try to get ID3 tags if available
+ try:
+ if isinstance(audio, EasyID3) or hasattr(audio, 'tags'):
+ tags = audio.tags or {}
+ metadata.update({
+ 'title': str(tags.get('title', [''])[0]),
+ 'artist': str(tags.get('artist', [''])[0]),
+ 'album': str(tags.get('album', [''])[0]),
+ 'genre': str(tags.get('genre', [''])[0])
+ })
+ except Exception as e:
+ logger.warning(f"Failed to extract audio tags: {e}")
+
+ # Generate waveform
+ try:
+ waveform_data = await self._generate_waveform(content)
+ metadata['waveform'] = waveform_data
+ except Exception as e:
+ logger.warning(f"Failed to generate waveform: {e}")
+
+ # Generate audio preview
+ try:
+ preview_url = await self._generate_audio_preview(content, filename)
+ metadata['preview_url'] = preview_url
+ except Exception as e:
+ logger.warning(f"Failed to generate audio preview: {e}")
+
+ return metadata
+
+ except Exception as e:
+ logger.error(f"Failed to process audio: {e}")
+ return {}
+
+ async def _generate_waveform(self, content: bytes, num_points: int = 100) -> str:
+ """Generate waveform data for audio visualization."""
try:
- with Image.open(file_path) as img:
- return img.width
- except Exception:
+ # Load audio using pydub
+ audio = AudioSegment.from_file(io.BytesIO(content))
+
+ # Convert to mono and get raw data
+ samples = np.array(audio.get_array_of_samples())
+
+ # Resample to desired number of points
+ samples = np.array_split(samples, num_points)
+ waveform = [float(abs(chunk).mean()) for chunk in samples]
+
+ # Normalize to 0-1 range
+ max_val = max(waveform)
+ if max_val > 0:
+ waveform = [val/max_val for val in waveform]
+
+ # Convert to base64 for efficient transfer
+ waveform_bytes = np.array(waveform, dtype=np.float32).tobytes()
+ return base64.b64encode(waveform_bytes).decode('utf-8')
+
+ except Exception as e:
+ logger.error(f"Failed to generate waveform: {e}")
return None
- async def _get_image_height(self, file_path: Path) -> Optional[int]:
- """Get image height."""
+ async def _generate_audio_preview(self, content: bytes, filename: str, duration: int = 30) -> Optional[str]:
+ """Generate a short preview of the audio file."""
try:
- with Image.open(file_path) as img:
- return img.height
- except Exception:
- return None
\ No newline at end of file
+ # Load audio
+ audio = AudioSegment.from_file(io.BytesIO(content))
+
+ # Take first 30 seconds
+ preview_duration = min(duration * 1000, len(audio))
+ preview = audio[:preview_duration]
+
+ # Convert to MP3 format
+ preview_bytes = io.BytesIO()
+ preview.export(preview_bytes, format='mp3', bitrate='128k')
+ preview_bytes.seek(0)
+
+ # Generate preview filename
+ preview_hash = hashlib.sha256(f"preview_{filename}".encode()).hexdigest()
+ preview_filename = f"preview_{preview_hash}.mp3"
+
+ # Save preview using storage backend
+ preview_url = await self.storage.save(preview_filename, preview_bytes.getvalue())
+
+ return preview_url
+
+ except Exception as e:
+ logger.error(f"Failed to generate audio preview: {e}")
+ return None
+
+ async def _process_video(self,
+ file_data: Union[bytes, BinaryIO, Path],
+ mime_type: str) -> Dict[str, Any]:
+ """Process video file and extract metadata."""
+ is_valid, detected_mime = await VideoProcessor.validate_video(file_data)
+ if not is_valid:
+ raise MediaError(f"Invalid or unsupported video format: {detected_mime}")
+
+ metadata = await VideoProcessor.process_video(file_data, mime_type)
+ return metadata
\ No newline at end of file
diff --git a/src/pyfed/content/video.py b/src/pyfed/content/video.py
new file mode 100644
index 0000000..75c0272
--- /dev/null
+++ b/src/pyfed/content/video.py
@@ -0,0 +1,139 @@
+"""
+Video processing module for PyFed.
+"""
+
+from typing import Dict, Any, Optional, Tuple, BinaryIO
+import asyncio
+from pathlib import Path
+import tempfile
+import os
+import ffmpeg
+from moviepy.editor import VideoFileClip
+import io
+from PIL import Image
+import magic
+
+from ..utils.exceptions import MediaError
+from ..utils.logging import get_logger
+
+logger = get_logger(__name__)
+
+class VideoProcessor:
+ """Handle video processing operations."""
+
+ SUPPORTED_FORMATS = {
+ 'video/mp4': '.mp4',
+ 'video/webm': '.webm',
+ 'video/ogg': '.ogv',
+ 'video/quicktime': '.mov',
+ }
+
+ MAX_THUMBNAIL_SIZE = (1280, 720) # 720p
+ THUMBNAIL_QUALITY = 85
+
+ @classmethod
+ async def process_video(cls,
+ video_data: Union[bytes, BinaryIO, Path],
+ mime_type: Optional[str] = None) -> Dict[str, Any]:
+ """Process video file and extract metadata.
+
+ Args:
+ video_data: Raw video data or file-like object
+ mime_type: Optional MIME type of the video
+
+ Returns:
+ Dict containing video metadata and preview
+ """
+ with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as temp_file:
+ if isinstance(video_data, bytes):
+ temp_file.write(video_data)
+ elif isinstance(video_data, Path):
+ temp_file.write(video_data.read_bytes())
+ else:
+ temp_file.write(video_data.read())
+ temp_path = temp_file.name
+
+ try:
+ # Get video information using ffmpeg
+ probe = await asyncio.to_thread(
+ ffmpeg.probe,
+ temp_path
+ )
+
+ video_info = next(s for s in probe['streams'] if s['codec_type'] == 'video')
+
+ # Generate thumbnail
+ thumbnail = await cls.generate_thumbnail(temp_path)
+
+ metadata = {
+ 'duration': float(probe['format'].get('duration', 0)),
+ 'width': int(video_info.get('width', 0)),
+ 'height': int(video_info.get('height', 0)),
+ 'codec': video_info.get('codec_name', ''),
+ 'bitrate': int(probe['format'].get('bit_rate', 0)),
+ 'thumbnail': thumbnail,
+ 'size': os.path.getsize(temp_path)
+ }
+
+ return metadata
+
+ except Exception as e:
+ raise MediaError(f"Failed to process video: {str(e)}") from e
+
+ finally:
+ os.unlink(temp_path)
+
+ @classmethod
+ async def generate_thumbnail(cls, video_path: str, time_offset: float = 1.0) -> bytes:
+ """Generate a thumbnail from the video.
+
+ Args:
+ video_path: Path to video file
+ time_offset: Time offset in seconds for thumbnail extraction
+
+ Returns:
+ bytes: Thumbnail image data in JPEG format
+ """
+ try:
+ # Extract frame using ffmpeg
+ out, _ = (
+ ffmpeg
+ .input(video_path, ss=time_offset)
+ .filter('scale', cls.MAX_THUMBNAIL_SIZE[0], cls.MAX_THUMBNAIL_SIZE[1])
+ .output('pipe:', format='image2', vframes=1)
+ .run(capture_stdout=True)
+ )
+
+ # Convert to PIL Image and optimize
+ image = Image.open(io.BytesIO(out))
+ output = io.BytesIO()
+ image.save(output,
+ format='JPEG',
+ quality=cls.THUMBNAIL_QUALITY,
+ optimize=True)
+
+ return output.getvalue()
+
+ except Exception as e:
+ raise MediaError(f"Failed to generate thumbnail: {str(e)}") from e
+
+ @classmethod
+ async def validate_video(cls, file_data: Union[bytes, BinaryIO, Path]) -> Tuple[bool, str]:
+ """Validate video file format and content.
+
+ Args:
+ file_data: Video file data
+
+ Returns:
+ Tuple[bool, str]: (is_valid, mime_type)
+ """
+ if isinstance(file_data, (bytes, BinaryIO)):
+ mime = magic.from_buffer(
+ file_data if isinstance(file_data, bytes) else file_data.read(2048),
+ mime=True
+ )
+ else:
+ mime = magic.from_file(str(file_data), mime=True)
+
+ is_valid = mime in cls.SUPPORTED_FORMATS
+ return is_valid, mime
diff --git a/src/pyfed/protocols/__init__.py b/src/pyfed/protocols/__init__.py
deleted file mode 100644
index b0d1be4..0000000
--- a/src/pyfed/protocols/__init__.py
+++ /dev/null
@@ -1,27 +0,0 @@
-"""
-Protocol implementations for PyFed.
-"""
-from .base import (
- BaseProtocol, BaseProtocolConfig, ProtocolVersion,
- ProtocolError, ValidationError, SecurityError, TimeoutError
-)
-from .c2s import C2SConfig, ClientToServerProtocol
-from .s2s import ServerToServerProtocol
-
-__all__ = [
- # Base
- 'BaseProtocol',
- 'BaseProtocolConfig',
- 'ProtocolVersion',
- 'ProtocolError',
- 'ValidationError',
- 'SecurityError',
- 'TimeoutError',
-
- # Client-to-Server
- 'C2SConfig',
- 'ClientToServerProtocol',
-
- # Server-to-Server
- 'ServerToServerProtocol',
-]
diff --git a/src/pyfed/storage/backend.py b/src/pyfed/storage/backend.py
new file mode 100644
index 0000000..036b080
--- /dev/null
+++ b/src/pyfed/storage/backend.py
@@ -0,0 +1,177 @@
+from abc import ABC, abstractmethod
+from typing import Protocol, Optional, Union, BinaryIO, Dict, Any
+from pathlib import Path
+import asyncio
+from datetime import datetime
+import mimetypes
+import os
+
+class StorageBackend(Protocol):
+ """Protocol defining the unified storage interface for both data and media."""
+
+ @abstractmethod
+ async def initialize(self, config: Dict[str, Any]) -> None:
+ """Initialize the storage backend with configuration."""
+ pass
+
+ @abstractmethod
+ async def store_file(self,
+ file_data: Union[bytes, BinaryIO, Path],
+ path: str,
+ mime_type: Optional[str] = None,
+ metadata: Optional[Dict[str, Any]] = None) -> str:
+ """Store a file in the storage backend.
+
+ Args:
+ file_data: The file data to store
+ path: The path/key where the file should be stored
+ mime_type: Optional MIME type of the file
+ metadata: Optional metadata to store with the file
+
+ Returns:
+ str: The URI/path where the file was stored
+ """
+ pass
+
+ @abstractmethod
+ async def retrieve_file(self, path: str) -> bytes:
+ """Retrieve a file from storage.
+
+ Args:
+ path: Path/key of the file to retrieve
+
+ Returns:
+ bytes: The file contents
+ """
+ pass
+
+ @abstractmethod
+ async def delete_file(self, path: str) -> None:
+ """Delete a file from storage.
+
+ Args:
+ path: Path/key of the file to delete
+ """
+ pass
+
+ @abstractmethod
+ async def file_exists(self, path: str) -> bool:
+ """Check if a file exists in storage.
+
+ Args:
+ path: Path/key to check
+
+ Returns:
+ bool: True if file exists, False otherwise
+ """
+ pass
+
+ @abstractmethod
+ async def get_file_metadata(self, path: str) -> Dict[str, Any]:
+ """Get metadata for a stored file.
+
+ Args:
+ path: Path/key of the file
+
+ Returns:
+ Dict[str, Any]: File metadata including size, creation time, etc.
+ """
+ pass
+
+class LocalStorageBackend:
+ """Local filesystem implementation of StorageBackend."""
+
+ def __init__(self):
+ self.base_path: Optional[Path] = None
+
+ async def initialize(self, config: Dict[str, Any]) -> None:
+ """Initialize local storage with base path."""
+ self.base_path = Path(config['base_path']).resolve()
+ self.base_path.mkdir(parents=True, exist_ok=True)
+
+ async def store_file(self,
+ file_data: Union[bytes, BinaryIO, Path],
+ path: str,
+ mime_type: Optional[str] = None,
+ metadata: Optional[Dict[str, Any]] = None) -> str:
+ """Store a file in the local filesystem."""
+ if self.base_path is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self.base_path / path
+ full_path.parent.mkdir(parents=True, exist_ok=True)
+
+ if isinstance(file_data, bytes):
+ full_path.write_bytes(file_data)
+ elif isinstance(file_data, Path):
+ if file_data.is_file():
+ full_path.write_bytes(file_data.read_bytes())
+ else: # BinaryIO
+ with full_path.open('wb') as f:
+ f.write(file_data.read())
+
+ if metadata:
+ meta_path = full_path.with_suffix(full_path.suffix + '.meta')
+ meta_path.write_text(str(metadata))
+
+ return str(full_path.relative_to(self.base_path))
+
+ async def retrieve_file(self, path: str) -> bytes:
+ """Retrieve a file from the local filesystem."""
+ if self.base_path is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self.base_path / path
+ if not full_path.exists():
+ raise FileNotFoundError(f"File not found: {path}")
+
+ return full_path.read_bytes()
+
+ async def delete_file(self, path: str) -> None:
+ """Delete a file from the local filesystem."""
+ if self.base_path is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self.base_path / path
+ if full_path.exists():
+ full_path.unlink()
+
+ # Also delete metadata file if it exists
+ meta_path = full_path.with_suffix(full_path.suffix + '.meta')
+ if meta_path.exists():
+ meta_path.unlink()
+
+ async def file_exists(self, path: str) -> bool:
+ """Check if a file exists in the local filesystem."""
+ if self.base_path is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ return (self.base_path / path).exists()
+
+ async def get_file_metadata(self, path: str) -> Dict[str, Any]:
+ """Get metadata for a stored file."""
+ if self.base_path is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self.base_path / path
+ if not full_path.exists():
+ raise FileNotFoundError(f"File not found: {path}")
+
+ stat = full_path.stat()
+ metadata = {
+ 'size': stat.st_size,
+ 'created': datetime.fromtimestamp(stat.st_ctime),
+ 'modified': datetime.fromtimestamp(stat.st_mtime),
+ 'mime_type': mimetypes.guess_type(str(full_path))[0]
+ }
+
+ # Check for additional metadata file
+ meta_path = full_path.with_suffix(full_path.suffix + '.meta')
+ if meta_path.exists():
+ try:
+ additional_meta = eval(meta_path.read_text()) # Simple evaluation for demo
+ metadata.update(additional_meta)
+ except Exception:
+ pass
+
+ return metadata
diff --git a/src/pyfed/storage/factory.py b/src/pyfed/storage/factory.py
new file mode 100644
index 0000000..4251ae8
--- /dev/null
+++ b/src/pyfed/storage/factory.py
@@ -0,0 +1,42 @@
+from typing import Dict, Any, Type
+from .backend import StorageBackend, LocalStorageBackend
+from .s3 import S3StorageBackend
+
+class StorageFactory:
+ """Factory for creating storage backend instances."""
+
+ _backends = {
+ 'local': LocalStorageBackend,
+ 's3': S3StorageBackend
+ }
+
+ @classmethod
+ def register_backend(cls, name: str, backend_class: Type[StorageBackend]) -> None:
+ """Register a new storage backend type.
+
+ Args:
+ name: Name to register the backend under
+ backend_class: The backend class to register
+ """
+ cls._backends[name] = backend_class
+
+ @classmethod
+ async def create_backend(cls, backend_type: str, config: Dict[str, Any]) -> StorageBackend:
+ """Create and initialize a storage backend instance.
+
+ Args:
+ backend_type: Type of backend to create ('local' or 's3')
+ config: Configuration dictionary for the backend
+
+ Returns:
+ StorageBackend: Initialized storage backend instance
+
+ Raises:
+ ValueError: If backend_type is not recognized
+ """
+ if backend_type not in cls._backends:
+ raise ValueError(f"Unknown storage backend type: {backend_type}")
+
+ backend = cls._backends[backend_type]()
+ await backend.initialize(config)
+ return backend
diff --git a/src/pyfed/storage/s3.py b/src/pyfed/storage/s3.py
new file mode 100644
index 0000000..ca7880d
--- /dev/null
+++ b/src/pyfed/storage/s3.py
@@ -0,0 +1,162 @@
+from typing import Union, BinaryIO, Dict, Any, Optional
+from pathlib import Path
+import aioboto3
+from datetime import datetime
+import io
+import mimetypes
+import json
+
+from .backend import StorageBackend
+
+class S3StorageBackend:
+ """S3-compatible storage backend implementation."""
+
+ def __init__(self):
+ self.session = None
+ self.bucket = None
+ self.prefix = None
+ self.client = None
+
+ async def initialize(self, config: Dict[str, Any]) -> None:
+ """Initialize S3 storage with credentials and configuration."""
+ self.session = aioboto3.Session(
+ aws_access_key_id=config.get('aws_access_key_id'),
+ aws_secret_access_key=config.get('aws_secret_access_key')
+ )
+ self.bucket = config['bucket']
+ self.prefix = config.get('prefix', '')
+ self.endpoint_url = config.get('endpoint_url') # For compatibility with other S3-compatible services
+
+ async def _get_client(self):
+ """Get or create S3 client."""
+ if self.client is None:
+ self.client = self.session.client('s3', endpoint_url=self.endpoint_url)
+ return self.client
+
+ def _get_full_path(self, path: str) -> str:
+ """Get full S3 path including prefix."""
+ return f"{self.prefix.rstrip('/')}/{path.lstrip('/')}" if self.prefix else path
+
+ async def store_file(self,
+ file_data: Union[bytes, BinaryIO, Path],
+ path: str,
+ mime_type: Optional[str] = None,
+ metadata: Optional[Dict[str, Any]] = None) -> str:
+ """Store a file in S3 storage."""
+ if self.session is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self._get_full_path(path)
+
+ # Prepare the file data
+ if isinstance(file_data, bytes):
+ data = io.BytesIO(file_data)
+ elif isinstance(file_data, Path):
+ data = io.BytesIO(file_data.read_bytes())
+ else:
+ data = file_data
+
+ # Prepare upload parameters
+ upload_kwargs = {
+ 'Bucket': self.bucket,
+ 'Key': full_path,
+ 'Body': data
+ }
+
+ if mime_type:
+ upload_kwargs['ContentType'] = mime_type
+
+ if metadata:
+ # S3 metadata must be strings
+ upload_kwargs['Metadata'] = {
+ k: str(v) for k, v in metadata.items()
+ }
+
+ async with await self._get_client() as client:
+ await client.upload_fileobj(**upload_kwargs)
+
+ return full_path
+
+ async def retrieve_file(self, path: str) -> bytes:
+ """Retrieve a file from S3 storage."""
+ if self.session is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self._get_full_path(path)
+ data = io.BytesIO()
+
+ async with await self._get_client() as client:
+ try:
+ await client.download_fileobj(
+ Bucket=self.bucket,
+ Key=full_path,
+ Fileobj=data
+ )
+ except Exception as e:
+ raise FileNotFoundError(f"File not found: {path}") from e
+
+ return data.getvalue()
+
+ async def delete_file(self, path: str) -> None:
+ """Delete a file from S3 storage."""
+ if self.session is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self._get_full_path(path)
+
+ async with await self._get_client() as client:
+ try:
+ await client.delete_object(
+ Bucket=self.bucket,
+ Key=full_path
+ )
+ except Exception as e:
+ # Ignore if file doesn't exist
+ pass
+
+ async def file_exists(self, path: str) -> bool:
+ """Check if a file exists in S3 storage."""
+ if self.session is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self._get_full_path(path)
+
+ async with await self._get_client() as client:
+ try:
+ await client.head_object(
+ Bucket=self.bucket,
+ Key=full_path
+ )
+ return True
+ except:
+ return False
+
+ async def get_file_metadata(self, path: str) -> Dict[str, Any]:
+ """Get metadata for a stored file."""
+ if self.session is None:
+ raise RuntimeError("Storage backend not initialized")
+
+ full_path = self._get_full_path(path)
+
+ async with await self._get_client() as client:
+ try:
+ response = await client.head_object(
+ Bucket=self.bucket,
+ Key=full_path
+ )
+
+ metadata = {
+ 'size': response['ContentLength'],
+ 'created': response.get('LastModified'),
+ 'mime_type': response.get('ContentType'),
+ 'etag': response.get('ETag'),
+ }
+
+ # Include custom metadata if present
+ if 'Metadata' in response:
+ metadata.update(response['Metadata'])
+
+ return metadata
+
+ except Exception as e:
+ raise FileNotFoundError(f"File not found: {path}") from e
diff --git a/tests/conftest.py b/tests/conftest.py
index 544021b..5729d04 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,75 +1,64 @@
"""
-Global test configuration and fixtures.
+Common test fixtures for PyFed unit tests.
"""
-
import pytest
-from aiohttp import web
-import asyncio
-from typing import AsyncGenerator, Callable
-from unittest.mock import AsyncMock, Mock
+from datetime import datetime, timezone
+from pathlib import Path
+import sys
-pytest_plugins = ['pytest_aiohttp']
+# Add src directory to Python path
+src_path = Path(__file__).parent.parent / "src"
+sys.path.insert(0, str(src_path))
-@pytest.fixture
-def app() -> web.Application:
- """Create base application for testing."""
- return web.Application()
+from pyfed.models import APObject, APPerson, APNote
@pytest.fixture
-def auth_middleware() -> Callable:
- """Create auth middleware for testing."""
- async def middleware(app: web.Application, handler: Callable):
- async def auth_handler(request: web.Request):
- request['user'] = {"sub": "test_user"}
- return await handler(request)
- return auth_handler
- return middleware
+def valid_actor():
+ """Return a valid actor URL."""
+ return "https://example.com/user/1"
@pytest.fixture
-def rate_limit_middleware() -> Callable:
- """Create rate limit middleware for testing."""
- async def middleware(app: web.Application, handler: Callable):
- async def rate_limit_handler(request: web.Request):
- # Simple in-memory rate limiting for tests
- key = f"rate_limit:{request.remote}"
- count = app.get(key, 0) + 1
- app[key] = count
-
- if count > 5: # Rate limit after 5 requests
- return web.Response(status=429)
-
- return await handler(request)
- return rate_limit_handler
- return middleware
+def valid_object():
+ """Return a valid object dictionary."""
+ return {
+ "id": "https://example.com/note/123",
+ "type": "Note",
+ "content": "Hello, World!"
+ }
@pytest.fixture
-async def client(aiohttp_client: Callable, app: web.Application) -> AsyncGenerator:
- """Create test client with configured application."""
- return await aiohttp_client(app)
-
-@pytest.fixture
-def mock_storage() -> AsyncMock:
- """Create mock storage."""
- storage = AsyncMock()
- storage.create_activity = AsyncMock(return_value="test_activity_id")
- storage.create_object = AsyncMock(return_value="test_object_id")
- storage.is_following = AsyncMock(return_value=False)
- return storage
+def valid_person():
+ """Return a valid APPerson object."""
+ return APPerson(
+ id="https://example.com/users/alice",
+ type="Person",
+ name="Alice",
+ preferredUsername="alice",
+ inbox="https://example.com/users/alice/inbox",
+ outbox="https://example.com/users/alice/outbox"
+ )
@pytest.fixture
-def mock_delivery() -> AsyncMock:
- """Create mock delivery service."""
- delivery = AsyncMock()
- delivery.deliver_activity = AsyncMock(
- return_value={"success": ["inbox1"], "failed": []}
+def valid_note():
+ """Return a valid APNote object."""
+ return APNote(
+ id="https://example.com/notes/123",
+ type="Note",
+ content="This is a test note",
+ published=datetime.now(timezone.utc)
)
- return delivery
@pytest.fixture
-def mock_resolver() -> AsyncMock:
- """Create mock resolver."""
- resolver = AsyncMock()
- resolver.resolve_actor = AsyncMock(
- return_value={"id": "https://example.com/users/test"}
+def complex_object(valid_person):
+ """Return a complex APObject with nested objects."""
+ return APObject(
+ id="https://example.com/object/123",
+ type="Object",
+ name="Test Object",
+ content="This is a test object.",
+ attachment=[
+ APObject(id="https://example.com/attachment/1", type="Image"),
+ APObject(id="https://example.com/attachment/2", type="Video")
+ ],
+ attributedTo=valid_person.id
)
- return resolver
\ No newline at end of file
diff --git a/tests/pytest.ini b/tests/pytest.ini
index f9025ee..669471c 100644
--- a/tests/pytest.ini
+++ b/tests/pytest.ini
@@ -1,5 +1,7 @@
-[pytest]
+[pytest]
asyncio_mode = auto
-filterwarnings =
+asyncio_default_fixture_loop_scope = function
+filterwarnings =
ignore::DeprecationWarning
-pythonpath = ../src
\ No newline at end of file
+pythonpath = ../src
+addopts = --import-mode=importlib
\ No newline at end of file
diff --git a/tests/unit_tests/models/test_activities.py b/tests/unit_tests/models/test_activities.py
index 5f51632..b870eea 100644
--- a/tests/unit_tests/models/test_activities.py
+++ b/tests/unit_tests/models/test_activities.py
@@ -1,135 +1,166 @@
"""
-test_activities.py
-This module contains tests for the Activity types in ActivityPub.
+Tests for ActivityPub activity types.
"""
import pytest
-from datetime import datetime
+from datetime import datetime, timezone
from pydantic import ValidationError
+
from pyfed.models import (
APCreate, APUpdate, APDelete, APFollow,
- APUndo, APLike, APAnnounce
+ APUndo, APLike, APAnnounce, APActivity
)
-def test_create_activity():
- """Test creating a Create activity."""
- activity = APCreate(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object={
- "id": "https://example.com/note/123",
- "type": "Note",
- "content": "Hello, World!"
- }
- )
- assert activity.type == "Create"
- assert str(activity.actor) == "https://example.com/user/1"
- assert activity.object["type"] == "Note"
-
-def test_update_activity():
- """Test creating an Update activity."""
- activity = APUpdate(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object={
+@pytest.fixture
+def valid_actor():
+ return "https://example.com/user/1"
+
+@pytest.fixture
+def valid_object():
+ return {"id": "https://example.com/note/123", "type": "Note", "content": "Hello, World!"}
+
+@pytest.fixture
+def valid_person():
+ return {"id": "https://example.com/person/1", "type": "Person"}
+
+@pytest.fixture
+def valid_note():
+ return {"id": "https://example.com/note/123", "type": "Note"}
+
+class TestActivityCreation:
+ @pytest.mark.parametrize("activity_class,activity_type", [
+ (APCreate, "Create"),
+ (APUpdate, "Update"),
+ (APDelete, "Delete"),
+ (APFollow, "Follow"),
+ (APUndo, "Undo"),
+ (APLike, "Like"),
+ (APAnnounce, "Announce")
+ ])
+ def test_activity_type(self, activity_class, activity_type, valid_actor, valid_object):
+ """Test that each activity has the correct type."""
+ activity = activity_class(
+ id="https://example.com/activity/123",
+ actor=valid_actor,
+ object=valid_object
+ )
+ assert activity.type == activity_type
+ assert str(activity.actor) == valid_actor
+
+ @pytest.mark.parametrize("invalid_actor", [
+ None,
+ "",
+ "not-a-url",
+ "http:/invalid.com"
+ ])
+ def test_invalid_actor(self, invalid_actor, valid_object):
+ """Test activity creation with invalid actors."""
+ with pytest.raises(ValidationError):
+ APCreate(
+ id="https://example.com/activity/123",
+ actor=invalid_actor,
+ object=valid_object
+ )
+
+ @pytest.mark.parametrize("activity_class", [
+ APCreate, APUpdate, APDelete, APLike, APAnnounce
+ ])
+ def test_missing_object(self, activity_class, valid_actor):
+ """Test that non-intransitive activities require an object."""
+ with pytest.raises(ValidationError):
+ activity_class(
+ id="https://example.com/activity/123",
+ actor=valid_actor
+ )
+
+class TestActivityMetadata:
+ def test_activity_with_published(self, valid_actor, valid_object):
+ """Test activity with published timestamp."""
+ now = datetime.now(timezone.utc)
+ activity = APCreate(
+ id="https://example.com/activity/123",
+ actor=valid_actor,
+ object=valid_object,
+ published=now
+ )
+ assert activity.published == now
+
+ def test_activity_with_target(self, valid_actor, valid_object):
+ """Test activity with a target."""
+ target = {"id": "https://example.com/target/123", "type": "Collection"}
+ activity = APCreate(
+ id="https://example.com/activity/123",
+ actor=valid_actor,
+ object=valid_object,
+ target=target
+ )
+ assert activity.target["id"] == target["id"]
+ assert activity.target["type"] == target["type"]
+
+ def test_activity_with_result(self, valid_actor, valid_object):
+ """Test activity with a result."""
+ result = {"id": "https://example.com/result/123", "type": "Note"}
+ activity = APCreate(
+ id="https://example.com/activity/123",
+ actor=valid_actor,
+ object=valid_object,
+ result=result
+ )
+ assert activity.result["id"] == result["id"]
+ assert activity.result["type"] == result["type"]
+
+class TestSpecificActivities:
+ def test_create_activity(self, valid_actor, valid_object):
+ """Test specific behavior of Create activity."""
+ activity = APCreate(
+ id="https://example.com/activity/123",
+ actor=valid_actor,
+ object=valid_object
+ )
+ assert activity.type == "Create"
+ assert activity.object["content"] == valid_object["content"]
+
+ def test_update_activity(self, valid_actor):
+ """Test specific behavior of Update activity."""
+ updated_object = {
"id": "https://example.com/note/123",
"type": "Note",
"content": "Updated content"
}
- )
- assert activity.type == "Update"
- assert activity.object["content"] == "Updated content"
-
-def test_delete_activity():
- """Test creating a Delete activity."""
- activity = APDelete(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object="https://example.com/note/123"
- )
- assert activity.type == "Delete"
- assert str(activity.object) == "https://example.com/note/123"
-
-def test_follow_activity():
- """Test creating a Follow activity."""
- activity = APFollow(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object="https://example.com/user/2"
- )
- assert activity.type == "Follow"
- assert str(activity.object) == "https://example.com/user/2"
-
-def test_undo_activity():
- """Test creating an Undo activity."""
- activity = APUndo(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object={
- "id": "https://example.com/activity/456",
- "type": "Follow",
- "actor": "https://example.com/user/1",
- "object": "https://example.com/user/2"
- }
- )
- assert activity.type == "Undo"
- assert activity.object["type"] == "Follow"
-
-def test_like_activity():
- """Test creating a Like activity."""
- activity = APLike(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object="https://example.com/note/123"
- )
- assert activity.type == "Like"
- assert str(activity.object) == "https://example.com/note/123"
-
-def test_announce_activity():
- """Test creating an Announce activity."""
- activity = APAnnounce(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object="https://example.com/note/123"
- )
- assert activity.type == "Announce"
- assert str(activity.object) == "https://example.com/note/123"
-
-def test_activity_with_target():
- """Test creating an activity with a target."""
- activity = APCreate(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object="https://example.com/note/123",
- target="https://example.com/collection/1"
- )
- assert str(activity.target) == "https://example.com/collection/1"
-
-def test_activity_with_result():
- """Test creating an activity with a result."""
- activity = APCreate(
- id="https://example.com/activity/123",
- actor="https://example.com/user/1",
- object="https://example.com/note/123",
- result={
- "id": "https://example.com/result/1",
- "type": "Note",
- "content": "Result content"
- }
- )
- assert activity.result["type"] == "Note"
+ activity = APUpdate(
+ id="https://example.com/activity/123",
+ actor=valid_actor,
+ object=updated_object
+ )
+ assert activity.type == "Update"
+ assert activity.object["content"] == "Updated content"
+
+ def test_delete_activity(self, valid_actor):
+ """Test specific behavior of Delete activity."""
+ object_id = "https://example.com/note/123"
+ activity = APDelete(
+ id="https://example.com/activity/123",
+ actor=valid_actor,
+ object=object_id
+ )
+ assert activity.type == "Delete"
+ assert str(activity.object) == object_id
-def test_invalid_activity_missing_actor():
- """Test that activity creation fails when actor is missing."""
- with pytest.raises(ValidationError):
- APCreate(
+ def test_follow_activity(self, valid_actor, valid_person):
+ """Test specific behavior of Follow activity."""
+ activity = APFollow(
id="https://example.com/activity/123",
- object="https://example.com/note/123"
+ actor=valid_actor,
+ object=valid_person.id
)
+ assert activity.type == "Follow"
+ assert str(activity.object) == valid_person.id
-def test_invalid_activity_missing_object():
- """Test that activity creation fails when object is missing for non-intransitive activities."""
- with pytest.raises(ValidationError):
- APCreate(
+ def test_like_activity(self, valid_actor, valid_note):
+ """Test specific behavior of Like activity."""
+ activity = APLike(
id="https://example.com/activity/123",
- actor="https://example.com/user/1"
+ actor=valid_actor,
+ object=valid_note.id
)
+ assert activity.type == "Like"
+ assert str(activity.object) == valid_note.id
diff --git a/tests/unit_tests/serializers/test_serialization.py b/tests/unit_tests/serializers/test_serialization.py
index 9d30b32..594d884 100644
--- a/tests/unit_tests/serializers/test_serialization.py
+++ b/tests/unit_tests/serializers/test_serialization.py
@@ -1,7 +1,6 @@
"""
Tests for ActivityPub serialization.
"""
-
import pytest
from datetime import datetime, timezone
import json
@@ -12,176 +11,194 @@ from pyfed.models import (
)
from pyfed.serializers.json_serializer import ActivityPubSerializer
-def test_serialize_ap_object():
- """Test basic object serialization."""
- obj = APObject(
- id="https://example.com/object/123",
- type="Object",
- name="Test Object",
- content="This is a test object."
- )
- serialized = obj.serialize()
-
- # Verify serialization
- assert serialized["@context"] == "https://www.w3.org/ns/activitystreams"
- assert serialized["id"] == "https://example.com/object/123"
- assert serialized["type"] == "Object"
- assert serialized["name"] == "Test Object"
- assert serialized["content"] == "This is a test object."
+class TestBasicSerialization:
+ @pytest.mark.parametrize("field,value", [
+ ("name", "Test Object"),
+ ("content", "This is a test object."),
+ ("summary", "A summary"),
+ ("url", "https://example.com/object"),
+ ])
+ def test_simple_field_serialization(self, field, value):
+ """Test serialization of simple field types."""
+ obj = APObject(
+ id="https://example.com/object/123",
+ type="Object",
+ **{field: value}
+ )
+ serialized = obj.serialize()
+ assert serialized[field] == value
-def test_serialize_with_datetime():
- """Test serialization of objects with datetime fields."""
- now = datetime.now(timezone.utc)
- obj = APObject(
- id="https://example.com/object/123",
- type="Object",
- published=now,
- updated=now
- )
- serialized = obj.serialize()
-
- # Verify datetime serialization
- assert serialized["published"] == now.isoformat()
- assert serialized["updated"] == now.isoformat()
+ def test_datetime_serialization(self):
+ """Test serialization of datetime fields."""
+ now = datetime.now(timezone.utc)
+ obj = APObject(
+ id="https://example.com/object/123",
+ type="Object",
+ published=now,
+ updated=now
+ )
+ serialized = obj.serialize()
+ assert serialized["published"] == now.isoformat()
+ assert serialized["updated"] == now.isoformat()
-def test_serialize_nested_objects():
- """Test serialization of objects with nested objects."""
- author = APPerson(
- id="https://example.com/users/alice",
- type="Person",
- name="Alice",
- inbox="https://example.com/users/alice/inbox",
- outbox="https://example.com/users/alice/outbox"
- )
- note = APNote(
- id="https://example.com/notes/123",
- type="Note",
- content="Hello, World!",
- attributed_to=author
- )
- serialized = note.serialize()
-
- # Verify nested object serialization
- assert serialized["attributedTo"]["id"] == "https://example.com/users/alice"
- assert serialized["attributedTo"]["type"] == "Person"
- assert serialized["attributedTo"]["name"] == "Alice"
+ def test_context_handling(self):
+ """Test that @context is properly handled."""
+ obj = APObject(
+ id="https://example.com/object/123",
+ type="Object"
+ )
+ serialized = obj.serialize()
+ assert "@context" in serialized
+ assert serialized["@context"] == "https://www.w3.org/ns/activitystreams"
-def test_serialize_collection():
- """Test serialization of collections."""
- items = [
- APNote(
- id=f"https://example.com/notes/{i}",
- type="Note",
- content=f"Note {i}"
- ) for i in range(3)
- ]
- collection = APCollection(
- id="https://example.com/collection/1",
- type="Collection",
- total_items=len(items),
- items=items
- )
- serialized = collection.serialize()
-
- # Verify collection serialization
- assert serialized["type"] == "Collection"
- assert serialized["totalItems"] == 3
- assert len(serialized["items"]) == 3
- assert all(item["type"] == "Note" for item in serialized["items"])
+class TestComplexSerialization:
+ def test_nested_object_serialization(self, valid_person, valid_note):
+ """Test serialization of objects with nested objects."""
+ note = valid_note
+ note.attributedTo = valid_person
+ serialized = note.serialize()
+
+ assert serialized["attributedTo"]["id"] == valid_person.id
+ assert serialized["attributedTo"]["type"] == "Person"
+ assert serialized["attributedTo"]["name"] == valid_person.name
+
+ def test_collection_serialization(self):
+ """Test serialization of collections."""
+ items = [
+ APObject(id=f"https://example.com/item/{i}", type="Object")
+ for i in range(3)
+ ]
+ collection = APCollection(
+ id="https://example.com/collection/1",
+ type="Collection",
+ totalItems=len(items),
+ items=items
+ )
+ serialized = collection.serialize()
+
+ assert serialized["type"] == "Collection"
+ assert serialized["totalItems"] == 3
+ assert len(serialized["items"]) == 3
+ assert all(item["type"] == "Object" for item in serialized["items"])
-def test_serialize_activity():
- """Test serialization of activities."""
- note = APNote(
- id="https://example.com/notes/123",
- type="Note",
- content="Hello, World!"
- )
- create = APCreate(
- id="https://example.com/activities/1",
- type="Create",
- actor="https://example.com/users/alice",
- object=note
- )
- serialized = create.serialize()
-
- # Verify activity serialization
- assert serialized["type"] == "Create"
- assert serialized["actor"] == "https://example.com/users/alice"
- assert serialized["object"]["type"] == "Note"
- assert serialized["object"]["content"] == "Hello, World!"
+ def test_activity_serialization(self, valid_person, valid_note):
+ """Test serialization of activities."""
+ activity = APCreate(
+ id="https://example.com/activity/1",
+ type="Create",
+ actor=valid_person,
+ object=valid_note
+ )
+ serialized = activity.serialize()
+
+ assert serialized["type"] == "Create"
+ assert serialized["actor"]["id"] == valid_person.id
+ assert serialized["object"]["id"] == valid_note.id
-def test_deserialize_ap_object():
- """Test basic object deserialization."""
- data = {
- "@context": "https://www.w3.org/ns/activitystreams",
- "type": "Object",
- "id": "https://example.com/object/123",
- "name": "Test Object",
- "content": "This is a test object."
- }
- obj = ActivityPubSerializer.deserialize(data, APObject)
-
- # Verify deserialization
- assert str(obj.id) == "https://example.com/object/123"
- assert obj.type == "Object"
- assert obj.name == "Test Object"
- assert obj.content == "This is a test object."
+class TestDeserialization:
+ def test_basic_deserialization(self):
+ """Test basic object deserialization."""
+ json_data = {
+ "@context": "https://www.w3.org/ns/activitystreams",
+ "id": "https://example.com/object/123",
+ "type": "Object",
+ "name": "Test Object"
+ }
+ obj = APObject.deserialize(json.dumps(json_data))
+ assert obj.id == json_data["id"]
+ assert obj.type == json_data["type"]
+ assert obj.name == json_data["name"]
-def test_deserialize_from_json_string():
- """Test deserialization from JSON string."""
- json_str = json.dumps({
- "type": "Object",
- "id": "https://example.com/object/123",
- "name": "Test Object"
- })
- obj = ActivityPubSerializer.deserialize(json_str, APObject)
-
- # Verify deserialization from string
- assert str(obj.id) == "https://example.com/object/123"
- assert obj.type == "Object"
- assert obj.name == "Test Object"
+ @pytest.mark.parametrize("invalid_json", [
+ "",
+ "not json",
+ "{invalid json}",
+ "[]",
+ "null"
+ ])
+ def test_invalid_json_deserialization(self, invalid_json):
+ """Test deserialization with invalid JSON."""
+ with pytest.raises((json.JSONDecodeError, ValidationError)):
+ APObject.deserialize(invalid_json)
-def test_deserialize_invalid_json():
- """Test deserialization of invalid JSON."""
- with pytest.raises(ValueError):
- ActivityPubSerializer.deserialize("invalid json", APObject)
+ def test_missing_required_fields(self):
+ """Test deserialization with missing required fields."""
+ json_data = {
+ "@context": "https://www.w3.org/ns/activitystreams",
+ "name": "Test Object"
+ }
+ with pytest.raises(ValidationError):
+ APObject.deserialize(json.dumps(json_data))
-def test_deserialize_missing_required_fields():
- """Test deserialization with missing required fields."""
- data = {"type": "Object", "name": "Test"} # Missing required 'id'
- with pytest.raises(Exception): # Pydantic will raise validation error
- ActivityPubSerializer.deserialize(data, APObject)
+ def test_extra_fields_handling(self):
+ """Test deserialization with extra unknown fields."""
+ json_data = {
+ "@context": "https://www.w3.org/ns/activitystreams",
+ "id": "https://example.com/object/123",
+ "type": "Object",
+ "name": "Test Object",
+ "unknownField": "value",
+ "anotherUnknown": 123
+ }
+ obj = APObject.deserialize(json.dumps(json_data))
+ assert obj.id == json_data["id"]
+ assert obj.type == json_data["type"]
+ assert obj.name == json_data["name"]
-def test_serialize_deserialize_complex_object():
- """Test round-trip serialization and deserialization."""
- original = APNote(
- id="https://example.com/notes/123",
- type="Note",
- content="Test content",
- to=["https://example.com/users/bob"],
- cc=["https://www.w3.org/ns/activitystreams#Public"]
- )
- serialized = original.serialize()
- deserialized = ActivityPubSerializer.deserialize(serialized, APNote)
-
- # Verify round-trip
- assert str(deserialized.id) == str(original.id)
- assert deserialized.type == original.type
- assert deserialized.content == original.content
- assert deserialized.to == original.to
- assert deserialized.cc == original.cc
+class TestEdgeCases:
+ def test_empty_collection_serialization(self):
+ """Test serialization of empty collections."""
+ collection = APCollection(
+ id="https://example.com/collection/1",
+ type="Collection",
+ totalItems=0,
+ items=[]
+ )
+ serialized = collection.serialize()
+ assert serialized["totalItems"] == 0
+ assert serialized["items"] == []
-def test_deserialize_with_extra_fields():
- """Test deserialization with extra fields in JSON."""
- data = {
- "type": "Object",
- "id": "https://example.com/object/123",
- "name": "Test Object",
- "extra_field": "Should be ignored"
- }
- obj = ActivityPubSerializer.deserialize(data, APObject)
-
- # Verify extra fields are handled
- assert str(obj.id) == "https://example.com/object/123"
- assert obj.type == "Object"
- assert obj.name == "Test Object"
+ def test_large_collection_serialization(self):
+ """Test serialization of large collections."""
+ items = [
+ APObject(id=f"https://example.com/item/{i}", type="Object")
+ for i in range(1000)
+ ]
+ collection = APCollection(
+ id="https://example.com/collection/1",
+ type="Collection",
+ totalItems=len(items),
+ items=items
+ )
+ serialized = collection.serialize()
+ assert len(serialized["items"]) == 1000
+
+ @pytest.mark.parametrize("long_string", [
+ "a" * 1000,
+ "b" * 10000,
+ "c" * 100000
+ ])
+ def test_long_string_serialization(self, long_string):
+ """Test serialization of objects with very long string fields."""
+ obj = APObject(
+ id="https://example.com/object/123",
+ type="Object",
+ content=long_string
+ )
+ serialized = obj.serialize()
+ assert serialized["content"] == long_string
+
+ def test_circular_reference_handling(self, valid_person):
+ """Test handling of circular references in serialization."""
+ person = valid_person
+ note = APNote(
+ id="https://example.com/note/123",
+ type="Note",
+ content="Test note",
+ attributedTo=person
+ )
+ person.liked = [note] # Create circular reference
+
+ # Should not cause infinite recursion
+ serialized_person = person.serialize()
+ assert serialized_person["liked"][0]["id"] == note.id
--
GitLab