diff --git a/app/models/__init__.py b/app/models/__init__.py index b534741..51228b1 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -11,6 +11,8 @@ from app.models.message import Message from app.models.notification import Notification from app.models.notification_preference import NotificationPreference +from app.models.plugin import Plugin +from app.models.plugin import PluginInstallation from app.models.project import Project from app.models.project_comment import ProjectComment from app.models.project_file import ProjectFile @@ -31,6 +33,8 @@ "Message", "Notification", "NotificationPreference", + "Plugin", + "PluginInstallation", "Project", "ProjectComment", "ProjectFile", diff --git a/app/models/plugin.py b/app/models/plugin.py new file mode 100644 index 0000000..6d290b4 --- /dev/null +++ b/app/models/plugin.py @@ -0,0 +1,108 @@ +"""Plugin database models. + +Stores plugin metadata, installations, and configuration per workspace. +""" + +from datetime import datetime + +from app.extensions import db + + +class Plugin(db.Model): + """Installed plugin metadata (global).""" + + __tablename__ = "plugins" + + id = db.Column(db.String(64), primary_key=True) # e.g., "stellar-tools" + name = db.Column(db.String(256), nullable=False) + version = db.Column(db.String(32), nullable=False) + description = db.Column(db.Text) + author = db.Column(db.String(256)) + entry_point = db.Column(db.String(512), nullable=False) + capabilities = db.Column(db.JSON, default=[]) # List of capability strings + permissions = db.Column(db.JSON, default=[]) + dependencies = db.Column(db.JSON, default=[]) + configuration = db.Column(db.JSON, default={}) + + enabled = db.Column(db.Boolean, default=True) + installed_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False) + updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False) + + # Relationships + installations = db.relationship( + "PluginInstallation", + back_populates="plugin", + cascade="all, delete-orphan", + lazy="select", + ) + + def to_dict(self) -> dict: + """Convert to dictionary representation.""" + return { + "id": self.id, + "name": self.name, + "version": self.version, + "description": self.description, + "author": self.author, + "entry_point": self.entry_point, + "capabilities": self.capabilities or [], + "permissions": self.permissions or [], + "dependencies": self.dependencies or [], + "configuration": self.configuration or {}, + "enabled": self.enabled, + "installed_at": self.installed_at.isoformat() if self.installed_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } + + def __repr__(self) -> str: + """String representation.""" + return f"" + + +class PluginInstallation(db.Model): + """Plugin installation in a specific workspace. + + Tracks plugin-specific installation, configuration, and granted capabilities + within a workspace. + """ + + __tablename__ = "plugin_installations" + + id = db.Column(db.Integer, primary_key=True) + plugin_id = db.Column(db.String(64), db.ForeignKey("plugins.id"), nullable=False, index=True) + workspace_id = db.Column(db.Integer, db.ForeignKey("workspaces.id"), nullable=False, index=True) + + enabled = db.Column(db.Boolean, default=True) + granted_capabilities = db.Column(db.JSON, default=[]) # List of capability strings + config = db.Column(db.JSON, default={}) # Workspace-specific plugin configuration + + installed_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False) + updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False) + installed_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True) + + # Relationships + plugin = db.relationship("Plugin", back_populates="installations", lazy="select") + installed_by = db.relationship("User", foreign_keys=[installed_by_id], lazy="select") + workspace = db.relationship("Workspace", foreign_keys=[workspace_id], lazy="select") + + __table_args__ = ( + db.UniqueConstraint("plugin_id", "workspace_id", name="uq_plugin_installation"), + ) + + def to_dict(self) -> dict: + """Convert to dictionary representation.""" + return { + "id": self.id, + "plugin_id": self.plugin_id, + "workspace_id": self.workspace_id, + "enabled": self.enabled, + "granted_capabilities": self.granted_capabilities or [], + "config": self.config or {}, + "installed_at": self.installed_at.isoformat() if self.installed_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + "installed_by_id": self.installed_by_id, + } + + def __repr__(self) -> str: + """String representation.""" + return f"" diff --git a/app/services/capabilities.py b/app/services/capabilities.py new file mode 100644 index 0000000..7ce0be8 --- /dev/null +++ b/app/services/capabilities.py @@ -0,0 +1,377 @@ +"""Plugin capability model and management. + +Capabilities define what actions plugins are allowed to perform within +the application. They integrate with workspace roles and permissions. +""" + +from enum import Enum +from typing import Any + +from app.extensions import db + + +class Capability(Enum): + """Plugin capabilities (permissions). + + Capabilities must be explicitly granted to plugins. They are mapped + to workspace member roles to control access. + """ + + # Project capabilities + PROJECT_READ = "project:read" + PROJECT_WRITE = "project:write" + PROJECT_DELETE = "project:delete" + + # Workspace capabilities + WORKSPACE_READ = "workspace:read" + WORKSPACE_WRITE = "workspace:write" + + # GitHub capabilities + GITHUB_READ = "github:read" + GITHUB_WRITE = "github:write" + + # AI capabilities + AI_ACCESS = "ai:access" + AI_ANALYSIS = "ai:analysis" + + # Notification capabilities + NOTIFICATION_CREATE = "notification:create" + + # Stellar capabilities + STELLAR_READ = "stellar:read" + STELLAR_WRITE = "stellar:write" + STELLAR_ANALYSIS = "stellar:analysis" + + # Code review capabilities + REVIEW_READ = "review:read" + REVIEW_CREATE = "review:create" + + @classmethod + def from_string(cls, value: str) -> "Capability | None": + """Get capability from string value. + + Args: + value: String value (e.g., "project:read") + + Returns: + Capability enum or None if not found + """ + for cap in cls: + if cap.value == value: + return cap + return None + + @classmethod + def all_values(cls) -> list[str]: + """Get all capability string values. + + Returns: + List of capability strings + """ + return [cap.value for cap in cls] + + +# Mapping of workspace member roles to capabilities. +# This defines what capabilities are granted based on role. +ROLE_CAPABILITY_MAPPING = { + "viewer": [ + Capability.PROJECT_READ, + Capability.WORKSPACE_READ, + Capability.AI_ACCESS, + Capability.STELLAR_READ, + Capability.REVIEW_READ, + ], + "contributor": [ + Capability.PROJECT_READ, + Capability.PROJECT_WRITE, + Capability.WORKSPACE_READ, + Capability.AI_ACCESS, + Capability.AI_ANALYSIS, + Capability.GITHUB_READ, + Capability.STELLAR_READ, + Capability.STELLAR_ANALYSIS, + Capability.REVIEW_READ, + Capability.REVIEW_CREATE, + ], + "admin": [ + Capability.PROJECT_READ, + Capability.PROJECT_WRITE, + Capability.PROJECT_DELETE, + Capability.WORKSPACE_READ, + Capability.WORKSPACE_WRITE, + Capability.AI_ACCESS, + Capability.AI_ANALYSIS, + Capability.GITHUB_READ, + Capability.GITHUB_WRITE, + Capability.NOTIFICATION_CREATE, + Capability.STELLAR_READ, + Capability.STELLAR_WRITE, + Capability.STELLAR_ANALYSIS, + Capability.REVIEW_READ, + Capability.REVIEW_CREATE, + ], + "owner": [ + Capability.PROJECT_READ, + Capability.PROJECT_WRITE, + Capability.PROJECT_DELETE, + Capability.WORKSPACE_READ, + Capability.WORKSPACE_WRITE, + Capability.AI_ACCESS, + Capability.AI_ANALYSIS, + Capability.GITHUB_READ, + Capability.GITHUB_WRITE, + Capability.NOTIFICATION_CREATE, + Capability.STELLAR_READ, + Capability.STELLAR_WRITE, + Capability.STELLAR_ANALYSIS, + Capability.REVIEW_READ, + Capability.REVIEW_CREATE, + ], +} + + +class CapabilityGrant(db.Model): + """Track capabilities granted to a plugin within a workspace. + + A capability grant represents the intersection of: + - Plugin ID + - Workspace ID + - Granted capabilities + - User who granted the capability (for audit) + """ + + __tablename__ = "plugin_capability_grants" + + id = db.Column(db.Integer, primary_key=True) + plugin_id = db.Column(db.String(64), db.ForeignKey("plugins.id"), nullable=False, index=True) + workspace_id = db.Column(db.Integer, db.ForeignKey("workspaces.id"), nullable=False, index=True) + capability = db.Column(db.String(64), nullable=False) + + granted_at = db.Column(db.DateTime, default=db.func.now(), nullable=False) + granted_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True) + + # Relationships + granted_by = db.relationship("User", foreign_keys=[granted_by_id]) + + __table_args__ = ( + db.UniqueConstraint( + "plugin_id", + "workspace_id", + "capability", + name="uq_plugin_capability_grant", + ), + ) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary representation.""" + return { + "id": self.id, + "plugin_id": self.plugin_id, + "workspace_id": self.workspace_id, + "capability": self.capability, + "granted_at": self.granted_at.isoformat() if self.granted_at else None, + "granted_by_id": self.granted_by_id, + } + + def __repr__(self) -> str: + """String representation.""" + return f"" + + +class CapabilityStore: + """Manage plugin capability grants. + + Capabilities define what actions plugins can perform. This store manages + granting and revoking capabilities to plugins within specific workspaces. + """ + + @staticmethod + def grant( + plugin_id: str, + workspace_id: int, + capability: str, + granted_by_id: int | None = None, + ) -> CapabilityGrant: + """Grant a capability to a plugin in a workspace. + + Args: + plugin_id: Plugin identifier + workspace_id: Workspace identifier + capability: Capability string (e.g., "project:read") + granted_by_id: User ID who is granting the capability + + Returns: + CapabilityGrant record + + Raises: + ValueError: If capability is invalid + """ + # Validate capability + if Capability.from_string(capability) is None: + raise ValueError(f"Invalid capability: {capability}") + + # Check if grant already exists + existing = CapabilityGrant.query.filter_by( + plugin_id=plugin_id, + workspace_id=workspace_id, + capability=capability, + ).first() + + if existing: + return existing + + grant = CapabilityGrant( + plugin_id=plugin_id, + workspace_id=workspace_id, + capability=capability, + granted_by_id=granted_by_id, + ) + db.session.add(grant) + db.session.commit() + return grant + + @staticmethod + def revoke( + plugin_id: str, + workspace_id: int, + capability: str, + ) -> bool: + """Revoke a capability from a plugin in a workspace. + + Args: + plugin_id: Plugin identifier + workspace_id: Workspace identifier + capability: Capability string + + Returns: + True if revoked, False if grant did not exist + """ + grant = CapabilityGrant.query.filter_by( + plugin_id=plugin_id, + workspace_id=workspace_id, + capability=capability, + ).first() + + if not grant: + return False + + db.session.delete(grant) + db.session.commit() + return True + + @staticmethod + def has_capability( + plugin_id: str, + workspace_id: int, + capability: str, + ) -> bool: + """Check if plugin has capability in workspace. + + Args: + plugin_id: Plugin identifier + workspace_id: Workspace identifier + capability: Capability string + + Returns: + True if plugin has capability + """ + grant = CapabilityGrant.query.filter_by( + plugin_id=plugin_id, + workspace_id=workspace_id, + capability=capability, + ).first() + return grant is not None + + @staticmethod + def list_capabilities(plugin_id: str, workspace_id: int) -> list[str]: + """List all capabilities granted to plugin in workspace. + + Args: + plugin_id: Plugin identifier + workspace_id: Workspace identifier + + Returns: + List of capability strings + """ + grants = CapabilityGrant.query.filter_by( + plugin_id=plugin_id, + workspace_id=workspace_id, + ).all() + return [grant.capability for grant in grants] + + @staticmethod + def list_granted_plugins(workspace_id: int, capability: str | None = None) -> list[str]: + """List all plugins with capabilities in a workspace. + + Args: + workspace_id: Workspace identifier + capability: Filter by specific capability (optional) + + Returns: + List of plugin IDs + """ + query = CapabilityGrant.query.filter_by(workspace_id=workspace_id) + if capability: + query = query.filter_by(capability=capability) + grants = query.all() + return list(set(grant.plugin_id for grant in grants)) + + @staticmethod + def revoke_all(plugin_id: str, workspace_id: int) -> int: + """Revoke all capabilities from a plugin in a workspace. + + Args: + plugin_id: Plugin identifier + workspace_id: Workspace identifier + + Returns: + Number of capabilities revoked + """ + grants = CapabilityGrant.query.filter_by( + plugin_id=plugin_id, + workspace_id=workspace_id, + ).all() + count = len(grants) + for grant in grants: + db.session.delete(grant) + db.session.commit() + return count + + @staticmethod + def get_role_capabilities(role: str) -> list[str]: + """Get capabilities for a workspace member role. + + Args: + role: Workspace member role (e.g., "viewer", "contributor", "admin") + + Returns: + List of capability strings available for the role + """ + capabilities = ROLE_CAPABILITY_MAPPING.get(role, []) + return [cap.value for cap in capabilities] + + +def validate_plugin_capability( + plugin_id: str, + workspace_id: int, + capability: str, +) -> tuple[bool, str]: + """Validate that a plugin has been granted a capability. + + Args: + plugin_id: Plugin identifier + workspace_id: Workspace identifier + capability: Capability to check + + Returns: + (is_valid, error_message or empty string) + """ + # Validate capability string format + if Capability.from_string(capability) is None: + return (False, f"Invalid capability: {capability}") + + # Check if grant exists + if not CapabilityStore.has_capability(plugin_id, workspace_id, capability): + return (False, f"Plugin {plugin_id} does not have capability {capability} in workspace {workspace_id}") + + return (True, "") diff --git a/app/services/events.py b/app/services/events.py new file mode 100644 index 0000000..a05df1d --- /dev/null +++ b/app/services/events.py @@ -0,0 +1,286 @@ +"""Event system for plugins and application integration. + +Provides centralized event dispatch with plugin subscription support. +Failures in event handlers are isolated - one failure doesn't affect others. +""" + +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Callable + +logger = logging.getLogger(__name__) + + +# Predefined event types that plugins can subscribe to +SUPPORTED_EVENTS = { + # Project events + "project.created": "Project imported or created", + "project.updated": "Project metadata/config changed", + "project.deleted": "Project deleted", + # Review events + "review.created": "Code review started", + "review.completed": "Code review finished", + "review.finding.added": "Review finding added", + # Workspace events + "workspace.created": "Workspace created", + "workspace.member_added": "Member added to workspace", + "workspace.member_removed": "Member removed from workspace", + # GitHub events + "github.connected": "GitHub account connected", + "github.disconnected": "GitHub account disconnected", + # AI events + "ai.analysis.completed": "AI analysis complete", + # Stellar events + "stellar.analysis.completed": "Stellar project analysis complete", + "stellar.network.detected": "Stellar network detected in project", +} + + +@dataclass +class Event: + """Represents an event in the system. + + Events carry type information, data payload, and optional context + about where they occurred (workspace, user). + """ + + event_type: str + data: dict[str, Any] | None = None + timestamp: datetime | None = None + workspace_id: int | None = None + user_id: int | None = None + + def __post_init__(self) -> None: + """Normalize defaults.""" + if self.data is None: + self.data = {} + if self.timestamp is None: + self.timestamp = datetime.now(timezone.utc) + + def to_dict(self) -> dict[str, Any]: + """Convert event to dictionary for serialization.""" + return { + "event_type": self.event_type, + "data": self.data or {}, + "timestamp": self.timestamp.isoformat() if self.timestamp else None, + "workspace_id": self.workspace_id, + "user_id": self.user_id, + } + + def __repr__(self) -> str: + """String representation.""" + return f"" + + +class EventError(Exception): + """Base exception for event system.""" + + pass + + +class EventDispatcher: + """Central event dispatch and subscription management. + + The dispatcher maintains a registry of event handlers and dispatches + events to all subscribers. Handler failures are isolated to prevent + cascade effects. + """ + + def __init__(self) -> None: + """Initialize dispatcher.""" + # event_type -> list of (plugin_id, handler_func) tuples + self._subscribers: dict[str, list[tuple[str, Callable]]] = {} + + def subscribe( + self, + event_type: str, + handler: Callable, + plugin_id: str | None = None, + ) -> None: + """Subscribe to an event type. + + Args: + event_type: Event type to subscribe to (must be in SUPPORTED_EVENTS) + handler: Callable to invoke when event is dispatched + plugin_id: Optional plugin identifier for tracking/logging + + Raises: + EventError: If event type is not supported + """ + if event_type not in SUPPORTED_EVENTS: + raise EventError(f"Unsupported event type: {event_type}") + + if event_type not in self._subscribers: + self._subscribers[event_type] = [] + + self._subscribers[event_type].append((plugin_id or "unknown", handler)) + logger.debug(f"Subscribed {plugin_id or 'handler'} to {event_type}") + + def unsubscribe( + self, + event_type: str, + handler: Callable, + ) -> bool: + """Unsubscribe from an event type. + + Args: + event_type: Event type to unsubscribe from + handler: Handler function to remove + + Returns: + True if handler was found and removed, False otherwise + """ + if event_type not in self._subscribers: + return False + + original_count = len(self._subscribers[event_type]) + self._subscribers[event_type] = [ + (plugin_id, h) for plugin_id, h in self._subscribers[event_type] if h != handler + ] + return len(self._subscribers[event_type]) < original_count + + def dispatch( + self, + event: Event, + raise_on_error: bool = False, + ) -> dict[str, Any]: + """Dispatch an event to all subscribed handlers. + + Handler failures are isolated. If one handler fails, others still run. + + Args: + event: Event to dispatch + raise_on_error: If True, re-raise first handler error + + Returns: + Dictionary with dispatch results: + { + "event_type": str, + "total_handlers": int, + "successful": int, + "failed": int, + "errors": list of (plugin_id, exception) tuples + } + """ + handlers = self._subscribers.get(event.event_type, []) + results = { + "event_type": event.event_type, + "total_handlers": len(handlers), + "successful": 0, + "failed": 0, + "errors": [], + } + + for plugin_id, handler in handlers: + try: + handler(event=event) + results["successful"] += 1 + except Exception as e: + results["failed"] += 1 + results["errors"].append((plugin_id, e)) + logger.error( + f"Error dispatching {event.event_type} to plugin {plugin_id}: {e}", + exc_info=True, + ) + if raise_on_error: + raise + + return results + + def dispatch_async(self, event: Event) -> None: + """Dispatch event asynchronously (placeholder for job queue). + + Currently dispatches synchronously. In future, could use Celery, + RQ, or similar job queue for true async dispatch. + + Args: + event: Event to dispatch + """ + # TODO: Implement with job queue (Celery, RQ, etc.) + self.dispatch(event) + + def list_subscribers(self, event_type: str | None = None) -> dict[str, list[str]]: + """List all subscribers. + + Args: + event_type: Filter by event type (optional) + + Returns: + Dictionary mapping event types to list of plugin IDs + """ + if event_type: + if event_type not in self._subscribers: + return {event_type: []} + plugin_ids = [plugin_id for plugin_id, _ in self._subscribers[event_type]] + return {event_type: plugin_ids} + + result = {} + for event_t, handlers in self._subscribers.items(): + result[event_t] = [plugin_id for plugin_id, _ in handlers] + return result + + @staticmethod + def list_supported_events() -> dict[str, str]: + """List all supported event types with descriptions. + + Returns: + Dictionary mapping event type to description + """ + return SUPPORTED_EVENTS.copy() + + def get_subscription_count(self) -> int: + """Get total number of subscriptions. + + Returns: + Total count of all subscriptions + """ + return sum(len(handlers) for handlers in self._subscribers.values()) + + def __repr__(self) -> str: + """String representation.""" + return f"" + + +# Global dispatcher instance (singleton pattern) +# In production Flask app, this would be bound to app context +_dispatcher: EventDispatcher | None = None + + +def get_dispatcher() -> EventDispatcher: + """Get or create the global event dispatcher.""" + global _dispatcher + if _dispatcher is None: + _dispatcher = EventDispatcher() + return _dispatcher + + +def create_event( + event_type: str, + data: dict[str, Any] | None = None, + workspace_id: int | None = None, + user_id: int | None = None, +) -> Event: + """Factory function to create and validate an event. + + Args: + event_type: Type of event + data: Event payload data + workspace_id: Workspace context (if applicable) + user_id: User context (if applicable) + + Returns: + Event instance + + Raises: + EventError: If event type is not supported + """ + if event_type not in SUPPORTED_EVENTS: + raise EventError(f"Unsupported event type: {event_type}") + + return Event( + event_type=event_type, + data=data or {}, + workspace_id=workspace_id, + user_id=user_id, + ) diff --git a/app/services/plugins.py b/app/services/plugins.py new file mode 100644 index 0000000..b180e03 --- /dev/null +++ b/app/services/plugins.py @@ -0,0 +1,451 @@ +"""Plugin system core: manifest, registry, and lifecycle management. + +This module provides: +- PluginManifest: Parse and validate plugin manifest files +- Plugin: Runtime plugin instance +- PluginRegistry: Central registration and management +- Custom exceptions for plugin system +""" + +import importlib +import json +import logging +import re +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from types import ModuleType +from typing import Any, Callable + +logger = logging.getLogger(__name__) + + +class PluginError(Exception): + """Base exception for plugin system.""" + + pass + + +class ManifestValidationError(PluginError): + """Plugin manifest validation failed.""" + + pass + + +class PluginRegistrationError(PluginError): + """Plugin registration failed.""" + + pass + + +class PluginLoadError(PluginError): + """Plugin load failed.""" + + pass + + +@dataclass +class PluginManifest: + """Parsed and validated plugin manifest.""" + + id: str + name: str + version: str + description: str + author: str + entry_point: str + capabilities: list[str] + compatibility: str = ">=0.8.0" + permissions: list[str] | None = None + dependencies: list[str] | None = None + configuration: dict[str, Any] | None = None + + def __post_init__(self) -> None: + """Normalize defaults.""" + if self.permissions is None: + self.permissions = [] + if self.dependencies is None: + self.dependencies = [] + if self.configuration is None: + self.configuration = {} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "PluginManifest": + """Parse manifest from dictionary. + + Args: + data: Manifest dictionary + + Returns: + PluginManifest instance + + Raises: + ManifestValidationError: If manifest is invalid + """ + errors = [] + + # Validate required fields + required = ["id", "name", "version", "description", "author", "entry_point", "capabilities"] + for field in required: + if field not in data: + errors.append(f"Missing required field: {field}") + + if errors: + raise ManifestValidationError("; ".join(errors)) + + # Validate id format + if not re.match(r"^[a-z][a-z0-9_-]*$", data.get("id", "")): + errors.append("Invalid id format: must start with lowercase letter, contain only lowercase letters, numbers, hyphens, underscores") + + # Validate version format (semantic versioning) + if not _is_valid_semver(data.get("version", "")): + errors.append("Invalid version format: must be semantic version (e.g., 0.1.0)") + + # Validate entry_point format + if not re.match(r"^[a-zA-Z0-9_][a-zA-Z0-9_.:]*:[a-zA-Z_][a-zA-Z0-9_]*$", data.get("entry_point", "")): + errors.append("Invalid entry_point format: must be 'module.path:ClassName'") + + # Validate capabilities + capabilities = data.get("capabilities", []) + valid_capabilities = { + "PROJECT_READ", + "PROJECT_WRITE", + "PROJECT_DELETE", + "WORKSPACE_READ", + "WORKSPACE_WRITE", + "GITHUB_READ", + "GITHUB_WRITE", + "AI_ACCESS", + "AI_ANALYSIS", + "NOTIFICATION_CREATE", + "STELLAR_READ", + "STELLAR_WRITE", + "STELLAR_ANALYSIS", + "REVIEW_READ", + "REVIEW_CREATE", + } + if not isinstance(capabilities, list) or not capabilities: + errors.append("Capabilities must be a non-empty list") + else: + for cap in capabilities: + if cap not in valid_capabilities: + errors.append(f"Unknown capability: {cap}") + + if errors: + raise ManifestValidationError("; ".join(errors)) + + return cls( + id=data["id"], + name=data["name"], + version=data["version"], + description=data["description"], + author=data["author"], + entry_point=data["entry_point"], + capabilities=data["capabilities"], + compatibility=data.get("compatibility", ">=0.8.0"), + permissions=data.get("permissions", []), + dependencies=data.get("dependencies", []), + configuration=data.get("configuration", {}), + ) + + @classmethod + def from_file(cls, manifest_path: str | Path) -> "PluginManifest": + """Load manifest from JSON file. + + Args: + manifest_path: Path to manifest.json + + Returns: + PluginManifest instance + + Raises: + ManifestValidationError: If manifest is invalid + PluginError: If file cannot be read + """ + try: + with open(manifest_path, "r") as f: + data = json.load(f) + except json.JSONDecodeError as e: + raise PluginError(f"Invalid JSON in manifest: {e}") from e + except FileNotFoundError as e: + raise PluginError(f"Manifest file not found: {manifest_path}") from e + except IOError as e: + raise PluginError(f"Cannot read manifest file: {e}") from e + + return cls.from_dict(data) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary representation.""" + return { + "id": self.id, + "name": self.name, + "version": self.version, + "description": self.description, + "author": self.author, + "entry_point": self.entry_point, + "capabilities": self.capabilities, + "compatibility": self.compatibility, + "permissions": self.permissions or [], + "dependencies": self.dependencies or [], + "configuration": self.configuration or {}, + } + + +class Plugin: + """Runtime plugin instance.""" + + def __init__(self, manifest: PluginManifest): + """Initialize plugin. + + Args: + manifest: Parsed plugin manifest + """ + self.manifest = manifest + self.module: ModuleType | None = None + self.enabled = True + self.loaded_at: datetime | None = None + + def load(self, app_config: dict[str, Any] | None = None) -> None: + """Load plugin module dynamically. + + Args: + app_config: Application configuration for context + + Raises: + PluginLoadError: If plugin cannot be loaded + """ + try: + module_path, class_name = self.manifest.entry_point.split(":") + self.module = importlib.import_module(module_path) + if not hasattr(self.module, class_name): + raise PluginLoadError(f"Class {class_name} not found in module {module_path}") + self.loaded_at = datetime.utcnow() + logger.info(f"Loaded plugin: {self.manifest.id}") + except ImportError as e: + raise PluginLoadError(f"Cannot import plugin module: {e}") from e + except Exception as e: + raise PluginLoadError(f"Error loading plugin {self.manifest.id}: {e}") from e + + def unload(self) -> None: + """Unload plugin safely.""" + self.module = None + self.loaded_at = None + logger.info(f"Unloaded plugin: {self.manifest.id}") + + def has_capability(self, capability: str) -> bool: + """Check if plugin has capability. + + Args: + capability: Capability to check (e.g., "PROJECT_READ") + + Returns: + True if plugin has capability + """ + return capability in self.manifest.capabilities + + def get_instance(self, app: Any | None = None) -> Any: + """Get instantiated plugin class. + + Args: + app: Flask app instance for plugin context + + Returns: + Instance of plugin class + + Raises: + PluginLoadError: If plugin not loaded + """ + if not self.module: + raise PluginLoadError(f"Plugin {self.manifest.id} not loaded") + + _, class_name = self.manifest.entry_point.split(":") + plugin_class = getattr(self.module, class_name) + return plugin_class(app=app, manifest=self.manifest) if app else plugin_class(manifest=self.manifest) + + def __repr__(self) -> str: + """String representation.""" + return f"" + + +class PluginRegistry: + """Central plugin registration and management.""" + + def __init__(self) -> None: + """Initialize registry.""" + self._plugins: dict[str, Plugin] = {} + self._hooks: dict[str, list[Callable]] = {} + + def register(self, plugin: Plugin) -> None: + """Register plugin. + + Args: + plugin: Plugin to register + + Raises: + PluginRegistrationError: If plugin already registered or invalid + """ + if plugin.manifest.id in self._plugins: + raise PluginRegistrationError(f"Plugin {plugin.manifest.id} already registered") + self._plugins[plugin.manifest.id] = plugin + logger.info(f"Registered plugin: {plugin.manifest.id}") + + def discover(self, plugin_dir: str | Path) -> list[PluginManifest]: + """Discover plugins in directory. + + Args: + plugin_dir: Directory containing plugin folders + + Returns: + List of discovered manifests + """ + plugin_dir = Path(plugin_dir) + manifests = [] + + if not plugin_dir.exists(): + logger.warning(f"Plugin directory does not exist: {plugin_dir}") + return manifests + + for item in plugin_dir.iterdir(): + if not item.is_dir() or item.name.startswith("_") or item.name.startswith("."): + continue + + manifest_file = item / "manifest.json" + if not manifest_file.exists(): + logger.debug(f"No manifest found in: {item}") + continue + + try: + manifest = PluginManifest.from_file(manifest_file) + manifests.append(manifest) + logger.debug(f"Discovered plugin: {manifest.id}") + except ManifestValidationError as e: + logger.warning(f"Invalid manifest in {item}: {e}") + except PluginError as e: + logger.warning(f"Error reading manifest from {item}: {e}") + + return manifests + + def get(self, plugin_id: str) -> Plugin | None: + """Look up plugin by ID. + + Args: + plugin_id: Plugin identifier + + Returns: + Plugin instance or None + """ + return self._plugins.get(plugin_id) + + def list_all(self) -> list[Plugin]: + """List all registered plugins. + + Returns: + List of all plugins + """ + return list(self._plugins.values()) + + def list_enabled(self) -> list[Plugin]: + """List all enabled plugins. + + Returns: + List of enabled plugins + """ + return [p for p in self._plugins.values() if p.enabled] + + def enable(self, plugin_id: str) -> None: + """Enable plugin. + + Args: + plugin_id: Plugin identifier + + Raises: + PluginRegistrationError: If plugin not found + """ + plugin = self.get(plugin_id) + if not plugin: + raise PluginRegistrationError(f"Plugin not found: {plugin_id}") + plugin.enabled = True + logger.info(f"Enabled plugin: {plugin_id}") + + def disable(self, plugin_id: str) -> None: + """Disable plugin. + + Args: + plugin_id: Plugin identifier + + Raises: + PluginRegistrationError: If plugin not found + """ + plugin = self.get(plugin_id) + if not plugin: + raise PluginRegistrationError(f"Plugin not found: {plugin_id}") + plugin.enabled = False + logger.info(f"Disabled plugin: {plugin_id}") + + def validate_capability(self, plugin_id: str, capability: str) -> bool: + """Check if plugin has capability. + + Args: + plugin_id: Plugin identifier + capability: Capability to check + + Returns: + True if plugin has capability + """ + plugin = self.get(plugin_id) + if not plugin: + return False + return plugin.has_capability(capability) + + def subscribe(self, plugin_id: str, event_type: str, handler: Callable) -> None: + """Subscribe plugin to event type. + + Args: + plugin_id: Plugin identifier + event_type: Event type (e.g., "project.created") + handler: Callable to handle event + """ + if event_type not in self._hooks: + self._hooks[event_type] = [] + self._hooks[event_type].append((plugin_id, handler)) + logger.debug(f"Subscribed plugin {plugin_id} to {event_type}") + + def dispatch(self, event_type: str, data: dict[str, Any] | None = None) -> None: + """Dispatch event to subscribed plugins. + + Failures are isolated - one plugin failure does not affect others. + + Args: + event_type: Event type (e.g., "project.created") + data: Event data dictionary + """ + if not data: + data = {} + + handlers = self._hooks.get(event_type, []) + for plugin_id, handler in handlers: + try: + handler(event_type=event_type, data=data) + except Exception as e: + logger.error(f"Error dispatching event {event_type} for plugin {plugin_id}: {e}", exc_info=True) + + def __len__(self) -> int: + """Number of registered plugins.""" + return len(self._plugins) + + def __repr__(self) -> str: + """String representation.""" + return f"" + + +def _is_valid_semver(version: str) -> bool: + """Check if version string is valid semantic version. + + Args: + version: Version string to check + + Returns: + True if valid semver + """ + pattern = r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d?)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$" + return bool(re.match(pattern, version)) diff --git a/app/services/stellar.py b/app/services/stellar.py new file mode 100644 index 0000000..614b7c8 --- /dev/null +++ b/app/services/stellar.py @@ -0,0 +1,229 @@ +"""Stellar blockchain network integration and configuration. + +Supports Stellar public networks (mainnet, testnet, futurenet) and custom +networks. Provides network detection, configuration, and metadata. +""" + +from dataclasses import dataclass +from enum import Enum +from typing import Optional + + +class StellarNetwork(str, Enum): + """Supported Stellar networks.""" + + MAINNET = "mainnet" + TESTNET = "testnet" + FUTURENET = "futurenet" + CUSTOM = "custom" + + +class StellarNetworkMode(str, Enum): + """Stellar development mode.""" + + DEVELOPMENT = "development" # Local/private development + TESTING = "testing" # Testnet + PRODUCTION = "production" # Mainnet + + +@dataclass +class NetworkConfig: + """Stellar network configuration and connection details.""" + + network: StellarNetwork + network_passphrase: str + horizon_url: str + rpc_url: Optional[str] = None # For Soroban smart contracts + mode: StellarNetworkMode = StellarNetworkMode.DEVELOPMENT + is_public: bool = False + chain_id: Optional[str] = None + + @staticmethod + def mainnet() -> "NetworkConfig": + """Get Stellar mainnet configuration.""" + return NetworkConfig( + network=StellarNetwork.MAINNET, + network_passphrase="Public Global Stellar Network ; September 2015", + horizon_url="https://horizon.stellar.org", + rpc_url="https://soroban-mainnet.stellar.org", + mode=StellarNetworkMode.PRODUCTION, + is_public=True, + ) + + @staticmethod + def testnet() -> "NetworkConfig": + """Get Stellar testnet configuration.""" + return NetworkConfig( + network=StellarNetwork.TESTNET, + network_passphrase="Test SDF Network ; September 2015", + horizon_url="https://horizon-testnet.stellar.org", + rpc_url="https://soroban-testnet.stellar.org", + mode=StellarNetworkMode.TESTING, + is_public=True, + ) + + @staticmethod + def futurenet() -> "NetworkConfig": + """Get Stellar futurenet configuration.""" + return NetworkConfig( + network=StellarNetwork.FUTURENET, + network_passphrase="Test SDF Future Network ; October 2022", + horizon_url="https://horizon-futurenet.stellar.org", + rpc_url="https://soroban-futurenet.stellar.org", + mode=StellarNetworkMode.TESTING, + is_public=True, + ) + + @staticmethod + def local() -> "NetworkConfig": + """Get local development network configuration.""" + return NetworkConfig( + network=StellarNetwork.CUSTOM, + network_passphrase="Standalone Network ; February 2021", + horizon_url="http://localhost:8000", + rpc_url="http://localhost:8000/soroban/rpc", + mode=StellarNetworkMode.DEVELOPMENT, + is_public=False, + ) + + def to_dict(self) -> dict: + """Convert to dictionary.""" + return { + "network": self.network.value, + "network_passphrase": self.network_passphrase, + "horizon_url": self.horizon_url, + "rpc_url": self.rpc_url, + "mode": self.mode.value, + "is_public": self.is_public, + "chain_id": self.chain_id, + } + + +class StellarAssetType(str, Enum): + """Stellar asset types.""" + + NATIVE = "native" # XLM (Lumens) + STANDARD = "standard" # Custom issued asset + LIQUIDITY_POOL_SHARE = "liquidity_pool_share" + + +@dataclass +class StellarAsset: + """Represents a Stellar asset on-chain.""" + + code: str + issuer: Optional[str] = None # None for native XLM + type: StellarAssetType = StellarAssetType.STANDARD + balance: Optional[str] = None + is_authorized: Optional[bool] = None + + def is_native(self) -> bool: + """Check if asset is native XLM.""" + return self.type == StellarAssetType.NATIVE or self.code == "XLM" + + def __repr__(self) -> str: + """String representation.""" + if self.is_native(): + return "XLM" + return f"{self.code}:{self.issuer}" if self.issuer else self.code + + +@dataclass +class StellarAccount: + """Represents a Stellar account on-chain.""" + + public_key: str + sequence: int + balances: list[StellarAsset] + flags: dict = None + signers: list[dict] = None + data_entries: dict = None + + def __post_init__(self): + """Normalize defaults.""" + if self.flags is None: + self.flags = {} + if self.signers is None: + self.signers = [] + if self.data_entries is None: + self.data_entries = {} + + def has_trustline(self, asset: StellarAsset) -> bool: + """Check if account has trustline for asset.""" + for balance in self.balances: + if ( + balance.code == asset.code + and balance.issuer == asset.issuer + ): + return True + return False + + +class StellarError(Exception): + """Base exception for Stellar integration.""" + + pass + + +class NetworkError(StellarError): + """Network connection error.""" + + pass + + +class AccountError(StellarError): + """Account-related error.""" + + pass + + +class AssetError(StellarError): + """Asset-related error.""" + + pass + + +class ContractError(StellarError): + """Smart contract error.""" + + pass + + +# Stellar blockchain properties +STELLAR_PROPERTIES = { + "base_fee_stroops": 100, # 0.00001 XLM + "base_reserve_stroops": 500_000_000, # 50 XLM per account + "transaction_timeout_seconds": 3600, + "max_tx_size_bytes": 1024 * 100, # 100KB +} + +# Common Soroban contract types +SOROBAN_CONTRACT_TYPES = { + "payment": "Payment and transfer contracts", + "token": "Stellar CAP46-6 Token contract", + "nft": "NFT/DeFi contract", + "defi": "DeFi protocol contract", + "oracles": "Price oracle contract", + "governance": "DAO/governance contract", +} + +# Stellar development tools and SDKs +STELLAR_SDKS = { + "py-stellar-base": "Python SDK", + "py-soroban-env": "Soroban Python SDK", + "stellar-sdk": "JavaScript SDK", + "js-stellar-sdk": "JavaScript SDK (full name)", + "go-stellar-base": "Go SDK", + "stellar-go": "Go SDK (alternate)", + "stellar-rs": "Rust SDK", + "stellar-java-sdk": "Java SDK", +} + +# Stellar frameworks and tools +STELLAR_TOOLS = { + "stellar-cli": "Stellar CLI for development", + "soroban": "Soroban smart contracts CLI", + "horizon": "Stellar API server", + "stellar-laboratory": "Web-based Stellar IDE", + "friendbot": "Test network faucet API", +} diff --git a/migrations/versions/b3c2d1a0f9e8_add_plugin_system_tables_for_phase_8.py b/migrations/versions/b3c2d1a0f9e8_add_plugin_system_tables_for_phase_8.py new file mode 100644 index 0000000..a03ea18 --- /dev/null +++ b/migrations/versions/b3c2d1a0f9e8_add_plugin_system_tables_for_phase_8.py @@ -0,0 +1,94 @@ +"""add plugin system tables for phase 8 + +Revision ID: b3c2d1a0f9e8 +Revises: a38bfbf71b9e +Create Date: 2026-08-27 18:30:00.000000 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'b3c2d1a0f9e8' +down_revision = 'a38bfbf71b9e' +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('plugins', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('plugin_id', sa.String(length=100), nullable=False), + sa.Column('name', sa.String(length=200), nullable=False), + sa.Column('description', sa.Text(), nullable=True), + sa.Column('version', sa.String(length=50), nullable=False), + sa.Column('author', sa.String(length=200), nullable=True), + sa.Column('entry_point', sa.String(length=500), nullable=False), + sa.Column('manifest_path', sa.String(length=2000), nullable=True), + sa.Column('enabled', sa.Boolean(), nullable=False, server_default='true'), + sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('plugin_id', name='uq_plugins_plugin_id') + ) + with op.batch_alter_table('plugins', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_plugins_enabled'), ['enabled'], unique=False) + batch_op.create_index(batch_op.f('ix_plugins_plugin_id'), ['plugin_id'], unique=True) + + op.create_table('plugin_installations', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('plugin_id', sa.Integer(), nullable=False), + sa.Column('workspace_id', sa.Integer(), nullable=False), + sa.Column('enabled', sa.Boolean(), nullable=False, server_default='true'), + sa.Column('config', sa.JSON(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(['plugin_id'], ['plugins.id'], ondelete='CASCADE'), + sa.ForeignKeyConstraint(['workspace_id'], ['workspaces.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('plugin_id', 'workspace_id', name='uq_plugin_installations_plugin_workspace') + ) + with op.batch_alter_table('plugin_installations', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_plugin_installations_plugin_id'), ['plugin_id'], unique=False) + batch_op.create_index(batch_op.f('ix_plugin_installations_workspace_id'), ['workspace_id'], unique=False) + + op.create_table('capability_grants', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('plugin_installation_id', sa.Integer(), nullable=False), + sa.Column('capability', sa.String(length=50), nullable=False), + sa.Column('granted_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('expires_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('revoked_at', sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(['plugin_installation_id'], ['plugin_installations.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('plugin_installation_id', 'capability', name='uq_capability_grants_plugin_capability') + ) + with op.batch_alter_table('capability_grants', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_capability_grants_capability'), ['capability'], unique=False) + batch_op.create_index(batch_op.f('ix_capability_grants_plugin_installation_id'), ['plugin_installation_id'], unique=False) + + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('capability_grants', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_capability_grants_plugin_installation_id')) + batch_op.drop_index(batch_op.f('ix_capability_grants_capability')) + + op.drop_table('capability_grants') + + with op.batch_alter_table('plugin_installations', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_plugin_installations_workspace_id')) + batch_op.drop_index(batch_op.f('ix_plugin_installations_plugin_id')) + + op.drop_table('plugin_installations') + + with op.batch_alter_table('plugins', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_plugins_plugin_id')) + batch_op.drop_index(batch_op.f('ix_plugins_enabled')) + + op.drop_table('plugins') + # ### end Alembic commands ### diff --git a/plugins/plugin.schema.json b/plugins/plugin.schema.json new file mode 100644 index 0000000..4a8e260 --- /dev/null +++ b/plugins/plugin.schema.json @@ -0,0 +1,107 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "AI Code Assistant Plugin Manifest", + "description": "Schema for plugin manifest.json files", + "type": "object", + "required": [ + "id", + "name", + "version", + "description", + "author", + "entry_point", + "capabilities" + ], + "additionalProperties": false, + "properties": { + "id": { + "type": "string", + "pattern": "^[a-z][a-z0-9_-]*$", + "minLength": 1, + "maxLength": 64, + "description": "Unique plugin identifier (kebab-case or snake_case)" + }, + "name": { + "type": "string", + "minLength": 1, + "maxLength": 256, + "description": "Human-readable plugin name" + }, + "version": { + "type": "string", + "pattern": "^(0|[1-9]\\d*)\\.(0|[1-9]\\d*)\\.(0|[1-9]\\d?)(?:-((?:0|[1-9]\\d*|\\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\\.(?:0|[1-9]\\d*|\\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\\+([0-9a-zA-Z-]+(?:\\.[0-9a-zA-Z-]+)*))?$", + "description": "Semantic version (e.g., 0.1.0, 1.0.0-alpha, 1.0.0+build)" + }, + "description": { + "type": "string", + "minLength": 1, + "maxLength": 2048, + "description": "Plugin purpose and capabilities" + }, + "author": { + "type": "string", + "minLength": 1, + "maxLength": 256, + "description": "Plugin author or maintainer" + }, + "entry_point": { + "type": "string", + "pattern": "^[a-zA-Z0-9_][a-zA-Z0-9_.:]*:[a-zA-Z_][a-zA-Z0-9_]*$", + "description": "Python module path and class name, e.g., 'plugins.stellar_tools.plugin:StellarPlugin'" + }, + "compatibility": { + "type": "string", + "default": ">=0.8.0", + "description": "Compatible AI Code Assistant versions (PEP 440 syntax)" + }, + "capabilities": { + "type": "array", + "items": { + "type": "string", + "enum": [ + "PROJECT_READ", + "PROJECT_WRITE", + "PROJECT_DELETE", + "WORKSPACE_READ", + "WORKSPACE_WRITE", + "GITHUB_READ", + "GITHUB_WRITE", + "AI_ACCESS", + "AI_ANALYSIS", + "NOTIFICATION_CREATE", + "STELLAR_READ", + "STELLAR_WRITE", + "STELLAR_ANALYSIS", + "REVIEW_READ", + "REVIEW_CREATE" + ] + }, + "minItems": 1, + "uniqueItems": true, + "description": "List of required capabilities" + }, + "permissions": { + "type": "array", + "items": { + "type": "string" + }, + "default": [], + "uniqueItems": true, + "description": "Requested workspace/project permissions (informational)" + }, + "dependencies": { + "type": "array", + "items": { + "type": "string" + }, + "default": [], + "uniqueItems": true, + "description": "Required Python dependencies (PEP 508 format)" + }, + "configuration": { + "type": "object", + "default": {}, + "description": "Plugin configuration schema and defaults" + } + } +} diff --git a/tests/test_capabilities.py b/tests/test_capabilities.py new file mode 100644 index 0000000..cb2f579 --- /dev/null +++ b/tests/test_capabilities.py @@ -0,0 +1,299 @@ +"""Tests for plugin capability management.""" + +import pytest + +from app.services.capabilities import ( + Capability, + CapabilityGrant, + CapabilityStore, + ROLE_CAPABILITY_MAPPING, + validate_plugin_capability, +) + + +class TestCapability: + """Test Capability enum.""" + + def test_capability_enum_values(self): + """Test that all capabilities have values.""" + for cap in Capability: + assert cap.value is not None + assert isinstance(cap.value, str) + assert ":" in cap.value # All values should be "domain:action" + + def test_capability_from_string(self): + """Test getting capability from string.""" + cap = Capability.from_string("project:read") + assert cap == Capability.PROJECT_READ + + def test_capability_from_string_invalid(self): + """Test getting invalid capability from string.""" + cap = Capability.from_string("invalid:capability") + assert cap is None + + def test_capability_all_values(self): + """Test getting all capability values.""" + values = Capability.all_values() + assert len(values) == 15 # We have 15 capabilities defined + assert "project:read" in values + assert "stellar:read" in values + assert all(":" in v for v in values) + + def test_capability_enum_count(self): + """Test that we have expected number of capabilities.""" + capabilities = [ + "PROJECT_READ", + "PROJECT_WRITE", + "PROJECT_DELETE", + "WORKSPACE_READ", + "WORKSPACE_WRITE", + "GITHUB_READ", + "GITHUB_WRITE", + "AI_ACCESS", + "AI_ANALYSIS", + "NOTIFICATION_CREATE", + "STELLAR_READ", + "STELLAR_WRITE", + "STELLAR_ANALYSIS", + "REVIEW_READ", + "REVIEW_CREATE", + ] + for cap_name in capabilities: + assert hasattr(Capability, cap_name) + + +class TestRoleCapabilityMapping: + """Test role to capability mapping.""" + + def test_all_roles_defined(self): + """Test that all roles have capability mappings.""" + roles = ["viewer", "contributor", "admin", "owner"] + for role in roles: + assert role in ROLE_CAPABILITY_MAPPING + assert len(ROLE_CAPABILITY_MAPPING[role]) > 0 + + def test_viewer_capabilities(self): + """Test viewer role has expected capabilities.""" + viewer_caps = ROLE_CAPABILITY_MAPPING["viewer"] + assert Capability.PROJECT_READ in viewer_caps + assert Capability.WORKSPACE_READ in viewer_caps + assert Capability.AI_ACCESS in viewer_caps + assert Capability.STELLAR_READ in viewer_caps + # Viewer should not have write capabilities + assert Capability.PROJECT_WRITE not in viewer_caps + assert Capability.WORKSPACE_WRITE not in viewer_caps + + def test_contributor_capabilities(self): + """Test contributor role has expected capabilities.""" + contrib_caps = ROLE_CAPABILITY_MAPPING["contributor"] + assert Capability.PROJECT_READ in contrib_caps + assert Capability.PROJECT_WRITE in contrib_caps + assert Capability.AI_ANALYSIS in contrib_caps + assert Capability.STELLAR_ANALYSIS in contrib_caps + assert Capability.REVIEW_CREATE in contrib_caps + # Contributor should not have delete/admin capabilities + assert Capability.PROJECT_DELETE not in contrib_caps + assert Capability.WORKSPACE_WRITE not in contrib_caps + + def test_admin_capabilities(self): + """Test admin role has write capabilities.""" + admin_caps = ROLE_CAPABILITY_MAPPING["admin"] + assert Capability.PROJECT_DELETE in admin_caps + assert Capability.WORKSPACE_WRITE in admin_caps + assert Capability.GITHUB_WRITE in admin_caps + assert Capability.STELLAR_WRITE in admin_caps + + def test_owner_capabilities(self): + """Test owner role has all capabilities.""" + owner_caps = ROLE_CAPABILITY_MAPPING["owner"] + for cap in Capability: + assert cap in owner_caps, f"Owner should have {cap.name}" + + def test_role_hierarchy(self): + """Test that role permissions are hierarchical.""" + viewer = set(ROLE_CAPABILITY_MAPPING["viewer"]) + contributor = set(ROLE_CAPABILITY_MAPPING["contributor"]) + admin = set(ROLE_CAPABILITY_MAPPING["admin"]) + owner = set(ROLE_CAPABILITY_MAPPING["owner"]) + + # Each higher role should have at least the permissions of lower roles + assert viewer.issubset(contributor), "Contributor should have viewer permissions" + assert contributor.issubset(admin), "Admin should have contributor permissions" + assert admin.issubset(owner), "Owner should have admin permissions" + + +class TestCapabilityStore: + """Test capability store operations (requires database).""" + + def test_grant_capability(self, app): + """Test granting a capability.""" + with app.app_context(): + grant = CapabilityStore.grant("test-plugin", 1, "project:read") + assert grant.plugin_id == "test-plugin" + assert grant.workspace_id == 1 + assert grant.capability == "project:read" + + def test_grant_duplicate_capability(self, app): + """Test granting same capability twice returns existing grant.""" + with app.app_context(): + grant1 = CapabilityStore.grant("test-plugin", 1, "project:read") + grant2 = CapabilityStore.grant("test-plugin", 1, "project:read") + assert grant1.id == grant2.id + + def test_grant_invalid_capability(self, app): + """Test granting invalid capability raises error.""" + with app.app_context(): + with pytest.raises(ValueError) as exc_info: + CapabilityStore.grant("test-plugin", 1, "invalid:capability") + assert "Invalid capability" in str(exc_info.value) + + def test_revoke_capability(self, app): + """Test revoking a capability.""" + with app.app_context(): + CapabilityStore.grant("test-plugin", 1, "project:read") + result = CapabilityStore.revoke("test-plugin", 1, "project:read") + assert result is True + assert not CapabilityStore.has_capability("test-plugin", 1, "project:read") + + def test_revoke_nonexistent_capability(self, app): + """Test revoking non-existent capability returns False.""" + with app.app_context(): + result = CapabilityStore.revoke("test-plugin", 1, "project:read") + assert result is False + + def test_has_capability(self, app): + """Test checking if plugin has capability.""" + with app.app_context(): + CapabilityStore.grant("test-plugin", 1, "project:read") + assert CapabilityStore.has_capability("test-plugin", 1, "project:read") + assert not CapabilityStore.has_capability("test-plugin", 1, "project:write") + + def test_list_capabilities(self, app): + """Test listing all capabilities for plugin in workspace.""" + with app.app_context(): + CapabilityStore.grant("test-plugin", 1, "project:read") + CapabilityStore.grant("test-plugin", 1, "project:write") + CapabilityStore.grant("test-plugin", 1, "stellar:read") + + caps = CapabilityStore.list_capabilities("test-plugin", 1) + assert len(caps) == 3 + assert "project:read" in caps + assert "project:write" in caps + assert "stellar:read" in caps + + def test_list_capabilities_empty(self, app): + """Test listing capabilities for plugin with none granted.""" + with app.app_context(): + caps = CapabilityStore.list_capabilities("nonexistent", 1) + assert caps == [] + + def test_list_granted_plugins(self, app): + """Test listing all plugins granted a capability.""" + with app.app_context(): + CapabilityStore.grant("plugin-1", 1, "project:read") + CapabilityStore.grant("plugin-2", 1, "project:read") + CapabilityStore.grant("plugin-3", 1, "project:write") + + plugins = CapabilityStore.list_granted_plugins(1, "project:read") + assert len(plugins) == 2 + assert "plugin-1" in plugins + assert "plugin-2" in plugins + assert "plugin-3" not in plugins + + def test_revoke_all_capabilities(self, app): + """Test revoking all capabilities from a plugin.""" + with app.app_context(): + CapabilityStore.grant("test-plugin", 1, "project:read") + CapabilityStore.grant("test-plugin", 1, "project:write") + CapabilityStore.grant("test-plugin", 1, "stellar:read") + + count = CapabilityStore.revoke_all("test-plugin", 1) + assert count == 3 + + caps = CapabilityStore.list_capabilities("test-plugin", 1) + assert len(caps) == 0 + + def test_get_role_capabilities(self, app): + """Test getting capabilities for a role.""" + with app.app_context(): + viewer_caps = CapabilityStore.get_role_capabilities("viewer") + assert len(viewer_caps) > 0 + assert "project:read" in viewer_caps + assert "project:write" not in viewer_caps + + admin_caps = CapabilityStore.get_role_capabilities("admin") + assert len(admin_caps) > len(viewer_caps) + assert "project:write" in admin_caps + assert "project:delete" in admin_caps + + def test_get_role_capabilities_invalid_role(self, app): + """Test getting capabilities for invalid role.""" + with app.app_context(): + caps = CapabilityStore.get_role_capabilities("invalid_role") + assert caps == [] + + +class TestCapabilityGrant: + """Test CapabilityGrant model.""" + + def test_capability_grant_to_dict(self, app): + """Test converting grant to dictionary.""" + with app.app_context(): + grant = CapabilityStore.grant("test-plugin", 1, "project:read") + result = grant.to_dict() + assert result["plugin_id"] == "test-plugin" + assert result["workspace_id"] == 1 + assert result["capability"] == "project:read" + assert "granted_at" in result + + def test_capability_grant_repr(self, app): + """Test grant string representation.""" + with app.app_context(): + grant = CapabilityStore.grant("test-plugin", 1, "project:read") + repr_str = repr(grant) + assert "test-plugin" in repr_str + assert "project:read" in repr_str + + +class TestValidatePluginCapability: + """Test capability validation function.""" + + def test_validate_valid_capability(self, app): + """Test validating a valid capability grant.""" + with app.app_context(): + CapabilityStore.grant("test-plugin", 1, "project:read") + is_valid, error = validate_plugin_capability("test-plugin", 1, "project:read") + assert is_valid is True + assert error == "" + + def test_validate_invalid_capability_format(self, app): + """Test validating invalid capability format.""" + with app.app_context(): + is_valid, error = validate_plugin_capability("test-plugin", 1, "invalid") + assert is_valid is False + assert "Invalid capability" in error + + def test_validate_capability_not_granted(self, app): + """Test validating capability that was not granted.""" + with app.app_context(): + is_valid, error = validate_plugin_capability("test-plugin", 1, "project:read") + assert is_valid is False + assert "does not have capability" in error + + def test_validate_multiple_capabilities(self, app): + """Test validating multiple capabilities for same plugin.""" + with app.app_context(): + CapabilityStore.grant("test-plugin", 1, "project:read") + CapabilityStore.grant("test-plugin", 1, "project:write") + + # One should pass + is_valid, error = validate_plugin_capability("test-plugin", 1, "project:read") + assert is_valid is True + + # Other should pass + is_valid, error = validate_plugin_capability("test-plugin", 1, "project:write") + assert is_valid is True + + # Third should fail + is_valid, error = validate_plugin_capability("test-plugin", 1, "project:delete") + assert is_valid is False diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..bce4b25 --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,427 @@ +"""Tests for the event system and event dispatcher.""" + +import pytest +from datetime import datetime + +from app.services.events import ( + Event, + EventDispatcher, + EventError, + create_event, + get_dispatcher, + SUPPORTED_EVENTS, +) + + +class TestEvent: + """Test Event dataclass.""" + + def test_create_basic_event(self): + """Test creating a basic event.""" + event = Event(event_type="project.created") + assert event.event_type == "project.created" + assert event.data == {} + assert event.timestamp is not None + assert event.workspace_id is None + assert event.user_id is None + + def test_create_event_with_data(self): + """Test creating event with data payload.""" + data = {"project_id": 42, "name": "test-project"} + event = Event(event_type="project.created", data=data) + assert event.data == data + assert event.data["project_id"] == 42 + + def test_event_with_context(self): + """Test event with workspace/user context.""" + event = Event( + event_type="workspace.member_added", + data={"member_id": 5}, + workspace_id=10, + user_id=3, + ) + assert event.workspace_id == 10 + assert event.user_id == 3 + + def test_event_timestamp_default(self): + """Test that timestamp defaults to current time.""" + event1 = Event(event_type="project.created") + event2 = Event(event_type="project.created") + # Timestamps should be close (within 1 second) + delta = (event2.timestamp - event1.timestamp).total_seconds() + assert delta < 1.0 + + def test_event_custom_timestamp(self): + """Test setting custom timestamp.""" + ts = datetime(2024, 1, 15, 10, 30, 0) + event = Event(event_type="project.created", timestamp=ts) + assert event.timestamp == ts + + def test_event_to_dict(self): + """Test converting event to dictionary.""" + event = Event( + event_type="project.created", + data={"project_id": 1}, + workspace_id=10, + user_id=5, + ) + result = event.to_dict() + assert result["event_type"] == "project.created" + assert result["data"] == {"project_id": 1} + assert result["workspace_id"] == 10 + assert result["user_id"] == 5 + assert "timestamp" in result + + def test_event_repr(self): + """Test event string representation.""" + event = Event(event_type="review.completed") + repr_str = repr(event) + assert "Event" in repr_str + assert "review.completed" in repr_str + + +class TestEventDispatcher: + """Test EventDispatcher class.""" + + def test_create_dispatcher(self): + """Test creating a dispatcher.""" + dispatcher = EventDispatcher() + assert dispatcher.get_subscription_count() == 0 + + def test_subscribe_to_event(self): + """Test subscribing to an event.""" + dispatcher = EventDispatcher() + received = [] + + def handler(event): + received.append(event) + + dispatcher.subscribe("project.created", handler, plugin_id="test-plugin") + assert dispatcher.get_subscription_count() == 1 + + def test_subscribe_invalid_event_type(self): + """Test subscribing to invalid event type raises error.""" + dispatcher = EventDispatcher() + + def handler(event): + pass + + with pytest.raises(EventError, match="Unsupported event type"): + dispatcher.subscribe("invalid.event", handler) + + def test_unsubscribe_from_event(self): + """Test unsubscribing from an event.""" + dispatcher = EventDispatcher() + + def handler(event): + pass + + dispatcher.subscribe("project.created", handler) + assert dispatcher.get_subscription_count() == 1 + + result = dispatcher.unsubscribe("project.created", handler) + assert result is True + assert dispatcher.get_subscription_count() == 0 + + def test_unsubscribe_nonexistent(self): + """Test unsubscribing nonexistent handler returns False.""" + dispatcher = EventDispatcher() + + def handler(event): + pass + + result = dispatcher.unsubscribe("project.created", handler) + assert result is False + + def test_dispatch_event(self): + """Test dispatching an event to subscribers.""" + dispatcher = EventDispatcher() + received = [] + + def handler(event): + received.append(event) + + dispatcher.subscribe("project.created", handler) + event = Event(event_type="project.created", data={"project_id": 1}) + result = dispatcher.dispatch(event) + + assert len(received) == 1 + assert received[0] == event + assert result["total_handlers"] == 1 + assert result["successful"] == 1 + assert result["failed"] == 0 + + def test_dispatch_to_multiple_subscribers(self): + """Test dispatching to multiple subscribers.""" + dispatcher = EventDispatcher() + received1 = [] + received2 = [] + + def handler1(event): + received1.append(event) + + def handler2(event): + received2.append(event) + + dispatcher.subscribe("project.created", handler1, plugin_id="plugin1") + dispatcher.subscribe("project.created", handler2, plugin_id="plugin2") + event = Event(event_type="project.created") + result = dispatcher.dispatch(event) + + assert len(received1) == 1 + assert len(received2) == 1 + assert result["successful"] == 2 + assert result["failed"] == 0 + + def test_dispatch_handler_error_isolation(self): + """Test that handler errors are isolated.""" + dispatcher = EventDispatcher() + received = [] + + def failing_handler(event): + raise ValueError("Handler failed") + + def good_handler(event): + received.append(event) + + dispatcher.subscribe("project.created", failing_handler, plugin_id="bad-plugin") + dispatcher.subscribe("project.created", good_handler, plugin_id="good-plugin") + event = Event(event_type="project.created") + result = dispatcher.dispatch(event) + + # Good handler should still be called despite failing handler + assert len(received) == 1 + assert result["successful"] == 1 + assert result["failed"] == 1 + assert len(result["errors"]) == 1 + assert result["errors"][0][0] == "bad-plugin" + assert isinstance(result["errors"][0][1], ValueError) + + def test_dispatch_raises_on_error_flag(self): + """Test raise_on_error flag re-raises first error.""" + dispatcher = EventDispatcher() + + def failing_handler(event): + raise RuntimeError("Test error") + + dispatcher.subscribe("project.created", failing_handler) + event = Event(event_type="project.created") + + with pytest.raises(RuntimeError, match="Test error"): + dispatcher.dispatch(event, raise_on_error=True) + + def test_dispatch_to_no_subscribers(self): + """Test dispatching to event with no subscribers.""" + dispatcher = EventDispatcher() + event = Event(event_type="project.created") + result = dispatcher.dispatch(event) + + assert result["total_handlers"] == 0 + assert result["successful"] == 0 + assert result["failed"] == 0 + + def test_list_subscribers_all(self): + """Test listing all subscribers.""" + dispatcher = EventDispatcher() + + def handler1(event): + pass + + def handler2(event): + pass + + def handler3(event): + pass + + dispatcher.subscribe("project.created", handler1, plugin_id="plugin1") + dispatcher.subscribe("project.created", handler2, plugin_id="plugin2") + dispatcher.subscribe("review.completed", handler3, plugin_id="plugin3") + + subscribers = dispatcher.list_subscribers() + assert "project.created" in subscribers + assert "review.completed" in subscribers + assert len(subscribers["project.created"]) == 2 + assert len(subscribers["review.completed"]) == 1 + + def test_list_subscribers_by_event_type(self): + """Test listing subscribers for specific event type.""" + dispatcher = EventDispatcher() + + def handler1(event): + pass + + def handler2(event): + pass + + dispatcher.subscribe("project.created", handler1, plugin_id="plugin1") + dispatcher.subscribe("project.created", handler2, plugin_id="plugin2") + dispatcher.subscribe("review.completed", handler1, plugin_id="plugin3") + + subscribers = dispatcher.list_subscribers("project.created") + assert len(subscribers["project.created"]) == 2 + + def test_list_subscribers_empty_event_type(self): + """Test listing subscribers for event type with no subscribers.""" + dispatcher = EventDispatcher() + subscribers = dispatcher.list_subscribers("project.created") + assert subscribers == {"project.created": []} + + def test_list_supported_events(self): + """Test listing supported event types.""" + events = EventDispatcher.list_supported_events() + assert isinstance(events, dict) + assert len(events) > 0 + assert "project.created" in events + assert "review.completed" in events + + def test_dispatcher_repr(self): + """Test dispatcher string representation.""" + dispatcher = EventDispatcher() + + def handler(event): + pass + + dispatcher.subscribe("project.created", handler) + repr_str = repr(dispatcher) + assert "EventDispatcher" in repr_str + + def test_subscription_without_plugin_id(self): + """Test subscribing without explicit plugin_id.""" + dispatcher = EventDispatcher() + + def handler(event): + pass + + dispatcher.subscribe("project.created", handler) + subscribers = dispatcher.list_subscribers("project.created") + assert "unknown" in subscribers["project.created"] + + +class TestCreateEventFactory: + """Test create_event factory function.""" + + def test_create_event_basic(self): + """Test basic event creation.""" + event = create_event("project.created") + assert event.event_type == "project.created" + assert event.data == {} + + def test_create_event_with_data(self): + """Test event creation with data.""" + data = {"project_id": 1} + event = create_event("project.created", data=data) + assert event.data == data + + def test_create_event_with_context(self): + """Test event creation with workspace/user context.""" + event = create_event( + "workspace.member_added", + workspace_id=10, + user_id=5, + ) + assert event.workspace_id == 10 + assert event.user_id == 5 + + def test_create_event_invalid_type(self): + """Test creating event with invalid type raises error.""" + with pytest.raises(EventError, match="Unsupported event type"): + create_event("invalid.event") + + +class TestGetDispatcher: + """Test global dispatcher singleton.""" + + def test_get_dispatcher_singleton(self): + """Test that get_dispatcher returns singleton.""" + # This test is tricky because dispatcher is module-level + # We'll just verify it returns an EventDispatcher + dispatcher = get_dispatcher() + assert isinstance(dispatcher, EventDispatcher) + + +class TestEventIntegration: + """Integration tests for event system.""" + + def test_full_event_workflow(self): + """Test complete event creation, subscription, and dispatch.""" + dispatcher = EventDispatcher() + results = [] + + def project_handler(event): + results.append({ + "type": "project", + "data": event.data, + }) + + def review_handler(event): + results.append({ + "type": "review", + "data": event.data, + }) + + dispatcher.subscribe("project.created", project_handler, plugin_id="core") + dispatcher.subscribe("review.completed", review_handler, plugin_id="review-plugin") + + # Dispatch project event + project_event = create_event( + "project.created", + data={"project_id": 1, "name": "test"}, + workspace_id=10, + ) + dispatcher.dispatch(project_event) + + # Dispatch review event + review_event = create_event( + "review.completed", + data={"review_id": 5}, + workspace_id=10, + ) + dispatcher.dispatch(review_event) + + assert len(results) == 2 + assert results[0]["type"] == "project" + assert results[0]["data"]["project_id"] == 1 + assert results[1]["type"] == "review" + assert results[1]["data"]["review_id"] == 5 + + def test_event_with_multiple_plugins(self): + """Test multiple plugins handling same event.""" + dispatcher = EventDispatcher() + plugin_actions = {"plugin1": 0, "plugin2": 0, "plugin3": 0} + + def make_handler(plugin_name): + def handler(event): + plugin_actions[plugin_name] += 1 + return handler + + dispatcher.subscribe("project.created", make_handler("plugin1"), plugin_id="plugin1") + dispatcher.subscribe("project.created", make_handler("plugin2"), plugin_id="plugin2") + dispatcher.subscribe("project.created", make_handler("plugin3"), plugin_id="plugin3") + + event = Event(event_type="project.created", data={"project_id": 1}) + dispatcher.dispatch(event) + + assert plugin_actions["plugin1"] == 1 + assert plugin_actions["plugin2"] == 1 + assert plugin_actions["plugin3"] == 1 + + def test_stellar_events(self): + """Test Stellar-specific events.""" + dispatcher = EventDispatcher() + stellar_events = [] + + def stellar_handler(event): + stellar_events.append(event) + + dispatcher.subscribe("stellar.network.detected", stellar_handler) + dispatcher.subscribe("stellar.analysis.completed", stellar_handler) + + event1 = create_event("stellar.network.detected", data={"network": "testnet"}) + event2 = create_event("stellar.analysis.completed", data={"findings": []}) + + dispatcher.dispatch(event1) + dispatcher.dispatch(event2) + + assert len(stellar_events) == 2 + assert stellar_events[0].event_type == "stellar.network.detected" + assert stellar_events[1].event_type == "stellar.analysis.completed" diff --git a/tests/test_plugins_manifest.py b/tests/test_plugins_manifest.py new file mode 100644 index 0000000..5b6bf31 --- /dev/null +++ b/tests/test_plugins_manifest.py @@ -0,0 +1,585 @@ +"""Tests for plugin manifest and registry.""" + +import json +import pytest +import tempfile +from pathlib import Path + +from app.services.plugins import ( + PluginManifest, + Plugin, + PluginRegistry, + PluginError, + ManifestValidationError, + PluginRegistrationError, + PluginLoadError, +) + + +class TestPluginManifest: + """Test plugin manifest parsing and validation.""" + + def test_manifest_valid_minimal(self): + """Test valid manifest with minimal fields.""" + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + manifest = PluginManifest.from_dict(data) + assert manifest.id == "test-plugin" + assert manifest.name == "Test Plugin" + assert manifest.version == "0.1.0" + assert manifest.capabilities == ["PROJECT_READ"] + assert manifest.permissions == [] + assert manifest.dependencies == [] + + def test_manifest_valid_full(self): + """Test valid manifest with all fields.""" + data = { + "id": "stellar-tools", + "name": "Stellar Developer Tools", + "version": "1.0.0", + "description": "Stellar/Soroban analysis", + "author": "AI Code Assistant Team", + "entry_point": "plugins.stellar_tools.plugin:StellarPlugin", + "compatibility": ">=0.8.0", + "capabilities": ["PROJECT_READ", "STELLAR_READ", "AI_ACCESS"], + "permissions": ["read:project:files", "read:stellar:networks"], + "dependencies": ["stellar-sdk>=11.0.0"], + "configuration": {"enabled_networks": ["testnet"]}, + } + manifest = PluginManifest.from_dict(data) + assert manifest.id == "stellar-tools" + assert manifest.name == "Stellar Developer Tools" + assert len(manifest.capabilities) == 3 + assert manifest.dependencies == ["stellar-sdk>=11.0.0"] + assert manifest.configuration["enabled_networks"] == ["testnet"] + + def test_manifest_missing_required_field(self): + """Test manifest with missing required field.""" + data = { + "id": "test-plugin", + "name": "Test Plugin", + # Missing version + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + with pytest.raises(ManifestValidationError) as exc_info: + PluginManifest.from_dict(data) + assert "Missing required field: version" in str(exc_info.value) + + def test_manifest_invalid_id_format(self): + """Test manifest with invalid id format.""" + test_cases = [ + "TestPlugin", # Uppercase not allowed + "123plugin", # Cannot start with number + "test plugin", # No spaces + "test@plugin", # No special chars + ] + for invalid_id in test_cases: + data = { + "id": invalid_id, + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + with pytest.raises(ManifestValidationError) as exc_info: + PluginManifest.from_dict(data) + assert "Invalid id format" in str(exc_info.value) + + def test_manifest_valid_id_formats(self): + """Test manifest with valid id formats.""" + valid_ids = [ + "test-plugin", + "test_plugin", + "testplugin", + "a", + "test-123-plugin", + ] + for valid_id in valid_ids: + data = { + "id": valid_id, + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + manifest = PluginManifest.from_dict(data) + assert manifest.id == valid_id + + def test_manifest_invalid_version(self): + """Test manifest with invalid semantic version.""" + invalid_versions = [ + "1", + "1.0", + "v1.0.0", + "01.02.03", + "1.0.0.0", + ] + for invalid_version in invalid_versions: + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": invalid_version, + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + with pytest.raises(ManifestValidationError) as exc_info: + PluginManifest.from_dict(data) + assert "Invalid version format" in str(exc_info.value) + + def test_manifest_valid_versions(self): + """Test manifest with valid semantic versions.""" + valid_versions = [ + "0.1.0", + "1.0.0", + "1.2.3", + "1.0.0-alpha", + "1.0.0-beta.1", + "1.0.0+build123", + "1.0.0-alpha+build", + ] + for valid_version in valid_versions: + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": valid_version, + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + manifest = PluginManifest.from_dict(data) + assert manifest.version == valid_version + + def test_manifest_invalid_entry_point(self): + """Test manifest with invalid entry_point format.""" + invalid_entry_points = [ + "module", # Missing class name + "module:", # Missing class + ":ClassName", # Missing module + "module.path@ClassName", # Invalid separator + ] + for invalid_ep in invalid_entry_points: + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": invalid_ep, + "capabilities": ["PROJECT_READ"], + } + with pytest.raises(ManifestValidationError) as exc_info: + PluginManifest.from_dict(data) + assert "Invalid entry_point format" in str(exc_info.value) + + def test_manifest_valid_entry_points(self): + """Test manifest with valid entry_point formats.""" + valid_entry_points = [ + "plugins.test:TestPlugin", + "my_plugin:MyPlugin", + "nested.modules.plugin:PluginClass", + "a:A", + ] + for valid_ep in valid_entry_points: + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": valid_ep, + "capabilities": ["PROJECT_READ"], + } + manifest = PluginManifest.from_dict(data) + assert manifest.entry_point == valid_ep + + def test_manifest_invalid_capability(self): + """Test manifest with unknown capability.""" + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["UNKNOWN_CAPABILITY"], + } + with pytest.raises(ManifestValidationError) as exc_info: + PluginManifest.from_dict(data) + assert "Unknown capability" in str(exc_info.value) + + def test_manifest_empty_capabilities(self): + """Test manifest with empty capabilities list.""" + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": [], + } + with pytest.raises(ManifestValidationError) as exc_info: + PluginManifest.from_dict(data) + assert "Capabilities must be a non-empty list" in str(exc_info.value) + + def test_manifest_valid_capabilities(self): + """Test manifest with all valid capabilities.""" + valid_caps = [ + "PROJECT_READ", + "PROJECT_WRITE", + "PROJECT_DELETE", + "WORKSPACE_READ", + "WORKSPACE_WRITE", + "GITHUB_READ", + "GITHUB_WRITE", + "AI_ACCESS", + "AI_ANALYSIS", + "NOTIFICATION_CREATE", + "STELLAR_READ", + "STELLAR_WRITE", + "STELLAR_ANALYSIS", + "REVIEW_READ", + "REVIEW_CREATE", + ] + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": valid_caps, + } + manifest = PluginManifest.from_dict(data) + assert len(manifest.capabilities) == len(valid_caps) + + def test_manifest_from_file(self): + """Test loading manifest from JSON file.""" + with tempfile.TemporaryDirectory() as tmpdir: + manifest_file = Path(tmpdir) / "manifest.json" + manifest_data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + with open(manifest_file, "w") as f: + json.dump(manifest_data, f) + + manifest = PluginManifest.from_file(manifest_file) + assert manifest.id == "test-plugin" + + def test_manifest_from_file_invalid_json(self): + """Test loading manifest with invalid JSON.""" + with tempfile.TemporaryDirectory() as tmpdir: + manifest_file = Path(tmpdir) / "manifest.json" + with open(manifest_file, "w") as f: + f.write("{invalid json") + + with pytest.raises(PluginError) as exc_info: + PluginManifest.from_file(manifest_file) + assert "Invalid JSON" in str(exc_info.value) + + def test_manifest_from_file_not_found(self): + """Test loading manifest from non-existent file.""" + with pytest.raises(PluginError) as exc_info: + PluginManifest.from_file("/nonexistent/manifest.json") + assert "Manifest file not found" in str(exc_info.value) + + def test_manifest_to_dict(self): + """Test converting manifest to dictionary.""" + data = { + "id": "test-plugin", + "name": "Test Plugin", + "version": "0.1.0", + "description": "A test plugin", + "author": "Test Author", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + "permissions": ["perm1"], + "dependencies": ["dep1"], + "configuration": {"key": "value"}, + } + manifest = PluginManifest.from_dict(data) + result = manifest.to_dict() + assert result["id"] == "test-plugin" + assert result["capabilities"] == ["PROJECT_READ"] + assert result["configuration"]["key"] == "value" + + +class TestPlugin: + """Test plugin class.""" + + def test_plugin_has_capability(self): + """Test checking if plugin has capability.""" + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ", "PROJECT_WRITE"], + ) + plugin = Plugin(manifest) + assert plugin.has_capability("PROJECT_READ") + assert plugin.has_capability("PROJECT_WRITE") + assert not plugin.has_capability("ADMIN") + + def test_plugin_initial_state(self): + """Test plugin initial state.""" + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + plugin = Plugin(manifest) + assert plugin.enabled + assert plugin.module is None + assert plugin.loaded_at is None + + def test_plugin_repr(self): + """Test plugin string representation.""" + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + plugin = Plugin(manifest) + assert "test-plugin" in repr(plugin) + assert "0.1.0" in repr(plugin) + + +class TestPluginRegistry: + """Test plugin registry.""" + + def test_registry_register_plugin(self): + """Test registering a plugin.""" + registry = PluginRegistry() + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + plugin = Plugin(manifest) + registry.register(plugin) + assert len(registry) == 1 + assert registry.get("test-plugin") == plugin + + def test_registry_register_duplicate(self): + """Test registering duplicate plugin.""" + registry = PluginRegistry() + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + plugin1 = Plugin(manifest) + plugin2 = Plugin(manifest) + registry.register(plugin1) + with pytest.raises(PluginRegistrationError) as exc_info: + registry.register(plugin2) + assert "already registered" in str(exc_info.value) + + def test_registry_get_plugin(self): + """Test getting plugin by ID.""" + registry = PluginRegistry() + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + plugin = Plugin(manifest) + registry.register(plugin) + assert registry.get("test-plugin") == plugin + assert registry.get("nonexistent") is None + + def test_registry_list_all(self): + """Test listing all plugins.""" + registry = PluginRegistry() + for i in range(3): + manifest = PluginManifest( + id=f"plugin-{i}", + name=f"Plugin {i}", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + registry.register(Plugin(manifest)) + assert len(registry.list_all()) == 3 + + def test_registry_enable_disable(self): + """Test enabling and disabling plugins.""" + registry = PluginRegistry() + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + plugin = Plugin(manifest) + registry.register(plugin) + assert plugin.enabled + registry.disable("test-plugin") + assert not plugin.enabled + registry.enable("test-plugin") + assert plugin.enabled + + def test_registry_enable_nonexistent(self): + """Test enabling non-existent plugin.""" + registry = PluginRegistry() + with pytest.raises(PluginRegistrationError): + registry.enable("nonexistent") + + def test_registry_validate_capability(self): + """Test validating plugin capability.""" + registry = PluginRegistry() + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + plugin = Plugin(manifest) + registry.register(plugin) + assert registry.validate_capability("test-plugin", "PROJECT_READ") + assert not registry.validate_capability("test-plugin", "PROJECT_WRITE") + assert not registry.validate_capability("nonexistent", "PROJECT_READ") + + def test_registry_discover_plugins(self): + """Test discovering plugins in directory.""" + with tempfile.TemporaryDirectory() as tmpdir: + # Create two plugins + for i in range(2): + plugin_dir = Path(tmpdir) / f"plugin-{i}" + plugin_dir.mkdir() + manifest_file = plugin_dir / "manifest.json" + manifest_data = { + "id": f"plugin-{i}", + "name": f"Plugin {i}", + "version": "0.1.0", + "description": "Test", + "author": "Test", + "entry_point": "plugins.test:Test", + "capabilities": ["PROJECT_READ"], + } + with open(manifest_file, "w") as f: + json.dump(manifest_data, f) + + registry = PluginRegistry() + discovered = registry.discover(tmpdir) + assert len(discovered) == 2 + + def test_registry_discover_empty_directory(self): + """Test discovering plugins in empty directory.""" + with tempfile.TemporaryDirectory() as tmpdir: + registry = PluginRegistry() + discovered = registry.discover(tmpdir) + assert len(discovered) == 0 + + def test_registry_discover_invalid_manifest(self): + """Test discovering with invalid manifest.""" + with tempfile.TemporaryDirectory() as tmpdir: + plugin_dir = Path(tmpdir) / "bad-plugin" + plugin_dir.mkdir() + manifest_file = plugin_dir / "manifest.json" + with open(manifest_file, "w") as f: + json.dump({"id": "bad"}, f) # Missing required fields + + registry = PluginRegistry() + discovered = registry.discover(tmpdir) + assert len(discovered) == 0 # Invalid manifest is skipped + + def test_registry_dispatch_event(self): + """Test dispatching events.""" + registry = PluginRegistry() + called = [] + + def handler(event_type: str, data: dict): + called.append((event_type, data)) + + registry.subscribe("test-plugin", "project.created", handler) + registry.dispatch("project.created", {"project_id": 123}) + assert len(called) == 1 + assert called[0] == ("project.created", {"project_id": 123}) + + def test_registry_dispatch_event_error_isolated(self): + """Test that event handler errors are isolated.""" + registry = PluginRegistry() + results = [] + + def bad_handler(event_type: str, data: dict): + raise ValueError("Handler error") + + def good_handler(event_type: str, data: dict): + results.append("success") + + registry.subscribe("plugin1", "project.created", bad_handler) + registry.subscribe("plugin2", "project.created", good_handler) + registry.dispatch("project.created", {}) + + # Good handler should still be called + assert results == ["success"] + + def test_registry_repr(self): + """Test registry string representation.""" + registry = PluginRegistry() + manifest = PluginManifest( + id="test-plugin", + name="Test", + version="0.1.0", + description="Test", + author="Test", + entry_point="plugins.test:Test", + capabilities=["PROJECT_READ"], + ) + registry.register(Plugin(manifest)) + assert "1 plugins" in repr(registry) diff --git a/tests/test_stellar.py b/tests/test_stellar.py new file mode 100644 index 0000000..8ed0727 --- /dev/null +++ b/tests/test_stellar.py @@ -0,0 +1,319 @@ +"""Tests for Stellar blockchain integration.""" + +import pytest + +from app.services.stellar import ( + StellarNetwork, + StellarNetworkMode, + NetworkConfig, + StellarAssetType, + StellarAsset, + StellarAccount, + StellarError, + NetworkError, + AccountError, + AssetError, + ContractError, + STELLAR_PROPERTIES, + SOROBAN_CONTRACT_TYPES, + STELLAR_SDKS, + STELLAR_TOOLS, +) + + +class TestStellarNetwork: + """Test StellarNetwork enum.""" + + def test_network_values(self): + """Test that all networks have correct values.""" + assert StellarNetwork.MAINNET.value == "mainnet" + assert StellarNetwork.TESTNET.value == "testnet" + assert StellarNetwork.FUTURENET.value == "futurenet" + assert StellarNetwork.CUSTOM.value == "custom" + + def test_network_count(self): + """Test that we have expected number of networks.""" + networks = list(StellarNetwork) + assert len(networks) == 4 + + +class TestStellarNetworkMode: + """Test StellarNetworkMode enum.""" + + def test_mode_values(self): + """Test mode values.""" + assert StellarNetworkMode.DEVELOPMENT.value == "development" + assert StellarNetworkMode.TESTING.value == "testing" + assert StellarNetworkMode.PRODUCTION.value == "production" + + def test_mode_count(self): + """Test number of modes.""" + modes = list(StellarNetworkMode) + assert len(modes) == 3 + + +class TestNetworkConfig: + """Test NetworkConfig dataclass.""" + + def test_create_custom_config(self): + """Test creating custom network config.""" + config = NetworkConfig( + network=StellarNetwork.CUSTOM, + network_passphrase="Custom Network", + horizon_url="http://localhost:8000", + ) + assert config.network == StellarNetwork.CUSTOM + assert config.network_passphrase == "Custom Network" + assert config.horizon_url == "http://localhost:8000" + assert config.mode == StellarNetworkMode.DEVELOPMENT + + def test_mainnet_preset(self): + """Test mainnet preset configuration.""" + config = NetworkConfig.mainnet() + assert config.network == StellarNetwork.MAINNET + assert config.mode == StellarNetworkMode.PRODUCTION + assert config.is_public is True + assert "stellar.org" in config.horizon_url + assert config.rpc_url is not None + + def test_testnet_preset(self): + """Test testnet preset configuration.""" + config = NetworkConfig.testnet() + assert config.network == StellarNetwork.TESTNET + assert config.mode == StellarNetworkMode.TESTING + assert config.is_public is True + assert "testnet" in config.horizon_url + assert config.rpc_url is not None + + def test_futurenet_preset(self): + """Test futurenet preset configuration.""" + config = NetworkConfig.futurenet() + assert config.network == StellarNetwork.FUTURENET + assert config.mode == StellarNetworkMode.TESTING + assert config.is_public is True + assert "futurenet" in config.horizon_url + + def test_local_preset(self): + """Test local development network.""" + config = NetworkConfig.local() + assert config.network == StellarNetwork.CUSTOM + assert config.mode == StellarNetworkMode.DEVELOPMENT + assert config.is_public is False + assert "localhost" in config.horizon_url + + def test_config_to_dict(self): + """Test converting config to dictionary.""" + config = NetworkConfig.mainnet() + result = config.to_dict() + assert result["network"] == "mainnet" + assert result["mode"] == "production" + assert result["is_public"] is True + assert "horizon_url" in result + assert "rpc_url" in result + + +class TestStellarAssetType: + """Test StellarAssetType enum.""" + + def test_asset_types(self): + """Test asset type values.""" + assert StellarAssetType.NATIVE.value == "native" + assert StellarAssetType.STANDARD.value == "standard" + assert StellarAssetType.LIQUIDITY_POOL_SHARE.value == "liquidity_pool_share" + + +class TestStellarAsset: + """Test StellarAsset dataclass.""" + + def test_create_native_asset(self): + """Test creating native XLM asset.""" + asset = StellarAsset(code="XLM", type=StellarAssetType.NATIVE) + assert asset.is_native() is True + assert asset.code == "XLM" + assert asset.issuer is None + + def test_create_standard_asset(self): + """Test creating standard asset.""" + issuer = "GBUQWP3BOUZX34ULNQG23RQ6F4BWFJXUR3CEEVNQT4TS4VJJBTCYL444" + asset = StellarAsset( + code="USDC", + issuer=issuer, + type=StellarAssetType.STANDARD, + ) + assert asset.is_native() is False + assert asset.code == "USDC" + assert asset.issuer == issuer + + def test_asset_repr_native(self): + """Test native asset representation.""" + asset = StellarAsset(code="XLM", type=StellarAssetType.NATIVE) + assert repr(asset) == "XLM" + + def test_asset_repr_standard(self): + """Test standard asset representation.""" + issuer = "GBUQWP3BOUZX34ULNQG23RQ6F4BWFJXUR3CEEVNQT4TS4VJJBTCYL444" + asset = StellarAsset(code="USDC", issuer=issuer) + assert repr(asset) == f"USDC:{issuer}" + + def test_asset_repr_no_issuer(self): + """Test asset without issuer.""" + asset = StellarAsset(code="FOO") + assert repr(asset) == "FOO" + + +class TestStellarAccount: + """Test StellarAccount dataclass.""" + + def test_create_account(self): + """Test creating a Stellar account.""" + public_key = "GBUQWP3BOUZX34ULNQG23RQ6F4BWFJXUR3CEEVNQT4TS4VJJBTCYL444" + account = StellarAccount( + public_key=public_key, + sequence=1, + balances=[StellarAsset(code="XLM", type=StellarAssetType.NATIVE)], + ) + assert account.public_key == public_key + assert account.sequence == 1 + assert len(account.balances) == 1 + assert account.flags == {} + assert account.signers == [] + + def test_account_has_trustline(self): + """Test checking trustline for asset.""" + usdc_asset = StellarAsset( + code="USDC", + issuer="GBUQWP3BOUZX34ULNQG23RQ6F4BWFJXUR3CEEVNQT4TS4VJJBTCYL444", + ) + account = StellarAccount( + public_key="GBUQWP3BOUZX34ULNQG23RQ6F4BWFJXUR3CEEVNQT4TS4VJJBTCYL444", + sequence=1, + balances=[usdc_asset], + ) + assert account.has_trustline(usdc_asset) is True + + def test_account_no_trustline(self): + """Test account without trustline.""" + account = StellarAccount( + public_key="GBUQWP3BOUZX34ULNQG23RQ6F4BWFJXUR3CEEVNQT4TS4VJJBTCYL444", + sequence=1, + balances=[], + ) + unknown_asset = StellarAsset(code="UNKNOWN") + assert account.has_trustline(unknown_asset) is False + + +class TestStellarExceptions: + """Test Stellar exception hierarchy.""" + + def test_exception_inheritance(self): + """Test exception inheritance chain.""" + assert issubclass(NetworkError, StellarError) + assert issubclass(AccountError, StellarError) + assert issubclass(AssetError, StellarError) + assert issubclass(ContractError, StellarError) + + def test_raise_network_error(self): + """Test raising network error.""" + with pytest.raises(NetworkError): + raise NetworkError("Connection failed") + + def test_raise_account_error(self): + """Test raising account error.""" + with pytest.raises(AccountError): + raise AccountError("Account not found") + + def test_raise_stellar_error(self): + """Test catching base exception.""" + with pytest.raises(StellarError): + raise ContractError("Contract error") + + +class TestStellarProperties: + """Test Stellar blockchain properties.""" + + def test_stellar_properties_exist(self): + """Test that blockchain properties are defined.""" + assert "base_fee_stroops" in STELLAR_PROPERTIES + assert "base_reserve_stroops" in STELLAR_PROPERTIES + assert "transaction_timeout_seconds" in STELLAR_PROPERTIES + assert "max_tx_size_bytes" in STELLAR_PROPERTIES + + def test_stellar_properties_values(self): + """Test property values make sense.""" + assert STELLAR_PROPERTIES["base_fee_stroops"] == 100 + assert STELLAR_PROPERTIES["base_reserve_stroops"] == 500_000_000 + assert STELLAR_PROPERTIES["transaction_timeout_seconds"] > 0 + assert STELLAR_PROPERTIES["max_tx_size_bytes"] > 0 + + +class TestSorobanContracts: + """Test Soroban contract types.""" + + def test_soroban_contract_types(self): + """Test Soroban contract types are defined.""" + assert "payment" in SOROBAN_CONTRACT_TYPES + assert "token" in SOROBAN_CONTRACT_TYPES + assert "nft" in SOROBAN_CONTRACT_TYPES + assert "defi" in SOROBAN_CONTRACT_TYPES + assert "oracles" in SOROBAN_CONTRACT_TYPES + assert "governance" in SOROBAN_CONTRACT_TYPES + + def test_contract_type_descriptions(self): + """Test contract descriptions are non-empty.""" + for contract_type, description in SOROBAN_CONTRACT_TYPES.items(): + assert len(description) > 0 + assert isinstance(contract_type, str) + assert isinstance(description, str) + + +class TestStellarSDKs: + """Test Stellar SDK definitions.""" + + def test_stellar_sdks_exist(self): + """Test that SDK definitions exist.""" + assert len(STELLAR_SDKS) > 0 + assert "py-stellar-base" in STELLAR_SDKS + assert "stellar-sdk" in STELLAR_SDKS + + def test_sdk_descriptions(self): + """Test SDK descriptions.""" + for sdk, description in STELLAR_SDKS.items(): + assert len(description) > 0 + assert "SDK" in description or "SDK" in description.lower() + + +class TestStellarTools: + """Test Stellar tools definitions.""" + + def test_stellar_tools_exist(self): + """Test that tool definitions exist.""" + assert len(STELLAR_TOOLS) > 0 + assert "stellar-cli" in STELLAR_TOOLS + assert "soroban" in STELLAR_TOOLS + + def test_tool_descriptions(self): + """Test tool descriptions.""" + for tool, description in STELLAR_TOOLS.items(): + assert len(description) > 0 + + +class TestNetworkConfigIntegration: + """Integration tests for network configurations.""" + + def test_all_public_networks_have_urls(self): + """Test that public networks have valid URLs.""" + for network_config in [ + NetworkConfig.mainnet(), + NetworkConfig.testnet(), + NetworkConfig.futurenet(), + ]: + assert network_config.is_public is True + assert network_config.horizon_url.startswith("http") + if network_config.rpc_url: + assert network_config.rpc_url.startswith("http") + + def test_local_network_development_mode(self): + """Test local network uses development mode.""" + config = NetworkConfig.local() + assert config.mode == StellarNetworkMode.DEVELOPMENT + assert config.is_public is False