Agent Zero
727d86614e
P0 (7): Auth-bypass removed, migrations fixed, plugin-upload disabled, RLS FORCE+WITH CHECK, plugin double-registration fixed, persistent volume, domain removed P1 (11): User/tenant model, Redis centralized, worker separated, transactional outbox, XSS fixed, DMS chunked streaming, permissions unified, password reset, metrics secured, config/docs fixed, cross-tenant FK P2 (4): Contact model normalized, cross-imports reduced 94%, commands+state machines for contacts/dms/mail/calendar, SPA path-traversal 8 new migrations, 99 unit tests, 13 commands, 8 contracts, 72 files changed
232 lines
7.9 KiB
Python
232 lines
7.9 KiB
Python
"""Tests for the plugin contract registry and contract modules.
|
|
|
|
Verifies that:
|
|
1. ContractRegistry singleton works correctly
|
|
2. Contracts for kommunikation, ai_assistant, and mail register and resolve
|
|
3. Contract objects expose the expected public symbols
|
|
4. Lazy loading works for unregistered plugins
|
|
5. ContractError is raised for missing contracts via require_contract
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import pytest
|
|
|
|
from app.plugins.builtins.contracts import (
|
|
ContractError,
|
|
ContractRegistry,
|
|
get_contract,
|
|
get_contract_registry,
|
|
reset_contract_registry_for_testing,
|
|
)
|
|
|
|
|
|
def _reload_contracts(plugin_name: str):
|
|
"""Force re-import of a plugin's contracts module so it re-registers."""
|
|
module_path = f"app.plugins.builtins.{plugin_name}.contracts"
|
|
mod = importlib.import_module(module_path)
|
|
importlib.reload(mod)
|
|
return mod
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_registry():
|
|
"""Ensure a fresh registry for each test."""
|
|
reset_contract_registry_for_testing()
|
|
yield
|
|
reset_contract_registry_for_testing()
|
|
|
|
|
|
# ─── ContractRegistry singleton ───
|
|
|
|
|
|
class TestContractRegistry:
|
|
def test_singleton_identity(self):
|
|
"""get_contract_registry returns the same instance."""
|
|
a = get_contract_registry()
|
|
b = get_contract_registry()
|
|
assert a is b
|
|
|
|
def test_register_and_get(self):
|
|
"""register stores and get_contract retrieves."""
|
|
reg = get_contract_registry()
|
|
sentinel = object()
|
|
reg.register("demo", sentinel)
|
|
assert reg.get_contract("demo") is sentinel
|
|
|
|
def test_unregister(self):
|
|
"""unregister removes the contract."""
|
|
reg = get_contract_registry()
|
|
sentinel = object()
|
|
reg.register("demo", sentinel)
|
|
reg.unregister("demo")
|
|
assert reg.get_contract("demo") is None
|
|
|
|
def test_get_contract_returns_none_for_unknown(self):
|
|
"""Unknown plugin returns None, not raises."""
|
|
reg = get_contract_registry()
|
|
assert reg.get_contract("does_not_exist") is None
|
|
|
|
def test_require_contract_raises_for_missing(self):
|
|
"""require_contract raises ContractError when missing."""
|
|
reg = get_contract_registry()
|
|
with pytest.raises(ContractError):
|
|
reg.require_contract("does_not_exist")
|
|
|
|
def test_require_contract_returns_contract(self):
|
|
"""require_contract returns the contract when registered."""
|
|
reg = get_contract_registry()
|
|
sentinel = object()
|
|
reg.register("demo", sentinel)
|
|
assert reg.require_contract("demo") is sentinel
|
|
|
|
def test_list_available(self):
|
|
"""list_available returns sorted plugin names."""
|
|
reg = get_contract_registry()
|
|
reg.register("zebra", object())
|
|
reg.register("alpha", object())
|
|
assert reg.list_available() == ["alpha", "zebra"]
|
|
|
|
def test_module_level_get_contract(self):
|
|
"""Module-level get_contract function works."""
|
|
reg = get_contract_registry()
|
|
sentinel = object()
|
|
reg.register("demo", sentinel)
|
|
assert get_contract("demo") is sentinel
|
|
|
|
def test_reset_for_testing_clears_state(self):
|
|
"""reset clears all registered contracts."""
|
|
reg = get_contract_registry()
|
|
reg.register("a", object())
|
|
reg.register("b", object())
|
|
assert len(reg.list_available()) == 2
|
|
reset_contract_registry_for_testing()
|
|
assert reg.list_available() == []
|
|
|
|
|
|
# ─── Kommunikation contract ───
|
|
|
|
|
|
class TestKommunikationContract:
|
|
@pytest.fixture(autouse=True)
|
|
def _load_komm(self):
|
|
"""Reload kommunikation contracts so it re-registers after reset."""
|
|
_reload_contracts("kommunikation")
|
|
|
|
def test_contract_registers(self):
|
|
"""Importing kommunikation.contracts registers it in the registry."""
|
|
contract = get_contract("kommunikation")
|
|
assert contract is not None
|
|
assert contract.contract_name == "kommunikation"
|
|
|
|
def test_exposes_services(self):
|
|
"""Contract exposes service functions."""
|
|
contract = get_contract("kommunikation")
|
|
assert callable(contract.parse_mentions)
|
|
assert callable(contract.get_conversation)
|
|
assert callable(contract.get_messages)
|
|
assert callable(contract.send_message)
|
|
assert callable(contract.create_plugin_room)
|
|
|
|
def test_exposes_participant_registry(self):
|
|
"""Contract exposes participant registry types."""
|
|
contract = get_contract("kommunikation")
|
|
assert callable(contract.get_participant_registry)
|
|
assert contract.ParticipantHandler is not None
|
|
|
|
def test_exposes_miniapp_registry(self):
|
|
"""Contract exposes MiniAppRegistry."""
|
|
contract = get_contract("kommunikation")
|
|
assert contract.MiniAppRegistry is not None
|
|
assert contract.MiniAppDef is not None
|
|
|
|
def test_exposes_models(self):
|
|
"""Contract exposes ORM models."""
|
|
contract = get_contract("kommunikation")
|
|
assert contract.CommConversation is not None
|
|
assert contract.CommMessage is not None
|
|
assert contract.CommParticipant is not None
|
|
|
|
def test_parse_mentions_works(self):
|
|
"""parse_mentions actually parses @mentions."""
|
|
contract = get_contract("kommunikation")
|
|
result = contract.parse_mentions("hello @ai_proactive and @system")
|
|
assert result == ["ai_proactive", "system"]
|
|
|
|
|
|
# ─── AI Assistant contract ───
|
|
|
|
|
|
class TestAIAssistantContract:
|
|
@pytest.fixture(autouse=True)
|
|
def _load_ai(self):
|
|
"""Reload ai_assistant contracts so it re-registers after reset."""
|
|
_reload_contracts("ai_assistant")
|
|
|
|
def test_contract_registers(self):
|
|
"""Importing ai_assistant.contracts registers it."""
|
|
contract = get_contract("ai_assistant")
|
|
assert contract is not None
|
|
assert contract.contract_name == "ai_assistant"
|
|
|
|
def test_exposes_tool_registry(self):
|
|
"""Contract exposes tool registry functions and types."""
|
|
contract = get_contract("ai_assistant")
|
|
assert callable(contract.get_tool_registry)
|
|
assert contract.ToolRegistry is not None
|
|
assert contract.AITool is not None
|
|
|
|
def test_exposes_get_default_provider(self):
|
|
"""Contract exposes get_default_provider."""
|
|
contract = get_contract("ai_assistant")
|
|
assert callable(contract.get_default_provider)
|
|
|
|
def test_tool_registry_singleton_works(self):
|
|
"""get_tool_registry returns a working singleton."""
|
|
contract = get_contract("ai_assistant")
|
|
reg = contract.get_tool_registry()
|
|
assert reg is not None
|
|
reg2 = contract.get_tool_registry()
|
|
assert reg is reg2
|
|
|
|
|
|
# ─── Mail contract ───
|
|
|
|
|
|
class TestMailContract:
|
|
@pytest.fixture(autouse=True)
|
|
def _load_mail(self):
|
|
"""Reload mail contracts so it re-registers after reset."""
|
|
_reload_contracts("mail")
|
|
|
|
def test_contract_registers(self):
|
|
"""Importing mail.contracts registers it."""
|
|
contract = get_contract("mail")
|
|
assert contract is not None
|
|
assert contract.contract_name == "mail"
|
|
|
|
def test_exposes_mail_model(self):
|
|
"""Contract exposes the Mail ORM model."""
|
|
contract = get_contract("mail")
|
|
assert contract.Mail is not None
|
|
from app.plugins.builtins.mail.models import Mail as MailModel
|
|
assert contract.Mail is MailModel
|
|
|
|
|
|
# ─── Lazy loading ───
|
|
|
|
|
|
class TestLazyLoading:
|
|
def test_lazy_load_on_first_access(self):
|
|
"""get_contract triggers lazy load of contracts module."""
|
|
reg = get_contract_registry()
|
|
contract = reg.get_contract("kommunikation")
|
|
assert contract is not None
|
|
assert contract.contract_name == "kommunikation"
|
|
|
|
def test_lazy_load_missing_plugin_returns_none(self):
|
|
"""Lazy load of non-existent plugin returns None."""
|
|
reg = get_contract_registry()
|
|
assert reg.get_contract("nonexistent_plugin_xyz") is None
|