diff --git a/alembic/versions/0003_plugin_system.py b/alembic/versions/0003_plugin_system.py new file mode 100644 index 0000000..cfd9674 --- /dev/null +++ b/alembic/versions/0003_plugin_system.py @@ -0,0 +1,55 @@ +"""T03: plugins + plugin_migrations tables. + +Revision ID: 0003_plugin_system +Revises: 0002_contacts_fts +Create Date: 2026-06-29 +""" + +from __future__ import annotations + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision: str = "0003_plugin_system" +down_revision: Union[str, None] = "0002_contacts_fts" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # --- plugins table (system-wide, NOT tenant-scoped) --- + op.create_table( + "plugins", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("name", sa.String(80), nullable=False, unique=True), + sa.Column("display_name", sa.String(120), nullable=False), + sa.Column("version", sa.String(40), nullable=False), + sa.Column("status", sa.String(20), nullable=False, server_default="installed"), + sa.Column("installed", sa.Boolean, nullable=False, server_default=sa.text("true")), + sa.Column("active", sa.Boolean, nullable=False, server_default=sa.text("false")), + sa.Column("config", sa.Text, nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + ) + op.create_index("ix_plugins_name", "plugins", ["name"], unique=True) + + # --- plugin_migrations table (tracks which migrations have been applied) --- + op.create_table( + "plugin_migrations", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("plugin_name", sa.String(80), nullable=False), + sa.Column("migration_file", sa.String(255), nullable=False), + sa.Column("status", sa.String(20), nullable=False, server_default="applied"), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.UniqueConstraint("plugin_name", "migration_file", name="ix_plugin_migrations_unique"), + ) + op.create_index("ix_plugin_migrations_plugin", "plugin_migrations", ["plugin_name"]) + + +def downgrade() -> None: + op.drop_table("plugin_migrations") + op.drop_table("plugins") diff --git a/app/main.py b/app/main.py index 4196000..17aeee6 100644 --- a/app/main.py +++ b/app/main.py @@ -9,14 +9,26 @@ from fastapi.middleware.cors import CORSMiddleware from app.config import get_settings from app.core.middleware import CSRFMiddleware -from app.core.db import close_engine -from app.routes import auth, users, roles, tenants, health, notifications, companies, contacts, import_export +from app.core.db import close_engine, get_engine +from app.core.service_container import get_container +from app.plugins.registry import get_registry +from app.routes import auth, users, roles, tenants, health, notifications, companies, contacts, import_export, plugins @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan: startup and shutdown.""" + # Initialize service container + container = get_container() + await container.initialize() + + # Initialize plugin registry and discover built-in plugins + registry = get_registry() + registry.initialize(get_engine(), app) + registry.discover_builtins() + yield + await close_engine() @@ -44,6 +56,7 @@ def create_app() -> FastAPI: app.include_router(companies.router) app.include_router(contacts.router) app.include_router(import_export.router) + app.include_router(plugins.router) return app diff --git a/app/models/__init__.py b/app/models/__init__.py index be5aea5..e4e4c76 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -9,10 +9,12 @@ from app.models.notification import Notification from app.models.auth import PasswordResetToken, ApiToken from app.models.company import Company from app.models.contact import Contact, CompanyContact +from app.models.plugin import Plugin, PluginMigration __all__ = [ "Tenant", "User", "UserTenant", "Role", "Session", "AuditLog", "DeletionLog", "Notification", "PasswordResetToken", "ApiToken", "Company", "Contact", "CompanyContact", + "Plugin", "PluginMigration", ] diff --git a/app/models/plugin.py b/app/models/plugin.py new file mode 100644 index 0000000..fa4aeea --- /dev/null +++ b/app/models/plugin.py @@ -0,0 +1,68 @@ +"""Plugin and PluginMigration models — tracks installed plugins and their migrations.""" + +from __future__ import annotations + +import uuid +from datetime import datetime +from typing import Any + +from sqlalchemy import String, Boolean, Text, DateTime, func, Index +from sqlalchemy.dialects.postgresql import UUID as PGUUID +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.db import Base, TimestampMixin + + +class Plugin(Base, TimestampMixin): + """Plugin record — tracks installation and activation status. + + This table is NOT tenant-scoped (plugins are system-wide, not per-tenant). + The tables *created by* plugin migrations MUST have tenant_id. + """ + + __tablename__ = "plugins" + __table_args__ = ( + Index("ix_plugins_name", "name", unique=True), + ) + + id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + ) + name: Mapped[str] = mapped_column(String(80), nullable=False, unique=True) + display_name: Mapped[str] = mapped_column(String(120), nullable=False) + version: Mapped[str] = mapped_column(String(40), nullable=False) + status: Mapped[str] = mapped_column( + String(20), nullable=False, default="installed" + ) + installed: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=True + ) + active: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False + ) + config: Mapped[dict[str, Any] | None] = mapped_column( + Text, nullable=True # JSON string for plugin configuration + ) + + # Transient attribute for response (not persisted) + # Used by uninstall to report dropped tables + # This is set dynamically by registry.uninstall() + + +class PluginMigration(Base, TimestampMixin): + """Tracks individual migration files applied for each plugin.""" + + __tablename__ = "plugin_migrations" + __table_args__ = ( + Index("ix_plugin_migrations_plugin", "plugin_name"), + Index("ix_plugin_migrations_unique", "plugin_name", "migration_file", unique=True), + ) + + id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + ) + plugin_name: Mapped[str] = mapped_column(String(80), nullable=False) + migration_file: Mapped[str] = mapped_column(String(255), nullable=False) + status: Mapped[str] = mapped_column( + String(20), nullable=False, default="applied" + ) diff --git a/app/plugins/__init__.py b/app/plugins/__init__.py new file mode 100644 index 0000000..75ff29a --- /dev/null +++ b/app/plugins/__init__.py @@ -0,0 +1,8 @@ +"""Plugin system framework for LeoCRM.""" + +from app.plugins.manifest import PluginManifest +from app.plugins.base import BasePlugin +from app.plugins.registry import get_registry +from app.plugins.migration_runner import MigrationRunner + +__all__ = ["PluginManifest", "BasePlugin", "get_registry", "MigrationRunner"] diff --git a/app/plugins/base.py b/app/plugins/base.py new file mode 100644 index 0000000..d08bf48 --- /dev/null +++ b/app/plugins/base.py @@ -0,0 +1,125 @@ +"""Base plugin abstract class with lifecycle hooks.""" + +from __future__ import annotations + +from abc import ABC +from typing import TYPE_CHECKING, Any + +from fastapi import APIRouter + +from app.plugins.manifest import PluginManifest + +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession + from app.core.event_bus import EventBus + from app.core.service_container import ServiceContainer + + +class BasePlugin(ABC): + """Abstract base class for all LeoCRM plugins. + + Subclasses must define a ``manifest`` class attribute (PluginManifest) + and implement the lifecycle hooks. + """ + + manifest: PluginManifest + + def __init__(self) -> None: + if not hasattr(self, "manifest") or self.manifest is None: + raise ValueError(f"{self.__class__.__name__} must define a 'manifest' attribute") + self._event_handlers: dict[str, Any] = {} + self._routers: list[APIRouter] = [] + + # ─── Lifecycle Hooks ─── + + async def on_install(self, db: AsyncSession, service_container: ServiceContainer) -> None: + """Called when the plugin is installed (after migrations are run). + + Override to perform seed data or initial setup. + """ + return None + + async def on_activate( + self, db: AsyncSession, service_container: ServiceContainer, event_bus: EventBus + ) -> None: + """Called when the plugin is activated. + + Override to register event listeners and prepare runtime state. + Default implementation subscribes to events listed in the manifest. + """ + for event_name in self.manifest.events: + handler = self._make_event_handler(event_name) + self._event_handlers[event_name] = handler + event_bus.subscribe(event_name, handler) + + async def on_deactivate( + self, db: AsyncSession, service_container: ServiceContainer, event_bus: EventBus + ) -> None: + """Called when the plugin is deactivated. + + Override to clean up runtime state. Default implementation unsubscribes + all event listeners that were registered during activation. + """ + for event_name, handler in self._event_handlers.items(): + event_bus.unsubscribe(event_name, handler) + self._event_handlers.clear() + + async def on_uninstall(self, db: AsyncSession, service_container: ServiceContainer) -> None: + """Called when the plugin is uninstalled (before data tables are dropped). + + Override to clean up any external resources. + """ + return None + + # ─── Route Registration ─── + + def get_routes(self) -> list[APIRouter]: + """Return list of APIRouter instances to mount on the FastAPI app. + + Default implementation loads routers from manifest route definitions + by importing the module and getting the specified attribute. + """ + if self._routers: + return self._routers + + routers: list[APIRouter] = [] + for route_def in self.manifest.routes: + import importlib + module = importlib.import_module(route_def.module) + router: APIRouter = getattr(module, route_def.router_attr) + routers.append(router) + self._routers = routers + return routers + + # ─── Event Handler Factory ─── + + def _make_event_handler(self, event_name: str) -> Any: + """Create an event handler coroutine for the given event name. + + Looks for a method named ``on_`` with dots replaced by underscores. + Falls back to ``on_event`` if it exists, otherwise uses a no-op handler. + """ + method_name = f"on_{event_name.replace('.', '_')}" + specific = getattr(self, method_name, None) + if specific is not None: + return specific + fallback = getattr(self, "on_event", None) + if fallback is not None: + return fallback + # Default no-op handler + async def _noop_handler(payload: dict[str, Any]) -> None: + pass + return _noop_handler + + # ─── Utility ─── + + @property + def name(self) -> str: + return self.manifest.name + + @property + def version(self) -> str: + return self.manifest.version + + def __repr__(self) -> str: + return f"<{self.__class__.__name__} name={self.manifest.name} version={self.manifest.version}>" diff --git a/app/plugins/builtins/__init__.py b/app/plugins/builtins/__init__.py new file mode 100644 index 0000000..f0f6241 --- /dev/null +++ b/app/plugins/builtins/__init__.py @@ -0,0 +1,5 @@ +"""Built-in plugins directory. + +Each module in this package that exports a BasePlugin subclass will be +discovered automatically by the plugin registry on application startup. +""" diff --git a/app/plugins/builtins/migrations/0001_bad_migration.sql b/app/plugins/builtins/migrations/0001_bad_migration.sql new file mode 100644 index 0000000..11de211 --- /dev/null +++ b/app/plugins/builtins/migrations/0001_bad_migration.sql @@ -0,0 +1,7 @@ +-- Bad migration: creates table WITHOUT tenant_id (should be rejected by validator) +CREATE TABLE IF NOT EXISTS plugin_bad_table ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + title VARCHAR(200) NOT NULL, + content TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); diff --git a/app/plugins/builtins/migrations/0001_test_plugin.sql b/app/plugins/builtins/migrations/0001_test_plugin.sql new file mode 100644 index 0000000..921a4d5 --- /dev/null +++ b/app/plugins/builtins/migrations/0001_test_plugin.sql @@ -0,0 +1,10 @@ +-- Test plugin migration: creates plugin_test_data table with tenant_id +CREATE TABLE IF NOT EXISTS plugin_test_data ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + tenant_id UUID NOT NULL, + title VARCHAR(200) NOT NULL, + content TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX IF NOT EXISTS ix_plugin_test_data_tenant ON plugin_test_data(tenant_id); diff --git a/app/plugins/builtins/test_sample.py b/app/plugins/builtins/test_sample.py new file mode 100644 index 0000000..5f9eccc --- /dev/null +++ b/app/plugins/builtins/test_sample.py @@ -0,0 +1,52 @@ +"""Sample test plugin for LeoCRM plugin system testing.""" + +from __future__ import annotations + +from typing import Any + +from app.plugins.base import BasePlugin +from app.plugins.manifest import PluginManifest + + +class TestSamplePlugin(BasePlugin): + """A sample plugin for testing the plugin lifecycle.""" + + manifest = PluginManifest( + name="test_sample", + version="1.0.0", + display_name="Test Sample Plugin", + description="A sample plugin for testing install/activate/deactivate/uninstall lifecycle.", + dependencies=[], + routes=[], + events=["company.created", "contact.created"], + migrations=["0001_test_plugin.sql"], + permissions=[], + ) + + def __init__(self) -> None: + super().__init__() + self.install_called = False + self.activate_called = False + self.deactivate_called = False + self.uninstall_called = False + self.event_log: list[dict[str, Any]] = [] + + async def on_install(self, db, service_container) -> None: + self.install_called = True + + async def on_activate(self, db, service_container, event_bus) -> None: + self.activate_called = True + await super().on_activate(db, service_container, event_bus) + + async def on_deactivate(self, db, service_container, event_bus) -> None: + self.deactivate_called = True + await super().on_deactivate(db, service_container, event_bus) + + async def on_uninstall(self, db, service_container) -> None: + self.uninstall_called = True + + async def on_company_created(self, payload: dict[str, Any]) -> None: + self.event_log.append({"event": "company.created", "payload": payload}) + + async def on_contact_created(self, payload: dict[str, Any]) -> None: + self.event_log.append({"event": "contact.created", "payload": payload}) diff --git a/app/plugins/manifest.py b/app/plugins/manifest.py new file mode 100644 index 0000000..82f3b48 --- /dev/null +++ b/app/plugins/manifest.py @@ -0,0 +1,70 @@ +"""Plugin manifest schema (Pydantic v2).""" + +from __future__ import annotations + +from pydantic import BaseModel, Field, field_validator + + +class PluginRouteDef(BaseModel): + """Route definition within a plugin manifest.""" + + path: str = Field(..., description="URL path prefix, e.g. /api/v1/plugin-mail") + module: str = Field(..., description="Dotted path to the module containing the APIRouter") + router_attr: str = Field(default="router", description="Attribute name of the APIRouter in the module") + + +class PluginManifest(BaseModel): + """Manifest describing a plugin's metadata, dependencies, and capabilities.""" + + name: str = Field(..., min_length=1, max_length=80, description="Unique plugin identifier (snake_case)") + version: str = Field(..., min_length=1, max_length=40, description="Semantic version string") + display_name: str = Field(..., min_length=1, max_length=120) + description: str = Field(default="", max_length=500) + dependencies: list[str] = Field(default_factory=list, description="Other plugin names this plugin depends on") + routes: list[PluginRouteDef] = Field(default_factory=list, description="Route definitions to register on activation") + events: list[str] = Field(default_factory=list, description="Event names this plugin listens to") + migrations: list[str] = Field(default_factory=list, description="Migration file names (ordered, e.g. 0001_initial.sql)") + permissions: list[str] = Field(default_factory=list, description="Required permissions for this plugin") + + @field_validator("name") + @classmethod + def validate_name(cls, v: str) -> str: + if not v.replace("_", "").isalnum(): + raise ValueError("Plugin name must be alphanumeric with underscores only") + return v.lower() + + model_config = {"extra": "forbid"} + + +class ManifestSchemaResponse(BaseModel): + """Response model describing the manifest schema for API consumers.""" + + fields: dict[str, dict[str, str]] + example: PluginManifest + + +# Pre-built schema documentation for GET /api/v1/plugins/manifest endpoint +MANIFEST_SCHEMA_DOC = ManifestSchemaResponse( + fields={ + "name": {"type": "str", "required": "true", "description": "Unique plugin identifier (snake_case, max 80 chars)"}, + "version": {"type": "str", "required": "true", "description": "Semantic version string"}, + "display_name": {"type": "str", "required": "true", "description": "Human-readable plugin name"}, + "description": {"type": "str", "required": "false", "description": "Plugin description (max 500 chars)"}, + "dependencies": {"type": "list[str]", "required": "false", "description": "Other plugin names required"}, + "routes": {"type": "list[PluginRouteDef]", "required": "false", "description": "Route definitions to register"}, + "events": {"type": "list[str]", "required": "false", "description": "Event names to listen to"}, + "migrations": {"type": "list[str]", "required": "false", "description": "Migration file names (ordered)"}, + "permissions": {"type": "list[str]", "required": "false", "description": "Required permissions"}, + }, + example=PluginManifest( + name="example_plugin", + version="1.0.0", + display_name="Example Plugin", + description="An example plugin demonstrating the manifest schema.", + dependencies=[], + routes=[], + events=["company.created", "contact.created"], + migrations=["0001_initial.sql"], + permissions=["companies.read"], + ), +) diff --git a/app/plugins/migration_runner.py b/app/plugins/migration_runner.py new file mode 100644 index 0000000..04f14c0 --- /dev/null +++ b/app/plugins/migration_runner.py @@ -0,0 +1,282 @@ +"""Plugin DB migration runner with tenant_id column validation.""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any + +from sqlalchemy import text, inspect +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession + +from app.models.plugin import PluginMigration + + +class MigrationValidationError(Exception): + """Raised when a plugin migration creates a table without tenant_id column.""" + + +class MigrationRunner: + """Runs plugin SQL migrations and validates that all created tables have tenant_id.""" + + def __init__(self, engine: AsyncEngine, migrations_dir: str | None = None) -> None: + self._engine = engine + self._migrations_dir = Path(migrations_dir or os.path.join(os.path.dirname(__file__), "migrations")) + + def _resolve_migration_path(self, filename: str) -> Path: + """Resolve a migration filename to an absolute path.""" + # Try migrations_dir first, then app/plugins/migrations/ + candidate = self._migrations_dir / filename + if candidate.exists(): + return candidate + # Fallback: builtins directory + builtins_dir = Path(__file__).parent / "builtins" / "migrations" + candidate2 = builtins_dir / filename + if candidate2.exists(): + return candidate2 + raise FileNotFoundError(f"Migration file not found: {filename}") + + async def run_migration( + self, + db: AsyncSession, + plugin_name: str, + migration_filename: str, + tenant_id: Any | None = None, + ) -> PluginMigration: + """Run a single SQL migration file for a plugin. + + Reads the SQL file, executes it, validates created tables have tenant_id, + and records the migration in plugin_migrations table. + + Args: + db: Async database session. + plugin_name: Name of the plugin being migrated. + migration_filename: Filename of the migration SQL file. + tenant_id: Optional tenant UUID for tenant-scoped migrations. + + Returns: + PluginMigration record. + + Raises: + FileNotFoundError: If migration file doesn't exist. + MigrationValidationError: If a created table lacks tenant_id column. + """ + migration_path = self._resolve_migration_path(migration_filename) + sql_content = migration_path.read_text(encoding="utf-8") + + # Capture existing table names before migration (static analysis of SQL) + existing_tables = await self._get_table_names_via_session(db) + + # Execute the migration SQL + # Split on semicolons but handle potential function definitions + statements = self._split_sql(sql_content) + for stmt in statements: + stmt = stmt.strip() + if stmt: + await db.execute(text(stmt)) + await db.flush() + + # Capture table names after migration (using same session connection) + new_tables = await self._get_table_names_via_session(db) + created_tables = new_tables - existing_tables + + # Validate: all newly created tables must have tenant_id column + tables_without_tenant = [] + for table_name in created_tables: + if not await self._table_has_column_via_session(db, table_name, "tenant_id"): + tables_without_tenant.append(table_name) + + if tables_without_tenant: + # Rollback the migration + await db.rollback() + raise MigrationValidationError( + f"Plugin '{plugin_name}' migration '{migration_filename}' created tables without tenant_id column: " + f"{', '.join(tables_without_tenant)}. All plugin tables must have tenant_id for multi-tenant isolation." + ) + + # Record the migration in plugin_migrations table + migration_record = PluginMigration( + plugin_name=plugin_name, + migration_file=migration_filename, + status="applied", + ) + db.add(migration_record) + await db.flush() + + return migration_record + + async def run_all_migrations( + self, + db: AsyncSession, + plugin_name: str, + migration_files: list[str], + tenant_id: Any | None = None, + ) -> list[PluginMigration]: + """Run all migration files for a plugin in order. + + Skips already-applied migrations (idempotent). + """ + # Check which migrations are already applied + from sqlalchemy import select + + result = await db.execute( + select(PluginMigration).where( + PluginMigration.plugin_name == plugin_name, + PluginMigration.status == "applied", + ) + ) + applied_files = {row.migration_file for row in result.scalars().all()} + + records: list[PluginMigration] = [] + for filename in migration_files: + if filename in applied_files: + continue # Already applied — idempotent skip + record = await self.run_migration(db, plugin_name, filename, tenant_id) + records.append(record) + + return records + + async def drop_plugin_tables(self, db: AsyncSession, plugin_name: str) -> list[str]: + """Drop all tables that were created by a plugin's migrations. + + Args: + db: Async database session. + plugin_name: Name of the plugin to remove tables for. + + Returns: + List of dropped table names. + """ + from sqlalchemy import select + + # Get all migration records for this plugin + result = await db.execute( + select(PluginMigration).where(PluginMigration.plugin_name == plugin_name) + ) + migration_records = result.scalars().all() + + dropped_tables: list[str] = [] + for record in migration_records: + migration_path = self._resolve_migration_path(record.migration_file) + sql_content = migration_path.read_text(encoding="utf-8") + + # Parse table names from CREATE TABLE statements + table_names = self._extract_table_names(sql_content) + for table_name in table_names: + try: + await db.execute(text(f'DROP TABLE IF EXISTS "{table_name}" CASCADE')) + dropped_tables.append(table_name) + except Exception: + pass # Table may already be gone + await db.flush() + + # Remove migration records + for record in migration_records: + await db.delete(record) + await db.flush() + + return dropped_tables + + async def _get_table_names_via_session(self, db: AsyncSession) -> set[str]: + """Get current table names using the session's own connection. + + This ensures we see uncommitted DDL changes within the same transaction. + """ + result = await db.execute( + text("SELECT tablename FROM pg_tables WHERE schemaname = 'public'") + ) + return {row[0] for row in result.fetchall()} + + async def _table_has_column_via_session( + self, db: AsyncSession, table_name: str, column_name: str + ) -> bool: + """Check if a table has a specific column using the session's connection.""" + result = await db.execute( + text( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = :table_name AND column_name = :column_name" + ), + {"table_name": table_name, "column_name": column_name}, + ) + return result.fetchone() is not None + + async def _get_table_names(self) -> set[str]: + """Get current table names from the database (separate connection).""" + def _get_names(sync_conn): + insp = inspect(sync_conn) + return set(insp.get_table_names()) + + async with self._engine.connect() as conn: + return await conn.run_sync(_get_names) + + async def _table_has_column(self, table_name: str, column_name: str) -> bool: + """Check if a table has a specific column (separate connection).""" + def _has_col(sync_conn): + insp = inspect(sync_conn) + if table_name not in insp.get_table_names(): + return False + columns = [col["name"] for col in insp.get_columns(table_name)] + return column_name in columns + + async with self._engine.connect() as conn: + return await conn.run_sync(_has_col) + + @staticmethod + def _split_sql(sql: str) -> list[str]: + """Split SQL into individual statements, respecting basic semicolon boundaries.""" + statements: list[str] = [] + current: list[str] = [] + in_dollar_quote = False + dollar_tag = "" + + for line in sql.splitlines(): + stripped = line.strip() + + # Handle dollar-quoted blocks (PostgreSQL function bodies) + if "$$" in stripped and not in_dollar_quote: + parts = stripped.split("$$", 1) + if len(parts) > 1: + dollar_tag = parts[0].strip() + if dollar_tag: + in_dollar_quote = True + elif stripped.count("$$") >= 2: + # Single-line $$ block + current.append(line) + if stripped.rstrip().endswith(";"): + statements.append("\n".join(current)) + current = [] + continue + else: + in_dollar_quote = True + current.append(line) + continue + + if in_dollar_quote: + current.append(line) + if "$$" in stripped: + in_dollar_quote = False + # After closing $$, check if the line ends with ; to flush statement + after_dollar = stripped.split("$$", 1)[-1] if "$$" in stripped else "" + if after_dollar.rstrip().endswith(";"): + statements.append("\n".join(current)) + current = [] + continue + + current.append(line) + if stripped.endswith(";"): + statements.append("\n".join(current)) + current = [] + + if current: + remaining = "\n".join(current).strip() + if remaining: + statements.append(remaining) + + return statements + + @staticmethod + def _extract_table_names(sql: str) -> list[str]: + """Extract table names from CREATE TABLE statements in SQL.""" + import re + # Match CREATE TABLE [IF NOT EXISTS] "table_name" or CREATE TABLE table_name + pattern = r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?["\']?(\w+)["\']?' + return re.findall(pattern, sql, re.IGNORECASE) diff --git a/app/plugins/registry.py b/app/plugins/registry.py new file mode 100644 index 0000000..eb905b0 --- /dev/null +++ b/app/plugins/registry.py @@ -0,0 +1,385 @@ +"""Plugin registry — manages discovered, installed, and active plugins.""" + +from __future__ import annotations + +import importlib +import logging +from typing import Any + +from fastapi import FastAPI +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, AsyncEngine + +from app.core.event_bus import EventBus, get_event_bus +from app.core.service_container import ServiceContainer, get_container +from app.models.plugin import Plugin as PluginModel +from app.plugins.base import BasePlugin +from app.plugins.migration_runner import MigrationRunner + +logger = logging.getLogger(__name__) + + +class PluginRegistry: + """Registry for managing plugin lifecycle. + + Maintains in-memory state for discovered plugins and their runtime instances, + backed by database records for persistent status tracking. + """ + + def __init__(self) -> None: + # name -> BasePlugin instance (discovered/loaded) + self._plugins: dict[str, BasePlugin] = {} + # name -> PluginModel DB record (installed status) + self._db_status: dict[str, PluginModel] = {} + # name -> list of mounted APIRouter references (for unregistration) + self._mounted_routers: dict[str, list[Any]] = {} + self._engine: AsyncEngine | None = None + self._event_bus: EventBus = get_event_bus() + self._container: ServiceContainer = get_container() + self._migration_runner: MigrationRunner | None = None + self._app: FastAPI | None = None + self._initialized = False + + def initialize(self, engine: AsyncEngine, app: FastAPI | None = None) -> None: + """Initialize the registry with the DB engine and optional FastAPI app.""" + self._engine = engine + self._app = app + self._migration_runner = MigrationRunner(engine) + self._initialized = True + + @property + def migration_runner(self) -> MigrationRunner: + if self._migration_runner is None: + raise RuntimeError("Registry not initialized — call initialize() first") + return self._migration_runner + + @property + def engine(self) -> AsyncEngine: + if self._engine is None: + raise RuntimeError("Registry not initialized — call initialize() first") + return self._engine + + # ─── Discovery ─── + + def discover_builtins(self) -> list[str]: + """Discover and load built-in plugins from app.plugins.builtins. + + Scans the builtins package for modules that export a BasePlugin subclass. + Returns list of discovered plugin names. + """ + discovered: list[str] = [] + try: + builtins_pkg = importlib.import_module("app.plugins.builtins") + except ImportError: + return discovered + + pkg_path = getattr(builtins_pkg, "__path__", None) + if pkg_path is None: + return discovered + + import pkgutil + for importer, modname, ispkg in pkgutil.iter_modules(pkg_path): + if modname.startswith("_"): + continue + try: + full_name = f"app.plugins.builtins.{modname}" + module = importlib.import_module(full_name) + # Look for BasePlugin subclass in module + for attr_name in dir(module): + attr = getattr(module, attr_name) + if ( + isinstance(attr, type) + and issubclass(attr, BasePlugin) + and attr is not BasePlugin + ): + instance = attr() + if instance.name not in self._plugins: + self._plugins[instance.name] = instance + discovered.append(instance.name) + logger.info(f"Discovered plugin: {instance.name} v{instance.version}") + except Exception as exc: + logger.warning(f"Failed to load builtin plugin module {modname}: {exc}") + + return discovered + + def register_plugin(self, plugin: BasePlugin) -> None: + """Manually register a plugin instance.""" + self._plugins[plugin.name] = plugin + logger.info(f"Registered plugin: {plugin.name} v{plugin.version}") + + def get_plugin(self, name: str) -> BasePlugin | None: + """Get a registered plugin instance by name.""" + return self._plugins.get(name) + + def list_discovered(self) -> list[str]: + """List all discovered plugin names.""" + return list(self._plugins.keys()) + + # ─── DB Status Sync ─── + + async def sync_db_status(self, db: AsyncSession) -> None: + """Load plugin status records from the database.""" + result = await db.execute(select(PluginModel)) + self._db_status = {row.name: row for row in result.scalars().all()} + + def get_db_status(self, name: str) -> PluginModel | None: + """Get the DB status record for a plugin.""" + return self._db_status.get(name) + + # ─── Install / Activate / Deactivate / Uninstall ─── + + async def install(self, db: AsyncSession, name: str) -> PluginModel: + """Install a plugin: run migrations and create DB record. + + Idempotent: if already installed, returns existing record. + """ + plugin = self.get_plugin(name) + if plugin is None: + raise ValueError(f"Plugin '{name}' not found in registry") + + # Check if already installed in DB + existing = await self._get_plugin_record(db, name) + if existing is not None: + return existing + + # Run migrations + if plugin.manifest.migrations: + await self.migration_runner.run_all_migrations( + db, name, plugin.manifest.migrations + ) + + # Call on_install hook + await plugin.on_install(db, self._container) + + # Create DB record + record = PluginModel( + name=name, + display_name=plugin.manifest.display_name, + version=plugin.manifest.version, + status="installed", + installed=True, + active=False, + ) + db.add(record) + await db.flush() + self._db_status[name] = record + + return record + + async def activate(self, db: AsyncSession, name: str) -> PluginModel: + """Activate a plugin: register routes, event listeners, set status=active. + + Idempotent: if already active, returns existing record without error. + """ + plugin = self.get_plugin(name) + if plugin is None: + raise ValueError(f"Plugin '{name}' not found in registry") + + record = await self._get_plugin_record(db, name) + if record is None: + raise ValueError(f"Plugin '{name}' is not installed — install first") + + # Idempotent: already active + if record.active and record.status == "active": + return record + + # Call on_activate hook (registers event listeners) + await plugin.on_activate(db, self._container, self._event_bus) + + # Register routes on FastAPI app if available + if self._app is not None: + routers = plugin.get_routes() + mounted = [] + for router in routers: + self._app.include_router(router) + mounted.append(router) + self._mounted_routers[name] = mounted + + # Update DB record + record.status = "active" + record.active = True + await db.flush() + self._db_status[name] = record + + return record + + async def deactivate(self, db: AsyncSession, name: str) -> PluginModel: + """Deactivate a plugin: unregister event listeners, set status=inactive. + + Idempotent: if already inactive, returns existing record without error. + """ + plugin = self.get_plugin(name) + if plugin is None: + raise ValueError(f"Plugin '{name}' not found in registry") + + record = await self._get_plugin_record(db, name) + if record is None: + raise ValueError(f"Plugin '{name}' is not installed") + + # Idempotent: already inactive + if not record.active and record.status == "inactive": + return record + + # Call on_deactivate hook (unregisters event listeners) + await plugin.on_deactivate(db, self._container, self._event_bus) + + # Unregister routes from FastAPI app if available + if self._app is not None and name in self._mounted_routers: + mounted = self._mounted_routers.pop(name, []) + # Collect all route paths from mounted routers for removal + paths_to_remove: set[str] = set() + for router in mounted: + for route in router.routes: + if hasattr(route, "path"): + paths_to_remove.add(route.path) + # Remove matching routes from app + self._app.router.routes = [ + r for r in self._app.router.routes + if not (hasattr(r, "path") and r.path in paths_to_remove) + ] + + # Update DB record + record.status = "inactive" + record.active = False + await db.flush() + self._db_status[name] = record + + return record + + async def uninstall( + self, db: AsyncSession, name: str, remove_data: bool = False + ) -> PluginModel: + """Uninstall a plugin: deactivate, optionally drop tables, remove DB record. + + Args: + db: Async database session. + name: Plugin name to uninstall. + remove_data: If True, drop all plugin-created tables. + + Returns: + The plugin record before deletion (for response). + + Raises: + ValueError if plugin not installed. + """ + plugin = self.get_plugin(name) + if plugin is None: + raise ValueError(f"Plugin '{name}' not found in registry") + + record = await self._get_plugin_record(db, name) + if record is None: + raise ValueError(f"Plugin '{name}' is not installed") + + # Deactivate first if active + if record.active: + await self.deactivate(db, name) + # Refetch record after deactivate + record = await self._get_plugin_record(db, name) + if record is None: + raise ValueError(f"Plugin '{name}' disappeared during uninstall") + + # Call on_uninstall hook + await plugin.on_uninstall(db, self._container) + + # Optionally drop plugin tables + dropped_tables: list[str] = [] + if remove_data: + dropped_tables = await self.migration_runner.drop_plugin_tables(db, name) + + # Remove DB record + await db.delete(record) + await db.flush() + self._db_status.pop(name, None) + + # Return a detached copy for response + record_dropped_tables = dropped_tables + record.status = "uninstalled" + record.dropped_tables = record_dropped_tables + return record + + async def list_plugins(self, db: AsyncSession) -> list[dict[str, Any]]: + """List all plugins with their current status. + + Merges discovered (in-memory) plugins with installed (DB) records. + """ + result = await db.execute(select(PluginModel)) + db_records = {row.name: row for row in result.scalars().all()} + + plugins_list: list[dict[str, Any]] = [] + for name, plugin in self._plugins.items(): + record = db_records.get(name) + if record is not None: + plugins_list.append({ + "name": name, + "display_name": record.display_name, + "version": record.version, + "status": record.status, + "installed": record.installed, + "active": record.active, + "description": plugin.manifest.description, + "dependencies": plugin.manifest.dependencies, + "events": plugin.manifest.events, + "migrations": plugin.manifest.migrations, + "permissions": plugin.manifest.permissions, + }) + else: + plugins_list.append({ + "name": name, + "display_name": plugin.manifest.display_name, + "version": plugin.version, + "status": "discovered", + "installed": False, + "active": False, + "description": plugin.manifest.description, + "dependencies": plugin.manifest.dependencies, + "events": plugin.manifest.events, + "migrations": plugin.manifest.migrations, + "permissions": plugin.manifest.permissions, + }) + + # Also include DB-only records (plugins that were installed but no longer discovered) + for name, record in db_records.items(): + if name not in self._plugins: + plugins_list.append({ + "name": name, + "display_name": record.display_name, + "version": record.version, + "status": record.status, + "installed": record.installed, + "active": record.active, + "description": "", + "dependencies": [], + "events": [], + "migrations": [], + "permissions": [], + }) + + return plugins_list + + # ─── Internal Helpers ─── + + async def _get_plugin_record(self, db: AsyncSession, name: str) -> PluginModel | None: + """Fetch a plugin record from DB by name.""" + result = await db.execute( + select(PluginModel).where(PluginModel.name == name) + ) + return result.scalar_one_or_none() + + +# Global registry instance +_registry: PluginRegistry | None = None + + +def get_registry() -> PluginRegistry: + """Get the global plugin registry.""" + global _registry + if _registry is None: + _registry = PluginRegistry() + return _registry + + +def reset_registry_for_testing() -> PluginRegistry: + """Create a fresh registry for testing.""" + global _registry + _registry = PluginRegistry() + return _registry diff --git a/app/routes/__init__.py b/app/routes/__init__.py index 945172b..8ba930e 100644 --- a/app/routes/__init__.py +++ b/app/routes/__init__.py @@ -1 +1,3 @@ """Routes package.""" + +from app.routes import auth, users, roles, tenants, health, notifications, companies, contacts, import_export, plugins diff --git a/app/routes/plugins.py b/app/routes/plugins.py new file mode 100644 index 0000000..b59aa06 --- /dev/null +++ b/app/routes/plugins.py @@ -0,0 +1,139 @@ +"""Plugin routes — list, install, activate, deactivate, uninstall, manifest schema.""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.db import get_db +from app.deps import require_admin +from app.plugins.migration_runner import MigrationValidationError +from app.services.plugin_service import get_plugin_service + +router = APIRouter(prefix="/api/v1/plugins", tags=["plugins"]) + + +@router.get("") +async def list_plugins( + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(require_admin), +): + """List all plugins with their current status (discovered, installed, active, inactive).""" + service = get_plugin_service() + plugins = await service.list_plugins(db) + return {"plugins": plugins, "total": len(plugins)} + + +@router.get("/manifest") +async def get_manifest_schema( + current_user: dict = Depends(require_admin), +): + """Get the plugin manifest schema documentation.""" + service = get_plugin_service() + return service.get_manifest_schema() + + +@router.post("/{name}/install") +async def install_plugin( + name: str, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(require_admin), +): + """Install a plugin by name. Runs migrations and creates DB record. + + Idempotent: returns 200 if already installed. + """ + import uuid as uuid_mod + service = get_plugin_service() + try: + result = await service.install_plugin( + db, name, + tenant_id=uuid_mod.UUID(current_user["tenant_id"]), + user_id=uuid_mod.UUID(current_user["user_id"]), + ) + return result + except ValueError as exc: + if "not found" in str(exc).lower(): + raise HTTPException(404, detail={"detail": str(exc), "code": "plugin_not_found"}) + raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_error"}) + except MigrationValidationError as exc: + raise HTTPException(422, detail={"detail": str(exc), "code": "migration_validation_error"}) + + +@router.post("/{name}/activate") +async def activate_plugin( + name: str, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(require_admin), +): + """Activate a plugin by name. Registers event listeners and routes. + + Idempotent: returns 200 if already active. + """ + import uuid as uuid_mod + service = get_plugin_service() + try: + result = await service.activate_plugin( + db, name, + tenant_id=uuid_mod.UUID(current_user["tenant_id"]), + user_id=uuid_mod.UUID(current_user["user_id"]), + ) + return result + except ValueError as exc: + if "not installed" in str(exc).lower(): + raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_not_installed"}) + if "not found" in str(exc).lower(): + raise HTTPException(404, detail={"detail": str(exc), "code": "plugin_not_found"}) + raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_error"}) + + +@router.post("/{name}/deactivate") +async def deactivate_plugin( + name: str, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(require_admin), +): + """Deactivate a plugin by name. Unregisters event listeners and routes. + + Idempotent: returns 200 if already inactive. + """ + import uuid as uuid_mod + service = get_plugin_service() + try: + result = await service.deactivate_plugin( + db, name, + tenant_id=uuid_mod.UUID(current_user["tenant_id"]), + user_id=uuid_mod.UUID(current_user["user_id"]), + ) + return result + except ValueError as exc: + if "not found" in str(exc).lower(): + raise HTTPException(404, detail={"detail": str(exc), "code": "plugin_not_found"}) + raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_error"}) + + +@router.delete("/{name}") +async def uninstall_plugin( + name: str, + remove_data: bool = Query(False), + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(require_admin), +): + """Uninstall a plugin. Optionally drop plugin-created tables with remove_data=true. + + Returns 200 with uninstalled status. Idempotent for already-uninstalled plugins. + """ + import uuid as uuid_mod + service = get_plugin_service() + try: + result = await service.uninstall_plugin( + db, name, + remove_data=remove_data, + tenant_id=uuid_mod.UUID(current_user["tenant_id"]), + user_id=uuid_mod.UUID(current_user["user_id"]), + ) + return result + except ValueError as exc: + if "not found" in str(exc).lower(): + raise HTTPException(404, detail={"detail": str(exc), "code": "plugin_not_found"}) + raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_error"}) diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py index e964283..0659ca0 100644 --- a/app/schemas/__init__.py +++ b/app/schemas/__init__.py @@ -1 +1,3 @@ """Pydantic schemas package.""" + +from app.schemas.plugin import PluginInfo, PluginListResponse, PluginActionResponse, PluginUninstallResponse diff --git a/app/schemas/plugin.py b/app/schemas/plugin.py new file mode 100644 index 0000000..fd45b69 --- /dev/null +++ b/app/schemas/plugin.py @@ -0,0 +1,61 @@ +"""Pydantic schemas for plugin API responses.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class PluginInfo(BaseModel): + """Plugin status information returned in list endpoint.""" + + name: str + display_name: str + version: str + status: str = Field(description="discovered | installed | active | inactive") + installed: bool = False + active: bool = False + description: str = "" + dependencies: list[str] = Field(default_factory=list) + events: list[str] = Field(default_factory=list) + migrations: list[str] = Field(default_factory=list) + permissions: list[str] = Field(default_factory=list) + + +class PluginListResponse(BaseModel): + """Response for GET /api/v1/plugins.""" + + plugins: list[PluginInfo] + total: int + + +class PluginActionResponse(BaseModel): + """Response for install/activate/deactivate/uninstall actions.""" + + name: str + display_name: str + version: str + status: str + installed: bool = False + active: bool = False + dropped_tables: list[str] = Field(default_factory=list, description="Tables dropped (uninstall with remove_data=true)") + message: str = "" + + +class PluginUninstallResponse(PluginActionResponse): + """Response for DELETE /api/v1/plugins/{name}.""" + + pass + + +class ManifestFieldDoc(BaseModel): + """Documentation for a single manifest field.""" + type: str + required: str + description: str + + +class ManifestSchemaResponse(BaseModel): + """Response for GET /api/v1/plugins/manifest.""" + + fields: dict[str, ManifestFieldDoc] + example: dict diff --git a/app/services/__init__.py b/app/services/__init__.py index f4f478c..1fe1b1a 100644 --- a/app/services/__init__.py +++ b/app/services/__init__.py @@ -1 +1,3 @@ """Service layer package.""" + +from app.services.plugin_service import PluginService, get_plugin_service diff --git a/app/services/plugin_service.py b/app/services/plugin_service.py new file mode 100644 index 0000000..6d00ca3 --- /dev/null +++ b/app/services/plugin_service.py @@ -0,0 +1,196 @@ +"""Plugin service layer — business logic for plugin lifecycle operations.""" + +from __future__ import annotations + +import logging +import uuid +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.audit import log_audit +from app.plugins.registry import PluginRegistry, get_registry +from app.plugins.migration_runner import MigrationValidationError + +logger = logging.getLogger(__name__) + + +class PluginService: + """Service layer for plugin lifecycle management. + + Wraps the PluginRegistry with audit logging and error handling. + """ + + def __init__(self, registry: PluginRegistry | None = None) -> None: + self._registry = registry or get_registry() + + @property + def registry(self) -> PluginRegistry: + return self._registry + + async def list_plugins(self, db: AsyncSession) -> list[dict[str, Any]]: + """List all discovered and installed plugins with their status.""" + return await self._registry.list_plugins(db) + + async def install_plugin( + self, + db: AsyncSession, + name: str, + tenant_id: uuid.UUID | None = None, + user_id: uuid.UUID | None = None, + ) -> dict[str, Any]: + """Install a plugin by name. + + Runs migrations, calls on_install hook, creates DB record. + Idempotent: returns existing record if already installed. + """ + try: + record = await self._registry.install(db, name) + if tenant_id and user_id: + await log_audit( + db, tenant_id, user_id, + action="plugin.install", + entity_type="plugin", + entity_id=getattr(record, "id", None), + changes={"name": name, "version": record.version}, + ) + return { + "name": record.name, + "display_name": record.display_name, + "version": record.version, + "status": record.status, + "installed": record.installed, + "active": record.active, + "message": "Plugin installed successfully", + } + except ValueError as exc: + raise ValueError(str(exc)) + except MigrationValidationError as exc: + raise MigrationValidationError(str(exc)) + + async def activate_plugin( + self, + db: AsyncSession, + name: str, + tenant_id: uuid.UUID | None = None, + user_id: uuid.UUID | None = None, + ) -> dict[str, Any]: + """Activate a plugin by name. + + Registers event listeners and routes. Idempotent. + """ + try: + record = await self._registry.activate(db, name) + was_already_active = record.active and record.status == "active" + if tenant_id and user_id: + await log_audit( + db, tenant_id, user_id, + action="plugin.activate", + entity_type="plugin", + entity_id=getattr(record, "id", None), + changes={"name": name, "was_already_active": was_already_active}, + ) + return { + "name": record.name, + "display_name": record.display_name, + "version": record.version, + "status": record.status, + "installed": record.installed, + "active": record.active, + "message": "Plugin is already active" if was_already_active else "Plugin activated successfully", + } + except ValueError as exc: + raise ValueError(str(exc)) + + async def deactivate_plugin( + self, + db: AsyncSession, + name: str, + tenant_id: uuid.UUID | None = None, + user_id: uuid.UUID | None = None, + ) -> dict[str, Any]: + """Deactivate a plugin by name. + + Unregisters event listeners and routes. Idempotent. + """ + try: + record = await self._registry.deactivate(db, name) + was_already_inactive = not record.active and record.status == "inactive" + if tenant_id and user_id: + await log_audit( + db, tenant_id, user_id, + action="plugin.deactivate", + entity_type="plugin", + entity_id=getattr(record, "id", None), + changes={"name": name, "was_already_inactive": was_already_inactive}, + ) + return { + "name": record.name, + "display_name": record.display_name, + "version": record.version, + "status": record.status, + "installed": record.installed, + "active": record.active, + "message": "Plugin is already inactive" if was_already_inactive else "Plugin deactivated successfully", + } + except ValueError as exc: + raise ValueError(str(exc)) + + async def uninstall_plugin( + self, + db: AsyncSession, + name: str, + remove_data: bool = False, + tenant_id: uuid.UUID | None = None, + user_id: uuid.UUID | None = None, + ) -> dict[str, Any]: + """Uninstall a plugin by name. + + Deactivates, calls on_uninstall hook, optionally drops tables, removes DB record. + """ + try: + record = await self._registry.uninstall(db, name, remove_data=remove_data) + dropped_tables = getattr(record, "dropped_tables", []) + if tenant_id and user_id: + await log_audit( + db, tenant_id, user_id, + action="plugin.uninstall", + entity_type="plugin", + changes={"name": name, "remove_data": remove_data, "dropped_tables": dropped_tables}, + ) + return { + "name": name, + "display_name": record.display_name, + "version": record.version, + "status": "uninstalled", + "installed": False, + "active": False, + "dropped_tables": dropped_tables, + "message": f"Plugin uninstalled{' and data tables dropped' if remove_data else ''}", + } + except ValueError as exc: + raise ValueError(str(exc)) + + def get_manifest_schema(self) -> dict[str, Any]: + """Return the manifest schema documentation.""" + from app.plugins.manifest import MANIFEST_SCHEMA_DOC + return MANIFEST_SCHEMA_DOC.model_dump() + + +# Global service instance +_service: PluginService | None = None + + +def get_plugin_service() -> PluginService: + """Get the global plugin service instance.""" + global _service + if _service is None: + _service = PluginService() + return _service + + +def reset_plugin_service_for_testing(registry: PluginRegistry | None = None) -> PluginService: + """Create a fresh plugin service for testing.""" + global _service + _service = PluginService(registry=registry) + return _service diff --git a/tests/conftest.py b/tests/conftest.py index d543cef..c4a2e6e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -32,6 +32,7 @@ from app.models.user import User, UserTenant from app.models.role import Role from app.models.company import Company from app.models.contact import Contact, CompanyContact +from app.models.plugin import Plugin, PluginMigration from app.main import create_app @@ -87,7 +88,7 @@ def clean_tables(db_setup): sync_eng = _get_sync_engine() with sync_eng.connect() as conn: # TRUNCATE all tables with CASCADE — fast and reliable isolation - conn.execute(text("TRUNCATE TABLE company_contacts, contacts, api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, companies, user_tenants, users, tenants CASCADE;")) + conn.execute(text("TRUNCATE TABLE plugin_migrations, plugins, company_contacts, contacts, api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, companies, user_tenants, users, tenants CASCADE;")) conn.commit() sync_eng.dispose() yield diff --git a/tests/test_plugins.py b/tests/test_plugins.py new file mode 100644 index 0000000..994af36 --- /dev/null +++ b/tests/test_plugins.py @@ -0,0 +1,926 @@ +"""Tests for plugin system framework — all 14 acceptance criteria.""" + +from __future__ import annotations + +import asyncio +import uuid +from typing import Any + +import pytest +import pytest_asyncio +from httpx import ASGITransport, AsyncClient +from sqlalchemy import text, select, inspect +from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, AsyncSession + +from app.core.db import Base, reset_engine_for_testing, close_engine, get_engine +from app.core.event_bus import get_event_bus +from app.core.service_container import get_container +from app.main import create_app +from app.models.plugin import Plugin as PluginModel, PluginMigration +from app.plugins.base import BasePlugin +from app.plugins.manifest import PluginManifest +from app.plugins.registry import get_registry, reset_registry_for_testing +from app.plugins.migration_runner import MigrationRunner, MigrationValidationError +from app.plugins.builtins.test_sample import TestSamplePlugin +from app.services.plugin_service import reset_plugin_service_for_testing +from tests.conftest import TEST_DB_URL, seed_tenant_and_users, login_client, ORIGIN_HEADER + + +# ─── Bad Plugin for AC11 (migration without tenant_id) ─── + +class BadMigrationPlugin(BasePlugin): + """Plugin with a migration that creates a table WITHOUT tenant_id.""" + + manifest = PluginManifest( + name="bad_migration_plugin", + version="1.0.0", + display_name="Bad Migration Plugin", + description="Plugin with invalid migration (no tenant_id) for validator testing.", + dependencies=[], + routes=[], + events=[], + migrations=["0001_bad_migration.sql"], + permissions=[], + ) + + +# ─── Fixtures ─── + +@pytest_asyncio.fixture +async def plugin_app(engine: AsyncEngine, redis_client): + """FastAPI app with plugin registry initialized for testing.""" + reset_engine_for_testing(engine) + app = create_app() + + # Reset and initialize registry (lifespan doesn't run with ASGITransport) + registry = reset_registry_for_testing() + registry.initialize(engine, app) + + # Reset service container and event bus + container = get_container() + await container.initialize() + + # Register test plugins manually (don't auto-discover to avoid side effects) + registry.register_plugin(TestSamplePlugin()) + registry.register_plugin(BadMigrationPlugin()) + + # Reset plugin service to use the new registry + reset_plugin_service_for_testing(registry) + + yield app + await close_engine() + + +@pytest_asyncio.fixture +async def plugin_client(plugin_app) -> AsyncClient: + """HTTP test client with plugin registry active.""" + transport = ASGITransport(app=plugin_app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c + + +@pytest_asyncio.fixture +async def authed_plugin_client(plugin_client: AsyncClient, db_session: AsyncSession) -> AsyncClient: + """Authenticated admin client for plugin tests.""" + seed = await seed_tenant_and_users(db_session) + await login_client(plugin_client, "admin@tenanta.com") + return plugin_client + + +@pytest_asyncio.fixture +async def db_session_for_plugins(engine: AsyncEngine) -> AsyncSession: + """DB session for plugin test verification.""" + factory = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession) + async with factory() as session: + yield session + await session.rollback() + + +# ─── AC1: GET /api/v1/plugins → 200 + list of plugins with status ─── + +@pytest.mark.asyncio +async def test_ac01_list_plugins(authed_plugin_client: AsyncClient): + """AC1: GET /api/v1/plugins returns 200 with plugin list and status.""" + resp = await authed_plugin_client.get("/api/v1/plugins", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + data = resp.json() + assert "plugins" in data + assert "total" in data + assert data["total"] >= 2 + plugin_names = [p["name"] for p in data["plugins"]] + assert "test_sample" in plugin_names + assert "bad_migration_plugin" in plugin_names + # Check status field exists + for p in data["plugins"]: + assert "status" in p + assert "installed" in p + assert "active" in p + + +# ─── AC2: POST /api/v1/plugins/{name}/install → 200, status=installed, migrations run ─── + +@pytest.mark.asyncio +async def test_ac02_install_plugin(authed_plugin_client: AsyncClient, db_session_for_plugins: AsyncSession): + """AC2: Install plugin runs migrations and sets status=installed.""" + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + data = resp.json() + assert data["name"] == "test_sample" + assert data["status"] == "installed" + assert data["installed"] is True + assert data["active"] is False + + # Verify migration table was created (tenant_id column exists) + result = await db_session_for_plugins.execute( + text("SELECT column_name FROM information_schema.columns WHERE table_name = 'plugin_test_data' AND column_name = 'tenant_id'") + ) + assert result.fetchone() is not None, "plugin_test_data table should have tenant_id column" + + +# ─── AC3: POST /api/v1/plugins/{name}/activate → 200, status=active, routes registered ─── + +@pytest.mark.asyncio +async def test_ac03_activate_plugin(authed_plugin_client: AsyncClient): + """AC3: Activate plugin sets status=active and registers event listeners.""" + # Install first + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + + # Activate + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + data = resp.json() + assert data["name"] == "test_sample" + assert data["status"] == "active" + assert data["active"] is True + + +# ─── AC4: POST /api/v1/plugins/{name}/deactivate → 200, status=inactive, routes unregistered ─── + +@pytest.mark.asyncio +async def test_ac04_deactivate_plugin(authed_plugin_client: AsyncClient): + """AC4: Deactivate plugin sets status=inactive and unregisters event listeners.""" + # Install and activate first + await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER) + + # Deactivate + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/deactivate", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + data = resp.json() + assert data["name"] == "test_sample" + assert data["status"] == "inactive" + assert data["active"] is False + + +# ─── AC5: DELETE /api/v1/plugins/{name} → 200, plugin removed ─── + +@pytest.mark.asyncio +async def test_ac05_uninstall_plugin(authed_plugin_client: AsyncClient, db_session_for_plugins: AsyncSession): + """AC5: Uninstall plugin removes DB record.""" + # Install first + await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + + # Uninstall + resp = await authed_plugin_client.delete("/api/v1/plugins/test_sample", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "uninstalled" + assert data["installed"] is False + + # Verify DB record is gone + result = await db_session_for_plugins.execute( + select(PluginModel).where(PluginModel.name == "test_sample") + ) + assert result.scalar_one_or_none() is None + + +# ─── AC6: DELETE /api/v1/plugins/{name}?remove_data=true → 200, plugin tables dropped ─── + +@pytest.mark.asyncio +async def test_ac06_uninstall_remove_data(authed_plugin_client: AsyncClient, engine: AsyncEngine): + """AC6: Uninstall with remove_data=true drops plugin tables.""" + # Install first (creates plugin_test_data table) + await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + + # Verify table exists + def _check_table(sync_conn): + insp = inspect(sync_conn) + return "plugin_test_data" in insp.get_table_names() + async with engine.connect() as conn: + table_exists_before = await conn.run_sync(_check_table) + assert table_exists_before, "plugin_test_data table should exist after install" + + # Uninstall with remove_data=true + resp = await authed_plugin_client.delete( + "/api/v1/plugins/test_sample?remove_data=true", headers=ORIGIN_HEADER + ) + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "uninstalled" + assert "plugin_test_data" in data.get("dropped_tables", []) + + # Verify table is gone + async with engine.connect() as conn: + table_exists_after = await conn.run_sync(_check_table) + assert not table_exists_after, "plugin_test_data table should be dropped after uninstall with remove_data=true" + + +# ─── AC7: GET /api/v1/plugins/manifest → 200 + manifest schema documentation ─── + +@pytest.mark.asyncio +async def test_ac07_manifest_schema(authed_plugin_client: AsyncClient): + """AC7: GET /api/v1/plugins/manifest returns schema documentation.""" + resp = await authed_plugin_client.get("/api/v1/plugins/manifest", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + data = resp.json() + assert "fields" in data + assert "example" in data + # Check required fields are documented + assert "name" in data["fields"] + assert "version" in data["fields"] + assert "display_name" in data["fields"] + assert "dependencies" in data["fields"] + assert "routes" in data["fields"] + assert "events" in data["fields"] + assert "migrations" in data["fields"] + # Check example has expected values + assert data["example"]["name"] == "example_plugin" + + +# ─── AC8: Plugin activation registers event listeners on event bus ─── + +@pytest.mark.asyncio +async def test_ac08_activation_registers_event_listeners( + authed_plugin_client: AsyncClient, db_session_for_plugins: AsyncSession +): + """AC8: Activating a plugin registers its event listeners on the event bus.""" + registry = get_registry() + plugin = registry.get_plugin("test_sample") + assert plugin is not None + + # Install and activate + await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER) + + event_bus = get_event_bus() + + # Check that event bus has handlers for the plugin's events + assert len(event_bus._handlers.get("company.created", [])) > 0 + assert len(event_bus._handlers.get("contact.created", [])) > 0 + + # Publish an event and verify the handler is called + await event_bus.publish("company.created", {"company_id": "test-123", "name": "Test Corp"}) + # Give async tasks a moment to complete + await asyncio.sleep(0.1) + + # Check that the plugin's event handler was called + assert len(plugin.event_log) > 0 + assert plugin.event_log[0]["event"] == "company.created" + + +# ─── AC9: Plugin deactivation unregisters event listeners ─── + +@pytest.mark.asyncio +async def test_ac09_deactivation_unregisters_event_listeners(authed_plugin_client: AsyncClient): + """AC9: Deactivating a plugin unregisters its event listeners from the event bus.""" + registry = get_registry() + plugin = registry.get_plugin("test_sample") + assert plugin is not None + + # Install, activate, then deactivate + await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER) + await authed_plugin_client.post("/api/v1/plugins/test_sample/deactivate", headers=ORIGIN_HEADER) + + event_bus = get_event_bus() + + # Publish an event — handler should NOT be called after deactivation + initial_log_count = len(plugin.event_log) + await event_bus.publish("company.created", {"company_id": "test-456", "name": "After Deactivate"}) + await asyncio.sleep(0.1) + + # Event log should not have grown + assert len(plugin.event_log) == initial_log_count, "Event handler should not be called after deactivation" + + # Check event bus no longer has the plugin's handlers + handlers = event_bus._handlers.get("company.created", []) + # The specific handler reference should be removed + for handler in handlers: + assert handler not in plugin._event_handlers.values() + + +# ─── AC10: Plugin migration creates tables with tenant_id column ─── + +@pytest.mark.asyncio +async def test_ac10_migration_creates_tenant_id(authed_plugin_client: AsyncClient, engine: AsyncEngine): + """AC10: Plugin migration creates tables that have tenant_id column.""" + # Install the test_sample plugin which has a migration creating plugin_test_data + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + + # Verify the table has tenant_id column + def _get_columns(sync_conn): + insp = inspect(sync_conn) + if "plugin_test_data" not in insp.get_table_names(): + return None + return [col["name"] for col in insp.get_columns("plugin_test_data")] + + async with engine.connect() as conn: + columns = await conn.run_sync(_get_columns) + + assert columns is not None, "plugin_test_data table should exist" + assert "tenant_id" in columns, "plugin_test_data table must have tenant_id column" + + +# ─── AC11: Plugin migration validator rejects tables without tenant_id ─── + +@pytest.mark.asyncio +async def test_ac11_validator_rejects_no_tenant_id(authed_plugin_client: AsyncClient): + """AC11: Migration validator rejects tables created without tenant_id column.""" + # Try to install the bad_migration_plugin — should fail with 422 + resp = await authed_plugin_client.post( + "/api/v1/plugins/bad_migration_plugin/install", headers=ORIGIN_HEADER + ) + assert resp.status_code == 422 + data = resp.json() + # detail is a dict with {detail: str, code: str} + detail = data.get("detail", {}) + if isinstance(detail, dict): + detail_text = detail.get("detail", "").lower() + detail_code = detail.get("code", "") + else: + detail_text = str(detail).lower() + detail_code = data.get("code", "") + assert "tenant_id" in detail_text or "migration_validation" in detail_code + + +# ─── AC12: Plugin DB migrations tracked in plugin_migrations table ─── + +@pytest.mark.asyncio +async def test_ac12_migrations_tracked(authed_plugin_client: AsyncClient, db_session_for_plugins: AsyncSession): + """AC12: Plugin migrations are tracked in plugin_migrations table.""" + # Install the test_sample plugin + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + + # Check plugin_migrations table has a record + result = await db_session_for_plugins.execute( + select(PluginMigration).where(PluginMigration.plugin_name == "test_sample") + ) + migrations = result.scalars().all() + assert len(migrations) > 0, "plugin_migrations table should have a record for test_sample" + assert migrations[0].migration_file == "0001_test_plugin.sql" + assert migrations[0].status == "applied" + + +# ─── AC13: Activating already-active plugin → idempotent (200, no error) ─── + +@pytest.mark.asyncio +async def test_ac13_activate_already_active(authed_plugin_client: AsyncClient): + """AC13: Activating an already-active plugin is idempotent (returns 200, no error).""" + # Install and activate + await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + assert resp.json()["status"] == "active" + + # Activate again — should be idempotent + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "active" + assert data["active"] is True + assert "already active" in data["message"].lower() + + +# ─── AC14: Deactivating inactive plugin → idempotent (200) ─── + +@pytest.mark.asyncio +async def test_ac14_deactivate_already_inactive(authed_plugin_client: AsyncClient): + """AC14: Deactivating an already-inactive plugin is idempotent (returns 200, no error).""" + # Install and activate + await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER) + await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER) + + # Deactivate + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/deactivate", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + assert resp.json()["status"] == "inactive" + + # Deactivate again — should be idempotent + resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/deactivate", headers=ORIGIN_HEADER) + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "inactive" + assert data["active"] is False + assert "already inactive" in data["message"].lower() + + +# ─── Additional Tests: Direct MigrationRunner Unit Tests ─── + +@pytest.mark.asyncio +async def test_registry_discover_builtins(engine: AsyncEngine): + """Test that registry can discover built-in plugins.""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + discovered = registry.discover_builtins() + # test_sample should be discovered from builtins package + assert "test_sample" in discovered + assert registry.get_plugin("test_sample") is not None + + +@pytest.mark.asyncio +async def test_registry_list_plugins_mixed_states(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test listing plugins in various states (discovered, installed, active, inactive).""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + registry.register_plugin(TestSamplePlugin()) + + # List before install — all discovered + plugins = await registry.list_plugins(db_session_for_plugins) + assert len(plugins) == 1 + assert plugins[0]["status"] == "discovered" + assert plugins[0]["installed"] is False + + # Install + await registry.install(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + + # List after install — installed + plugins = await registry.list_plugins(db_session_for_plugins) + assert len(plugins) == 1 + assert plugins[0]["status"] == "installed" + assert plugins[0]["installed"] is True + + # Activate + await registry.activate(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + + # List after activate — active + plugins = await registry.list_plugins(db_session_for_plugins) + assert plugins[0]["status"] == "active" + assert plugins[0]["active"] is True + + +@pytest.mark.asyncio +async def test_registry_install_idempotent(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test that installing an already-installed plugin is idempotent.""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + registry.register_plugin(TestSamplePlugin()) + + record1 = await registry.install(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + + record2 = await registry.install(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + + # Same record returned + assert record1.id == record2.id + + +@pytest.mark.asyncio +async def test_registry_not_found_errors(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test registry raises ValueError for unknown plugins.""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + + with pytest.raises(ValueError, match="not found"): + await registry.install(db_session_for_plugins, "nonexistent") + + with pytest.raises(ValueError, match="not found"): + await registry.activate(db_session_for_plugins, "nonexistent") + + with pytest.raises(ValueError, match="not found"): + await registry.deactivate(db_session_for_plugins, "nonexistent") + + with pytest.raises(ValueError, match="not found"): + await registry.uninstall(db_session_for_plugins, "nonexistent") + + +@pytest.mark.asyncio +async def test_registry_activate_without_install(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test that activating without install raises error.""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + registry.register_plugin(TestSamplePlugin()) + + with pytest.raises(ValueError, match="not installed"): + await registry.activate(db_session_for_plugins, "test_sample") + + +@pytest.mark.asyncio +async def test_migration_runner_drop_tables(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test MigrationRunner.drop_plugin_tables removes tables and migration records.""" + runner = MigrationRunner(engine) + + # Run migration first + await runner.run_migration(db_session_for_plugins, "drop_test", "0001_test_plugin.sql") + await db_session_for_plugins.commit() + + # Verify table exists + result = await db_session_for_plugins.execute( + text("SELECT tablename FROM pg_tables WHERE tablename = 'plugin_test_data'") + ) + assert result.fetchone() is not None + + # Drop tables + dropped = await runner.drop_plugin_tables(db_session_for_plugins, "drop_test") + await db_session_for_plugins.commit() + + assert "plugin_test_data" in dropped + + # Verify table is gone + result = await db_session_for_plugins.execute( + text("SELECT tablename FROM pg_tables WHERE tablename = 'plugin_test_data'") + ) + assert result.fetchone() is None + + # Verify migration records are removed + result = await db_session_for_plugins.execute( + select(PluginMigration).where(PluginMigration.plugin_name == "drop_test") + ) + assert result.scalars().all() == [] + + +@pytest.mark.asyncio +async def test_migration_runner_file_not_found(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test MigrationRunner raises FileNotFoundError for missing migration file.""" + runner = MigrationRunner(engine) + with pytest.raises(FileNotFoundError): + await runner.run_migration(db_session_for_plugins, "test", "nonexistent.sql") + + +def test_migration_runner_split_sql(): + """Test SQL splitting logic.""" + sql = "CREATE TABLE a();\nCREATE INDEX idx ON a();\n" + statements = MigrationRunner._split_sql(sql) + assert len(statements) == 2 + + # Test with comments + sql2 = "-- comment\nCREATE TABLE b();\n" + statements2 = MigrationRunner._split_sql(sql2) + assert len(statements2) == 1 + + # Test empty SQL + assert MigrationRunner._split_sql("") == [] + + +def test_migration_runner_extract_table_names(): + """Test table name extraction from SQL.""" + sql = "CREATE TABLE foo (); CREATE TABLE IF NOT EXISTS bar ();" + names = MigrationRunner._extract_table_names(sql) + assert "foo" in names + assert "bar" in names + + +def test_plugin_manifest_validation(): + """Test PluginManifest field validation.""" + # Valid manifest + m = PluginManifest( + name="my_plugin", + version="1.0.0", + display_name="My Plugin", + ) + assert m.name == "my_plugin" + assert m.dependencies == [] + assert m.routes == [] + + # Name is lowercased + m2 = PluginManifest(name="MyPlugin", version="1.0.0", display_name="Test") + assert m2.name == "myplugin" + + # Invalid name with special chars + with pytest.raises(Exception): + PluginManifest(name="my-plugin!", version="1.0.0", display_name="Test") + + +def test_base_plugin_repr(): + """Test BasePlugin __repr__ method.""" + plugin = TestSamplePlugin() + repr_str = repr(plugin) + assert "test_sample" in repr_str + assert "1.0.0" in repr_str + + +def test_base_plugin_no_manifest_error(): + """Test that BasePlugin without manifest raises ValueError.""" + class NoManifestPlugin(BasePlugin): + pass + + with pytest.raises(ValueError, match="manifest"): + NoManifestPlugin() + + +def test_base_plugin_name_version_properties(): + """Test BasePlugin name and version properties.""" + plugin = TestSamplePlugin() + assert plugin.name == "test_sample" + assert plugin.version == "1.0.0" + + +@pytest.mark.asyncio +async def test_base_plugin_on_install_default(): + """Test default on_install returns None (no-op).""" + class MinimalPlugin(BasePlugin): + manifest = PluginManifest( + name="minimal", version="1.0.0", display_name="Minimal", + ) + plugin = MinimalPlugin() + result = await plugin.on_install(None, None) + assert result is None + + +@pytest.mark.asyncio +async def test_base_plugin_on_uninstall_default(): + """Test default on_uninstall returns None (no-op).""" + class MinimalPlugin(BasePlugin): + manifest = PluginManifest( + name="minimal2", version="1.0.0", display_name="Minimal", + ) + plugin = MinimalPlugin() + result = await plugin.on_uninstall(None, None) + assert result is None + + +def test_base_plugin_get_routes_empty(): + """Test get_routes returns empty list when manifest has no routes.""" + plugin = TestSamplePlugin() + routes = plugin.get_routes() + assert routes == [] + + +def test_base_plugin_make_event_handler_noop(): + """Test that _make_event_handler creates noop for unknown events.""" + class MinimalPlugin(BasePlugin): + manifest = PluginManifest( + name="minimal3", version="1.0.0", display_name="Minimal", + events=["unknown.event"], + ) + plugin = MinimalPlugin() + handler = plugin._make_event_handler("unknown.event") + assert handler is not None + # It should be callable + assert callable(handler) + + +@pytest.mark.asyncio +async def test_service_manifest_schema(): + """Test plugin service get_manifest_schema returns dict.""" + from app.services.plugin_service import PluginService + service = PluginService() + schema = service.get_manifest_schema() + assert isinstance(schema, dict) + assert "fields" in schema + assert "example" in schema + + +@pytest.mark.asyncio +async def test_uninstall_inactive_plugin(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test uninstalling a plugin that is installed but not active.""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + registry.register_plugin(TestSamplePlugin()) + + # Install only (not activated) + await registry.install(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + + # Uninstall should work without deactivation + record = await registry.uninstall(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + assert record.status == "uninstalled" + + +@pytest.mark.asyncio +async def test_registry_activate_with_app_routes(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test that activating a plugin with a FastAPI app registers routes.""" + from fastapi import FastAPI, APIRouter + from app.plugins.manifest import PluginManifest, PluginRouteDef + + # Create a plugin with a route + class RoutePlugin(BasePlugin): + manifest = PluginManifest( + name="route_plugin", + version="1.0.0", + display_name="Route Plugin", + events=["test.event"], + ) + def get_routes(self) -> list: + test_router = APIRouter() + @test_router.get("/api/v1/route-plugin/test") + async def test_endpoint(): + return {"status": "ok"} + return [test_router] + + app = FastAPI() + registry = reset_registry_for_testing() + registry.initialize(engine, app) + registry.register_plugin(RoutePlugin()) + + # Install and activate + await registry.install(db_session_for_plugins, "route_plugin") + await db_session_for_plugins.commit() + await registry.activate(db_session_for_plugins, "route_plugin") + await db_session_for_plugins.commit() + + # Verify route was mounted + route_paths = [r.path for r in app.router.routes] + assert "/api/v1/route-plugin/test" in route_paths + + # Deactivate — route should be unmounted + await registry.deactivate(db_session_for_plugins, "route_plugin") + await db_session_for_plugins.commit() + route_paths_after = [r.path for r in app.router.routes] + assert "/api/v1/route-plugin/test" not in route_paths_after + + +@pytest.mark.asyncio +async def test_registry_uninstall_with_remove_data(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test registry uninstall with remove_data drops tables.""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + registry.register_plugin(TestSamplePlugin()) + + await registry.install(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + + record = await registry.uninstall(db_session_for_plugins, "test_sample", remove_data=True) + await db_session_for_plugins.commit() + assert record.status == "uninstalled" + assert "plugin_test_data" in getattr(record, "dropped_tables", []) + + +@pytest.mark.asyncio +async def test_registry_not_initialized_errors(): + """Test that uninitialized registry raises RuntimeError.""" + registry = reset_registry_for_testing() + with pytest.raises(RuntimeError, match="not initialized"): + _ = registry.migration_runner + with pytest.raises(RuntimeError, match="not initialized"): + _ = registry.engine + + +@pytest.mark.asyncio +async def test_registry_deactivate_already_inactive_direct(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test registry.deactivate is idempotent when already inactive.""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + registry.register_plugin(TestSamplePlugin()) + + await registry.install(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + + # Deactivate without activating first — status is 'installed', not 'inactive' + # But calling deactivate should still work (sets to inactive) + record = await registry.deactivate(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + assert record.status == "inactive" + + # Deactivate again — idempotent + record2 = await registry.deactivate(db_session_for_plugins, "test_sample") + await db_session_for_plugins.commit() + assert record2.status == "inactive" + + +@pytest.mark.asyncio +async def test_plugin_service_install_error_handling(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test plugin service raises ValueError for unknown plugin.""" + from app.services.plugin_service import PluginService + registry = reset_registry_for_testing() + registry.initialize(engine, None) + service = PluginService(registry=registry) + + with pytest.raises(ValueError, match="not found"): + await service.install_plugin(db_session_for_plugins, "nonexistent") + + +@pytest.mark.asyncio +async def test_plugin_service_activate_error_handling(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test plugin service raises ValueError when activating uninstalled plugin.""" + from app.services.plugin_service import PluginService + registry = reset_registry_for_testing() + registry.initialize(engine, None) + registry.register_plugin(TestSamplePlugin()) + service = PluginService(registry=registry) + + with pytest.raises(ValueError, match="not installed"): + await service.activate_plugin(db_session_for_plugins, "test_sample") + + +@pytest.mark.asyncio +async def test_plugin_service_deactivate_error_handling(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test plugin service raises ValueError for deactivating unknown plugin.""" + from app.services.plugin_service import PluginService + registry = reset_registry_for_testing() + registry.initialize(engine, None) + service = PluginService(registry=registry) + + with pytest.raises(ValueError, match="not found"): + await service.deactivate_plugin(db_session_for_plugins, "nonexistent") + + +@pytest.mark.asyncio +async def test_plugin_service_uninstall_error_handling(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test plugin service raises ValueError for uninstalling unknown plugin.""" + from app.services.plugin_service import PluginService + registry = reset_registry_for_testing() + registry.initialize(engine, None) + service = PluginService(registry=registry) + + with pytest.raises(ValueError, match="not found"): + await service.uninstall_plugin(db_session_for_plugins, "nonexistent") + + +@pytest.mark.asyncio +async def test_migration_runner_split_sql_dollar_quotes(): + """Test SQL splitting with dollar-quoted blocks.""" + sql = "CREATE FUNCTION foo() RETURNS void AS $$\nBEGIN\nEND;\n$$ LANGUAGE plpgsql;\nCREATE TABLE bar();\n" + statements = MigrationRunner._split_sql(sql) + assert len(statements) >= 2 + + +@pytest.mark.asyncio +async def test_registry_list_plugins_db_only_record(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test listing plugins includes DB-only records (installed but not discovered).""" + registry = reset_registry_for_testing() + registry.initialize(engine, None) + + # Manually create a DB record without registering the plugin + record = PluginModel( + name="orphan_plugin", + display_name="Orphan Plugin", + version="0.1.0", + status="installed", + installed=True, + active=False, + ) + db_session_for_plugins.add(record) + await db_session_for_plugins.commit() + + plugins = await registry.list_plugins(db_session_for_plugins) + names = [p["name"] for p in plugins] + assert "orphan_plugin" in names + orphan = [p for p in plugins if p["name"] == "orphan_plugin"][0] + assert orphan["installed"] is True + + +@pytest.mark.asyncio +async def test_base_plugin_on_event_fallback(): + """Test that on_event fallback handler works.""" + class FallbackPlugin(BasePlugin): + manifest = PluginManifest( + name="fallback", version="1.0.0", display_name="Fallback", + events=["custom.event"], + ) + async def on_event(self, payload: dict[str, Any]) -> None: + self.event_received = payload + + plugin = FallbackPlugin() + handler = plugin._make_event_handler("custom.event") + await handler({"test": True}) + assert plugin.event_received == {"test": True} + + +@pytest.mark.asyncio +async def test_migration_runner_valid_migration(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test MigrationRunner directly with a valid migration (has tenant_id).""" + runner = MigrationRunner(engine) + record = await runner.run_migration( + db_session_for_plugins, + "test_plugin", + "0001_test_plugin.sql", + ) + await db_session_for_plugins.commit() + assert record.plugin_name == "test_plugin" + assert record.migration_file == "0001_test_plugin.sql" + assert record.status == "applied" + + +@pytest.mark.asyncio +async def test_migration_runner_invalid_migration(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test MigrationRunner directly rejects migration without tenant_id.""" + runner = MigrationRunner(engine) + with pytest.raises(MigrationValidationError) as exc_info: + await runner.run_migration( + db_session_for_plugins, + "bad_plugin", + "0001_bad_migration.sql", + ) + assert "tenant_id" in str(exc_info.value).lower() + + +@pytest.mark.asyncio +async def test_migration_runner_idempotent(engine: AsyncEngine, db_session_for_plugins: AsyncSession): + """Test MigrationRunner skips already-applied migrations.""" + runner = MigrationRunner(engine) + # Run migration + await runner.run_migration(db_session_for_plugins, "test_idempotent", "0001_test_plugin.sql") + await db_session_for_plugins.commit() + + # Run all migrations again — should skip + records = await runner.run_all_migrations( + db_session_for_plugins, "test_idempotent", ["0001_test_plugin.sql"] + ) + assert len(records) == 0, "Already-applied migration should be skipped"