188 lines
6.3 KiB
Python
188 lines
6.3 KiB
Python
|
|
"""Tests for MiniAppRegistry singleton pattern and plugin lifecycle integration."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.plugins.builtins.kommunikation.miniapp_registry import (
|
||
|
|
MiniAppRegistry,
|
||
|
|
get_miniapp_registry,
|
||
|
|
reset_miniapp_registry,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(autouse=True)
|
||
|
|
def _reset_registry():
|
||
|
|
"""Ensure a clean singleton before and after each test."""
|
||
|
|
reset_miniapp_registry()
|
||
|
|
yield
|
||
|
|
reset_miniapp_registry()
|
||
|
|
|
||
|
|
|
||
|
|
class TestSingletonPattern:
|
||
|
|
def test_get_miniapp_registry_returns_same_instance(self):
|
||
|
|
"""get_miniapp_registry() must return the same object every call."""
|
||
|
|
r1 = get_miniapp_registry()
|
||
|
|
r2 = get_miniapp_registry()
|
||
|
|
assert r1 is r2
|
||
|
|
assert isinstance(r1, MiniAppRegistry)
|
||
|
|
|
||
|
|
def test_reset_miniapp_registry_creates_new_instance(self):
|
||
|
|
"""After reset, a new instance is created."""
|
||
|
|
r1 = get_miniapp_registry()
|
||
|
|
reset_miniapp_registry()
|
||
|
|
r2 = get_miniapp_registry()
|
||
|
|
assert r1 is not r2
|
||
|
|
|
||
|
|
def test_reset_miniapp_registry_sets_none(self):
|
||
|
|
"""reset_miniapp_registry() should set the global to None."""
|
||
|
|
get_miniapp_registry() # ensure singleton exists
|
||
|
|
reset_miniapp_registry()
|
||
|
|
# Access the module-level global directly
|
||
|
|
import app.plugins.builtins.kommunikation.miniapp_registry as mod
|
||
|
|
assert mod._registry is None
|
||
|
|
|
||
|
|
|
||
|
|
class TestRegisterAndList:
|
||
|
|
def test_register_and_list_apps(self):
|
||
|
|
"""register() adds an app that appears in list_apps()."""
|
||
|
|
registry = get_miniapp_registry()
|
||
|
|
registry.register(
|
||
|
|
app_id="test_app",
|
||
|
|
name="Test App",
|
||
|
|
icon="🧪",
|
||
|
|
description="A test app",
|
||
|
|
plugin_name="test_plugin",
|
||
|
|
render_schema={"type": "object"},
|
||
|
|
)
|
||
|
|
apps = registry.list_apps()
|
||
|
|
assert len(apps) == 1
|
||
|
|
assert apps[0]["app_id"] == "test_app"
|
||
|
|
assert apps[0]["name"] == "Test App"
|
||
|
|
assert apps[0]["plugin_name"] == "test_plugin"
|
||
|
|
|
||
|
|
def test_register_overwrites_same_app_id(self):
|
||
|
|
"""Registering the same app_id replaces the previous entry."""
|
||
|
|
registry = get_miniapp_registry()
|
||
|
|
registry.register(
|
||
|
|
app_id="dup_app",
|
||
|
|
name="First",
|
||
|
|
icon="1",
|
||
|
|
description="",
|
||
|
|
plugin_name="plugin_a",
|
||
|
|
)
|
||
|
|
registry.register(
|
||
|
|
app_id="dup_app",
|
||
|
|
name="Second",
|
||
|
|
icon="2",
|
||
|
|
description="",
|
||
|
|
plugin_name="plugin_b",
|
||
|
|
)
|
||
|
|
apps = registry.list_apps()
|
||
|
|
assert len(apps) == 1
|
||
|
|
assert apps[0]["name"] == "Second"
|
||
|
|
|
||
|
|
def test_list_apps_empty_by_default(self):
|
||
|
|
"""Fresh registry has no apps."""
|
||
|
|
registry = get_miniapp_registry()
|
||
|
|
assert registry.list_apps() == []
|
||
|
|
|
||
|
|
|
||
|
|
class TestUnregisterPlugin:
|
||
|
|
def test_unregister_plugin_removes_all_apps(self):
|
||
|
|
"""unregister_plugin() removes all apps from that plugin."""
|
||
|
|
registry = get_miniapp_registry()
|
||
|
|
registry.register("a1", "A1", "x", "", "plugin_x")
|
||
|
|
registry.register("a2", "A2", "x", "", "plugin_x")
|
||
|
|
registry.register("b1", "B1", "x", "", "plugin_y")
|
||
|
|
|
||
|
|
registry.unregister_plugin("plugin_x")
|
||
|
|
|
||
|
|
apps = registry.list_apps()
|
||
|
|
assert len(apps) == 1
|
||
|
|
assert apps[0]["app_id"] == "b1"
|
||
|
|
|
||
|
|
def test_unregister_plugin_noop_for_unknown(self):
|
||
|
|
"""unregister_plugin() for a non-existent plugin is a no-op."""
|
||
|
|
registry = get_miniapp_registry()
|
||
|
|
registry.register("a1", "A1", "x", "", "plugin_x")
|
||
|
|
registry.unregister_plugin("nonexistent")
|
||
|
|
assert len(registry.list_apps()) == 1
|
||
|
|
|
||
|
|
def test_unregister_single_app(self):
|
||
|
|
"""unregister() removes a specific app by id."""
|
||
|
|
registry = get_miniapp_registry()
|
||
|
|
registry.register("a1", "A1", "x", "", "plugin_x")
|
||
|
|
registry.register("a2", "A2", "x", "", "plugin_x")
|
||
|
|
|
||
|
|
registry.unregister("a1")
|
||
|
|
apps = registry.list_apps()
|
||
|
|
assert len(apps) == 1
|
||
|
|
assert apps[0]["app_id"] == "a2"
|
||
|
|
|
||
|
|
|
||
|
|
class TestPluginLifecycleIntegration:
|
||
|
|
"""Test that plugin activation/deactivation uses the shared singleton."""
|
||
|
|
|
||
|
|
def test_activation_registers_in_shared_registry(self):
|
||
|
|
"""When a plugin registers miniapps, they appear in the shared registry."""
|
||
|
|
registry = get_miniapp_registry()
|
||
|
|
# Simulate what automation/plugin.py on_activate does
|
||
|
|
registry.register(
|
||
|
|
app_id="automation_agent",
|
||
|
|
name="Agent Runner",
|
||
|
|
icon="🤖",
|
||
|
|
description="Run an AI agent",
|
||
|
|
plugin_name="automation",
|
||
|
|
render_schema={"type": "object"},
|
||
|
|
)
|
||
|
|
|
||
|
|
# A different caller (e.g. a route) should see the same apps
|
||
|
|
shared = get_miniapp_registry()
|
||
|
|
assert shared is registry
|
||
|
|
apps = shared.list_apps()
|
||
|
|
assert any(a["app_id"] == "automation_agent" for a in apps)
|
||
|
|
|
||
|
|
def test_deactivation_removes_from_shared_registry(self):
|
||
|
|
"""When a plugin deactivates, its miniapps are removed from the shared registry."""
|
||
|
|
registry = get_miniapp_registry()
|
||
|
|
registry.register(
|
||
|
|
app_id="automation_agent",
|
||
|
|
name="Agent Runner",
|
||
|
|
icon="🤖",
|
||
|
|
description="Run an AI agent",
|
||
|
|
plugin_name="automation",
|
||
|
|
)
|
||
|
|
registry.register(
|
||
|
|
app_id="komm_contact",
|
||
|
|
name="Contact Picker",
|
||
|
|
icon="👤",
|
||
|
|
description="Pick a contact",
|
||
|
|
plugin_name="kommunikation",
|
||
|
|
)
|
||
|
|
|
||
|
|
# Simulate on_deactivate for automation
|
||
|
|
registry.unregister_plugin("automation")
|
||
|
|
|
||
|
|
apps = registry.list_apps()
|
||
|
|
assert len(apps) == 1
|
||
|
|
assert apps[0]["plugin_name"] == "kommunikation"
|
||
|
|
|
||
|
|
def test_cross_plugin_visibility(self):
|
||
|
|
"""Apps registered by plugin A are visible to plugin B via the singleton."""
|
||
|
|
# Plugin A registers
|
||
|
|
registry_a = get_miniapp_registry()
|
||
|
|
registry_a.register(
|
||
|
|
app_id="plugin_a_app",
|
||
|
|
name="A App",
|
||
|
|
icon="x",
|
||
|
|
description="",
|
||
|
|
plugin_name="plugin_a",
|
||
|
|
)
|
||
|
|
|
||
|
|
# Plugin B queries
|
||
|
|
registry_b = get_miniapp_registry()
|
||
|
|
assert registry_a is registry_b
|
||
|
|
apps = registry_b.list_apps()
|
||
|
|
assert any(a["app_id"] == "plugin_a_app" for a in apps)
|