leocrm-bot
7a5a48fb4c
- Plugin registry with discover/install/activate/deactivate/uninstall lifecycle - PluginManifest Pydantic v2 schema (name, version, dependencies, routes, events, migrations) - BasePlugin abstract class with lifecycle hooks (on_install/activate/deactivate/uninstall) - Migration runner with tenant_id validator (rejects tables without tenant_id) - Event bus integration: register/unregister listeners on activate/deactivate - Service container DI: plugins receive db, cache, event_bus, storage, notifications - Idempotent operations (activate active=200, deactivate inactive=200) - UI registry for frontend component registration - 47 new tests (14 ACs + 33 unit tests), 103 total tests pass - Migration 0003: plugins + plugin_migrations tables - Coverage: 85.92% for plugin modules
53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
"""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})
|