Agent Zero
dfd9e778c5
Tests: - Registry initialization and engine requirement - Plugin registration and discovery - Load order with and without dependencies - Core plugin deactivation blocked - Deactivation blocked by active dependents - Event handler registration on activate - Event handler unregistration on deactivate - Activate → deactivate → reactivate cycle - Idempotent activate when already active - Idempotent deactivate when already inactive Phase 3 (Plugin-Lifecycle) verified: - install: idempotent, dependency checks, migrations via crm_migration - activate: idempotent, per-tenant with RLS context, event handlers - deactivate: idempotent, core protection, dependency check, handler cleanup - uninstall: deactivate first, then optional drop tables - main.py: per-tenant activation with set_tenant_context - Worker: event handlers only for active plugins (Gate 5) - Router: only in API, not in worker
296 lines
10 KiB
Python
296 lines
10 KiB
Python
"""Tests for plugin lifecycle: install, activate, deactivate, reactivate.
|
|
|
|
Phase 3 — Plugin-Lifecycle verification.
|
|
"""
|
|
|
|
import pytest
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from app.plugins.registry import PluginRegistry
|
|
from app.plugins.base import BasePlugin
|
|
from app.plugins.manifest import PluginManifest
|
|
|
|
|
|
class FakePlugin(BasePlugin):
|
|
"""Test plugin for lifecycle testing."""
|
|
manifest = PluginManifest(
|
|
name="test_fake",
|
|
display_name="Fake Plugin",
|
|
version="1.0.0",
|
|
description="Test plugin",
|
|
is_core=False,
|
|
)
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.activated = False
|
|
self.deactivated = False
|
|
|
|
async def on_activate(self, db, container, event_bus):
|
|
self.activated = True
|
|
self.deactivated = False
|
|
await super().on_activate(db, container, event_bus)
|
|
|
|
async def on_deactivate(self, db, container, event_bus):
|
|
self.deactivated = True
|
|
self.activated = False
|
|
await super().on_deactivate(db, container, event_bus)
|
|
|
|
|
|
class FakeCorePlugin(BasePlugin):
|
|
"""Core plugin that cannot be deactivated."""
|
|
manifest = PluginManifest(
|
|
name="test_core",
|
|
display_name="Core Plugin",
|
|
version="1.0.0",
|
|
description="Core test plugin",
|
|
is_core=True,
|
|
)
|
|
|
|
|
|
class FakeDependentPlugin(BasePlugin):
|
|
"""Plugin that depends on test_fake."""
|
|
manifest = PluginManifest(
|
|
name="test_dependent",
|
|
display_name="Dependent Plugin",
|
|
version="1.0.0",
|
|
description="Depends on test_fake",
|
|
dependencies=["test_fake"],
|
|
)
|
|
|
|
|
|
class TestPluginRegistryLifecycle:
|
|
"""Test plugin registry lifecycle methods."""
|
|
|
|
def test_registry_initialization(self):
|
|
"""Registry can be initialized with engine and app."""
|
|
registry = PluginRegistry()
|
|
engine = MagicMock()
|
|
app = MagicMock()
|
|
registry.initialize(engine, app)
|
|
assert registry._engine is engine
|
|
assert registry._app is app
|
|
assert registry._initialized is True
|
|
|
|
def test_registry_initialization_requires_engine(self):
|
|
"""Registry engine property raises if not initialized."""
|
|
registry = PluginRegistry()
|
|
with pytest.raises(RuntimeError, match="not initialized"):
|
|
_ = registry.engine
|
|
|
|
def test_register_and_get_plugin(self):
|
|
"""Plugin can be registered and retrieved."""
|
|
registry = PluginRegistry()
|
|
plugin = FakePlugin()
|
|
registry.register_plugin(plugin)
|
|
assert registry.get_plugin("test_fake") is plugin
|
|
assert registry.get_plugin("nonexistent") is None
|
|
|
|
def test_list_discovered(self):
|
|
"""list_discovered returns all registered plugin names."""
|
|
registry = PluginRegistry()
|
|
registry.register_plugin(FakePlugin())
|
|
registry.register_plugin(FakeCorePlugin())
|
|
discovered = registry.list_discovered()
|
|
assert "test_fake" in discovered
|
|
assert "test_core" in discovered
|
|
|
|
def test_resolve_load_order_no_deps(self):
|
|
"""Load order for plugins without dependencies."""
|
|
registry = PluginRegistry()
|
|
registry.register_plugin(FakePlugin())
|
|
registry.register_plugin(FakeCorePlugin())
|
|
order = registry.resolve_load_order()
|
|
assert order[0] == "test_core"
|
|
assert "test_fake" in order
|
|
|
|
def test_resolve_load_order_with_deps(self):
|
|
"""Load order respects dependencies."""
|
|
registry = PluginRegistry()
|
|
registry.register_plugin(FakePlugin())
|
|
registry.register_plugin(FakeDependentPlugin())
|
|
order = registry.resolve_load_order()
|
|
fake_idx = order.index("test_fake")
|
|
dep_idx = order.index("test_dependent")
|
|
assert fake_idx < dep_idx
|
|
|
|
def test_get_dependents(self):
|
|
"""get_dependents returns plugins that depend on the given plugin."""
|
|
registry = PluginRegistry()
|
|
registry.register_plugin(FakePlugin())
|
|
registry.register_plugin(FakeDependentPlugin())
|
|
dependents = registry.get_dependents("test_fake")
|
|
assert "test_dependent" in dependents
|
|
dependents = registry.get_dependents("test_core")
|
|
assert dependents == []
|
|
|
|
def test_core_plugin_cannot_deactivate(self):
|
|
"""Core plugins cannot be deactivated."""
|
|
registry = PluginRegistry()
|
|
registry.initialize(MagicMock())
|
|
registry.register_plugin(FakeCorePlugin())
|
|
|
|
record = MagicMock()
|
|
record.active = True
|
|
record.status = "active"
|
|
record.is_core = True
|
|
|
|
with patch.object(registry, '_get_plugin_record', return_value=record):
|
|
with pytest.raises(ValueError, match="core plugin"):
|
|
asyncio.get_event_loop().run_until_complete(
|
|
registry.deactivate(MagicMock(), "test_core")
|
|
)
|
|
|
|
def test_deactivate_blocked_by_active_dependents(self):
|
|
"""Cannot deactivate plugin when active plugins depend on it."""
|
|
registry = PluginRegistry()
|
|
registry.initialize(MagicMock())
|
|
registry.register_plugin(FakePlugin())
|
|
registry.register_plugin(FakeDependentPlugin())
|
|
|
|
fake_record = MagicMock()
|
|
fake_record.active = True
|
|
fake_record.status = "active"
|
|
fake_record.is_core = False
|
|
|
|
dependent_record = MagicMock()
|
|
dependent_record.active = True
|
|
|
|
async def mock_get_record(db, name):
|
|
if name == "test_fake":
|
|
return fake_record
|
|
if name == "test_dependent":
|
|
return dependent_record
|
|
return None
|
|
|
|
with patch.object(registry, '_get_plugin_record', side_effect=mock_get_record):
|
|
with pytest.raises(ValueError, match="depend on it"):
|
|
asyncio.get_event_loop().run_until_complete(
|
|
registry.deactivate(MagicMock(), "test_fake")
|
|
)
|
|
|
|
|
|
class TestBasePluginLifecycle:
|
|
"""Test BasePlugin lifecycle hooks."""
|
|
|
|
def test_on_activate_registers_event_handlers(self):
|
|
"""on_activate subscribes to events from manifest."""
|
|
plugin = FakePlugin()
|
|
plugin.manifest = PluginManifest(
|
|
name="test_fake",
|
|
display_name="Fake",
|
|
version="1.0.0",
|
|
events=["test.event"],
|
|
)
|
|
event_bus = MagicMock()
|
|
event_bus.subscribe = AsyncMock()
|
|
|
|
asyncio.get_event_loop().run_until_complete(
|
|
plugin.on_activate(MagicMock(), MagicMock(), event_bus)
|
|
)
|
|
|
|
assert plugin.activated is True
|
|
assert "test.event" in plugin._event_handlers
|
|
event_bus.subscribe.assert_called_once()
|
|
|
|
def test_on_deactivate_unregisters_event_handlers(self):
|
|
"""on_deactivate unsubscribes all event handlers."""
|
|
plugin = FakePlugin()
|
|
plugin.manifest = PluginManifest(
|
|
name="test_fake",
|
|
display_name="Fake",
|
|
version="1.0.0",
|
|
events=["test.event1", "test.event2"],
|
|
)
|
|
event_bus = MagicMock()
|
|
event_bus.subscribe = AsyncMock()
|
|
event_bus.unsubscribe = AsyncMock()
|
|
|
|
asyncio.get_event_loop().run_until_complete(
|
|
plugin.on_activate(MagicMock(), MagicMock(), event_bus)
|
|
)
|
|
assert len(plugin._event_handlers) == 2
|
|
|
|
asyncio.get_event_loop().run_until_complete(
|
|
plugin.on_deactivate(MagicMock(), MagicMock(), event_bus)
|
|
)
|
|
assert plugin.deactivated is True
|
|
assert len(plugin._event_handlers) == 0
|
|
assert event_bus.unsubscribe.call_count == 2
|
|
|
|
def test_activate_deactivate_reactivate_cycle(self):
|
|
"""Plugin can be activated, deactivated, and reactivated."""
|
|
plugin = FakePlugin()
|
|
plugin.manifest = PluginManifest(
|
|
name="test_fake",
|
|
display_name="Fake",
|
|
version="1.0.0",
|
|
events=["test.event"],
|
|
)
|
|
event_bus = MagicMock()
|
|
event_bus.subscribe = AsyncMock()
|
|
event_bus.unsubscribe = AsyncMock()
|
|
|
|
asyncio.get_event_loop().run_until_complete(
|
|
plugin.on_activate(MagicMock(), MagicMock(), event_bus)
|
|
)
|
|
assert plugin.activated is True
|
|
assert len(plugin._event_handlers) == 1
|
|
|
|
asyncio.get_event_loop().run_until_complete(
|
|
plugin.on_deactivate(MagicMock(), MagicMock(), event_bus)
|
|
)
|
|
assert plugin.deactivated is True
|
|
assert len(plugin._event_handlers) == 0
|
|
|
|
asyncio.get_event_loop().run_until_complete(
|
|
plugin.on_activate(MagicMock(), MagicMock(), event_bus)
|
|
)
|
|
assert plugin.activated is True
|
|
assert len(plugin._event_handlers) == 1
|
|
|
|
|
|
class TestPluginIdempotency:
|
|
"""Test that lifecycle methods are idempotent."""
|
|
|
|
def test_activate_idempotent_when_already_active(self):
|
|
"""Activating an already-active plugin should not fail."""
|
|
registry = PluginRegistry()
|
|
registry.initialize(MagicMock())
|
|
registry.register_plugin(FakePlugin())
|
|
|
|
record = MagicMock()
|
|
record.active = True
|
|
record.status = "active"
|
|
record.version = "1.0.0"
|
|
|
|
db = MagicMock()
|
|
db.flush = AsyncMock()
|
|
|
|
with patch.object(registry, '_get_plugin_record', return_value=record):
|
|
result = asyncio.get_event_loop().run_until_complete(
|
|
registry.activate(db, "test_fake")
|
|
)
|
|
assert result is record
|
|
|
|
def test_deactivate_idempotent_when_already_inactive(self):
|
|
"""Deactivating an already-inactive plugin should not fail."""
|
|
registry = PluginRegistry()
|
|
registry.initialize(MagicMock())
|
|
registry.register_plugin(FakePlugin())
|
|
|
|
record = MagicMock()
|
|
record.active = False
|
|
record.status = "inactive"
|
|
|
|
with patch.object(registry, '_get_plugin_record', return_value=record):
|
|
result = asyncio.get_event_loop().run_until_complete(
|
|
registry.deactivate(MagicMock(), "test_fake")
|
|
)
|
|
assert result is record
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|