Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
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,
|
|
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
|