Agent Zero
200 lines
7.2 KiB
Python
200 lines
7.2 KiB
Python
"""Echter Plugin-Lifecycle-Integrationstest ueber PluginService.
|
|
|
|
Regression fuer die Welle-1-Findings:
|
|
- P1: Runtime-Registrierungen (Permissions, active-set, Entity-Models) liefen nie,
|
|
weil der Vorher-Status erst NACH dem registry-Aufruf berechnet wurde.
|
|
- P3: sync_notification_types lief vor dem DB-Statusupdate.
|
|
Realer Produktionsweg: PluginService.install/activate/deactivate - keine manuellen
|
|
Registry-Manipulationen.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import Column, String, select
|
|
from sqlalchemy.dialects.postgresql import UUID as PGUUID
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
|
|
|
|
from app.core.db import Base, close_engine, reset_engine_for_testing
|
|
from app.core.hooks import get_hook_registry
|
|
from app.core.permission_registry import (
|
|
get_permission_registry,
|
|
init_permission_registry,
|
|
)
|
|
from app.models.notification import NotificationType
|
|
from app.plugins.base import BasePlugin
|
|
from app.plugins.manifest import PluginManifest
|
|
from app.plugins.registry import PluginRegistry, reset_registry_for_testing
|
|
from app.services.entity_permission_service import ENTITY_MODELS
|
|
from app.services.plugin_service import reset_plugin_service_for_testing
|
|
|
|
PLUGIN_NAME = "lifecycle_svc_plugin"
|
|
PERM = "lifecycletest:read"
|
|
ENTITY = "lifecycle_svc_item"
|
|
NT_KEY = f"{PLUGIN_NAME}.updated"
|
|
|
|
|
|
class LifecycleSvcItem(Base):
|
|
"""Minimal tenant-scoped model owned by the test plugin."""
|
|
|
|
__tablename__ = "lifecycle_svc_items"
|
|
|
|
id: Any = Column(PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
tenant_id: Any = Column(PGUUID(as_uuid=True), nullable=False, index=True)
|
|
name: Any = Column(String(120), nullable=False)
|
|
|
|
|
|
class LifecycleSvcPlugin(BasePlugin):
|
|
"""Test plugin contributing permissions, entity model and notification type."""
|
|
|
|
manifest = PluginManifest(
|
|
name=PLUGIN_NAME,
|
|
version="1.0.0",
|
|
display_name="Lifecycle Svc Plugin",
|
|
description="Integration fixture for the full service lifecycle.",
|
|
dependencies=[],
|
|
routes=[],
|
|
events=[],
|
|
migrations=[],
|
|
permissions=[PERM],
|
|
)
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.activate_calls = 0
|
|
self.deactivate_calls = 0
|
|
|
|
def get_entity_models(self) -> dict[str, type]:
|
|
return {ENTITY: LifecycleSvcItem}
|
|
|
|
async def on_activate(self, db, service_container, event_bus) -> None:
|
|
self.activate_calls += 1
|
|
await super().on_activate(db, service_container, event_bus)
|
|
|
|
async def on_deactivate(self, db, service_container, event_bus) -> None:
|
|
self.deactivate_calls += 1
|
|
await super().on_deactivate(db, service_container, event_bus)
|
|
|
|
def get_notification_types(self) -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"type_key": NT_KEY,
|
|
"category": "general",
|
|
"label": "Lifecycle update",
|
|
"description": "Type contributed by the lifecycle svc plugin",
|
|
"is_enabled_by_default": True,
|
|
}
|
|
]
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def svc_env(engine: AsyncEngine):
|
|
"""Fresh registries plus registered plugin instance and its table."""
|
|
reset_engine_for_testing(engine)
|
|
reg: PluginRegistry = reset_registry_for_testing()
|
|
init_permission_registry(set())
|
|
get_hook_registry()._reset_for_testing()
|
|
service = reset_plugin_service_for_testing(reg)
|
|
|
|
plugin = LifecycleSvcPlugin()
|
|
reg.register_plugin(plugin)
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(
|
|
lambda sync_conn: Base.metadata.create_all(
|
|
sync_conn, tables=[LifecycleSvcItem.__table__]
|
|
)
|
|
)
|
|
|
|
yield {"service": service, "registry": reg, "plugin": plugin}
|
|
|
|
await close_engine()
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(
|
|
lambda sync_conn: Base.metadata.drop_all(
|
|
sync_conn, tables=[LifecycleSvcItem.__table__]
|
|
)
|
|
)
|
|
|
|
|
|
def _plugin_perms() -> int:
|
|
return sum(
|
|
1 for p in get_permission_registry().get_all()
|
|
if p.get("plugin_name") == PLUGIN_NAME
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_lifecycle_registers_and_deregisters_runtime(
|
|
db_session: AsyncSession, svc_env,
|
|
):
|
|
service = svc_env["service"]
|
|
perm_reg = get_permission_registry()
|
|
|
|
# Install (real path)
|
|
install_resp = await service.install_plugin(db_session, PLUGIN_NAME)
|
|
assert install_resp["installed"] is True
|
|
|
|
# Activate #1 - runtime registrations MUST run
|
|
resp1 = await service.activate_plugin(db_session, PLUGIN_NAME)
|
|
assert resp1["message"] == "Plugin activated successfully", (
|
|
"Vorher-Status falsch berechnet: Runtime-Registrierungen wurden uebersprungen"
|
|
)
|
|
assert perm_reg.is_valid(PERM)
|
|
assert _plugin_perms() == 1
|
|
# Gate source _active_plugins must contain the plugin (today's prod-403 root cause)
|
|
assert perm_reg.is_plugin_active(PLUGIN_NAME)
|
|
assert ENTITY_MODELS.get(ENTITY) is LifecycleSvcItem
|
|
assert svc_env["plugin"].activate_calls == 1
|
|
|
|
# P3 regression: notification type synced AFTER status became active
|
|
nt_q = select(NotificationType).where(NotificationType.type_key == NT_KEY)
|
|
assert (await db_session.execute(nt_q)).scalar_one_or_none() is not None, (
|
|
"Notification-Type fehlt: sync lief vor dem Statusupdate"
|
|
)
|
|
|
|
# Activate #2 - idempotent, no double registration
|
|
resp2 = await service.activate_plugin(db_session, PLUGIN_NAME)
|
|
assert resp2["message"] == "Plugin is already active"
|
|
assert _plugin_perms() == 1
|
|
assert svc_env["plugin"].activate_calls == 1
|
|
|
|
# Deactivate #1 - runtime deregistrations MUST run
|
|
resp3 = await service.deactivate_plugin(db_session, PLUGIN_NAME)
|
|
assert resp3["message"] == "Plugin deactivated successfully"
|
|
assert not perm_reg.is_valid(PERM), "Permission blieb registriert (P1 dead-cleanup)"
|
|
assert not perm_reg.is_plugin_active(PLUGIN_NAME), "stale gate entry blieb stehen"
|
|
assert ENTITY not in ENTITY_MODELS, "Entity-Model blieb registriert"
|
|
assert svc_env["plugin"].deactivate_calls == 1
|
|
|
|
# Notification type removed by post-deactivation re-sync
|
|
assert (await db_session.execute(nt_q)).scalar_one_or_none() is None
|
|
|
|
# Deactivate #2 - safe no-op
|
|
resp4 = await service.deactivate_plugin(db_session, PLUGIN_NAME)
|
|
assert resp4["message"] == "Plugin is already inactive"
|
|
assert svc_env["plugin"].deactivate_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reactivate_restores_contributions_exactly_once(
|
|
db_session: AsyncSession, svc_env,
|
|
):
|
|
"""Activate-deactivate-activate restores every contribution exactly once."""
|
|
service = svc_env["service"]
|
|
perm_reg = get_permission_registry()
|
|
|
|
await service.install_plugin(db_session, PLUGIN_NAME)
|
|
await service.activate_plugin(db_session, PLUGIN_NAME)
|
|
await service.deactivate_plugin(db_session, PLUGIN_NAME)
|
|
|
|
resp = await service.activate_plugin(db_session, PLUGIN_NAME)
|
|
assert resp["message"] == "Plugin activated successfully"
|
|
assert _plugin_perms() == 1
|
|
assert perm_reg.is_plugin_active(PLUGIN_NAME)
|
|
assert ENTITY_MODELS.get(ENTITY) is LifecycleSvcItem
|