make instantiation explicit so easier to test.

This commit is contained in:
Moon 2025-06-13 13:53:41 +09:00
parent a516a9b55a
commit 40f018a83b
6 changed files with 88 additions and 18 deletions

View file

@ -1,7 +1,7 @@
import requests
import config
from fediverse_factory import get_fediverse_service
import db_utils
from config import RARITY_TO_WEIGHT
def add_card(name: str, rarity: int, weight: float, image_url: str) -> tuple[int, str]:
"""
@ -28,7 +28,7 @@ def add_card(name: str, rarity: int, weight: float, image_url: str) -> tuple[int
raise ValueError('Card name cannot be empty.')
if rarity < 1:
raise ValueError('Rarity must be a positive integer.')
if rarity not in RARITY_TO_WEIGHT.keys():
if rarity not in config.RARITY_TO_WEIGHT.keys():
raise ValueError(f'Invalid rarity: {rarity}')
if not image_url:
raise ValueError('Image URL must be provided.')
@ -40,7 +40,7 @@ def add_card(name: str, rarity: int, weight: float, image_url: str) -> tuple[int
raise RuntimeError(f"Failed to download image from {image_url}")
# Upload to Fediverse instance
fediverse_service = get_fediverse_service()
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
try:
uploaded_file = fediverse_service.upload_file(response.raw)
file_id = uploaded_file.id

View file

@ -1,5 +1,6 @@
import time
import traceback
import config
from notification import process_fediverse_notification
from db_utils import get_config, set_config, connect, setup_administrators
from fediverse_factory import get_fediverse_service
@ -12,7 +13,7 @@ def stream_notifications():
setup_administrators()
# Initialize the Fediverse service
fediverse_service = get_fediverse_service()
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
# Get the last seen notification ID from the database
last_seen_id = get_config("last_seen_notif_id")

View file

@ -1,20 +1,18 @@
from fediverse_service import FediverseService
from misskey_service import MisskeyService
from pleroma_service import PleromaService
import config
class FediverseServiceFactory:
"""Factory for creating FediverseService implementations based on configuration"""
"""Factory for creating FediverseService implementations"""
@staticmethod
def create_service(instance_type: str = None) -> FediverseService:
def create_service(instance_type: str) -> FediverseService:
"""
Create a FediverseService implementation based on the instance type.
Args:
instance_type: The type of instance ("misskey" or "pleroma").
If None, reads from config.INSTANCE_TYPE
instance_type: The type of instance ("misskey" or "pleroma")
Returns:
FediverseService implementation (MisskeyService or PleromaService)
@ -22,9 +20,6 @@ class FediverseServiceFactory:
Raises:
ValueError: If the instance type is not supported
"""
if instance_type is None:
instance_type = config.INSTANCE_TYPE
instance_type = instance_type.lower()
if instance_type == "misskey":
@ -35,11 +30,11 @@ class FediverseServiceFactory:
raise ValueError(f"Unsupported instance type: {instance_type}")
def get_fediverse_service(instance_type: str = None) -> FediverseService:
def get_fediverse_service(instance_type: str) -> FediverseService:
"""
Convenience function to get a FediverseService instance
Args:
instance_type: Optional instance type override for testing
instance_type: The instance type ("misskey" or "pleroma")
"""
return FediverseServiceFactory.create_service(instance_type)

View file

@ -0,0 +1,74 @@
"""Mock FediverseService for testing purposes"""
from typing import List, Optional, Union, BinaryIO
from fediverse_service import FediverseService
from fediverse_types import FediverseNotification, FediversePost, FediverseFile, Visibility
class MockFediverseService(FediverseService):
"""Mock implementation of FediverseService for testing"""
def __init__(self):
self.notifications = []
self.created_posts = []
self.uploaded_files = []
def get_notifications(self, since_id: Optional[str] = None) -> List[FediverseNotification]:
"""Return mock notifications, optionally filtered by since_id"""
if since_id is None:
return self.notifications
# Filter notifications newer than since_id
filtered = []
for notif in self.notifications:
if notif.id > since_id:
filtered.append(notif)
return filtered
def create_post(
self,
text: str,
reply_to_id: Optional[str] = None,
visibility: Visibility = Visibility.HOME,
file_ids: Optional[List[str]] = None,
visible_user_ids: Optional[List[str]] = None
) -> str:
"""Mock post creation, returns fake post ID"""
post_id = f"mock_post_{len(self.created_posts)}"
# Store the post for assertions
self.created_posts.append({
'id': post_id,
'text': text,
'reply_to_id': reply_to_id,
'visibility': visibility,
'file_ids': file_ids,
'visible_user_ids': visible_user_ids
})
return post_id
def upload_file(self, file_data: Union[BinaryIO, bytes]) -> FediverseFile:
"""Mock file upload, returns fake file"""
file_id = f"mock_file_{len(self.uploaded_files)}"
mock_file = FediverseFile(
id=file_id,
url=f"https://example.com/files/{file_id}",
type="image/png",
name="test_file.png"
)
self.uploaded_files.append(mock_file)
return mock_file
# Helper methods for testing
def add_mock_notification(self, notification: FediverseNotification):
"""Add a mock notification for testing"""
self.notifications.append(notification)
def clear_all(self):
"""Clear all mock data"""
self.notifications.clear()
self.created_posts.clear()
self.uploaded_files.clear()

View file

@ -1,4 +1,4 @@
from config import USE_WHITELIST
import config
from parsing import parse_notification
from db_utils import is_whitelisted, is_player_banned
from response import generate_response
@ -10,7 +10,7 @@ from fediverse_types import FediverseNotification, NotificationType, Visibility
def process_fediverse_notification(notification: FediverseNotification, fediverse_service=None) -> None:
'''Processes an individual fediverse notification using the abstraction'''
if fediverse_service is None:
fediverse_service = get_fediverse_service()
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
# Get user and instance info
username = notification.user.username
@ -18,7 +18,7 @@ def process_fediverse_notification(notification: FediverseNotification, fedivers
instance = host if host else 'local'
# Check whitelist
if USE_WHITELIST and not is_whitelisted(instance):
if config.USE_WHITELIST and not is_whitelisted(instance):
print(f'⚠️ Blocked notification from untrusted instance: {instance}')
return

View file

@ -10,7 +10,7 @@ def parse_notification(notification: FediverseNotification, fediverse_service=No
generate_response()'''
if fediverse_service is None:
fediverse_service = get_fediverse_service()
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
# We get the type of notification to filter the ones that we actually want
# to parse