Agent Zero
579 lines
20 KiB
Python
579 lines
20 KiB
Python
# Regression tests for Block A fixes: ARCH-043 (system tenant) and ARCH-052 (storage metadata).
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
|
|
# --- ARCH-043: get_system_tenant resolves by configured slug ---
|
|
|
|
|
|
class TestGetSystemTenant:
|
|
@pytest.mark.asyncio
|
|
async def test_resolves_by_slug_not_first_row(self, db_session):
|
|
from sqlalchemy import select
|
|
|
|
from app.config import get_settings
|
|
from app.core.db import get_system_tenant
|
|
from app.models.tenant import Tenant
|
|
|
|
settings = get_settings()
|
|
slug = settings.system_tenant_slug
|
|
|
|
# Insert a decoy first and the real system tenant second,
|
|
# so a limit(1)-style query would pick the wrong row.
|
|
decoy = Tenant(name='Decoy Org', slug=f'decoy-{uuid.uuid4().hex[:8]}')
|
|
db_session.add(decoy)
|
|
system = Tenant(name='Default Org', slug=slug)
|
|
db_session.add(system)
|
|
await db_session.commit()
|
|
|
|
try:
|
|
tenant = await get_system_tenant(db_session)
|
|
assert tenant is not None
|
|
assert tenant.slug == slug
|
|
# sanity: decoy really is the first row
|
|
first = (await db_session.execute(select(Tenant).limit(1))).scalar_one()
|
|
assert first.id != tenant.id
|
|
finally:
|
|
await db_session.delete(decoy)
|
|
await db_session.delete(system)
|
|
await db_session.commit()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_none_when_missing(self, db_session):
|
|
from sqlalchemy import delete
|
|
|
|
from app.core.db import get_system_tenant
|
|
from app.models.tenant import Tenant
|
|
|
|
await db_session.execute(delete(Tenant))
|
|
await db_session.commit()
|
|
|
|
tenant = await get_system_tenant(db_session)
|
|
assert tenant is None
|
|
|
|
|
|
# --- ARCH-052: async metadata without nested event loops ---
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_storage(tmp_path, monkeypatch):
|
|
from app.core.storage import LocalStorage, reset_storage_backend
|
|
|
|
monkeypatch.setenv('STORAGE_PATH', str(tmp_path))
|
|
monkeypatch.setenv('STORAGE_BACKEND', 'local')
|
|
reset_storage_backend()
|
|
backend = LocalStorage(base_path=str(tmp_path))
|
|
yield backend
|
|
reset_storage_backend()
|
|
|
|
|
|
class TestGetFileMetadataAsync:
|
|
def _use_local(self, tmp_storage, monkeypatch):
|
|
monkeypatch.setenv('STORAGE_PATH', str(tmp_storage.base_path))
|
|
monkeypatch.setenv('STORAGE_BACKEND', 'local')
|
|
from app.core.storage import reset_storage_backend
|
|
|
|
reset_storage_backend()
|
|
import app.core.storage as storage_mod
|
|
|
|
storage_mod._storage_backend = tmp_storage
|
|
|
|
def teardown_method(self):
|
|
from app.core.storage import reset_storage_backend
|
|
|
|
reset_storage_backend()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_awaitable_inside_running_loop(self, tmp_storage, monkeypatch):
|
|
self._use_local(tmp_storage, monkeypatch)
|
|
|
|
from app.core.storage import get_file_metadata_async
|
|
|
|
await tmp_storage.save('meta/async.txt', b'async metadata')
|
|
|
|
meta = await get_file_metadata_async('meta/async.txt')
|
|
assert meta['exists'] is True
|
|
assert meta['size'] == len(b'async metadata')
|
|
assert meta['modified'] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_existing_inside_running_loop(self, tmp_storage, monkeypatch):
|
|
self._use_local(tmp_storage, monkeypatch)
|
|
|
|
from app.core.storage import get_file_metadata_async
|
|
|
|
meta = await get_file_metadata_async('nonexistent/async.txt')
|
|
assert meta['exists'] is False
|
|
assert meta['size'] is None
|
|
assert meta['modified'] is None
|
|
|
|
def test_sync_wrapper_still_works(self, tmp_storage, monkeypatch):
|
|
self._use_local(tmp_storage, monkeypatch)
|
|
|
|
from app.core.storage import get_file_metadata
|
|
|
|
asyncio.run(tmp_storage.save('meta/sync.txt', b'sync metadata'))
|
|
meta = get_file_metadata('meta/sync.txt')
|
|
assert meta['exists'] is True
|
|
assert meta['size'] == len(b'sync metadata')
|
|
|
|
def test_sync_wrapper_non_existing(self, tmp_storage, monkeypatch):
|
|
self._use_local(tmp_storage, monkeypatch)
|
|
|
|
from app.core.storage import get_file_metadata
|
|
|
|
meta = get_file_metadata('nonexistent/sync.txt')
|
|
assert meta == {'size': None, 'modified': None, 'exists': False}
|
|
|
|
|
|
# --- ARCH-008/009: 2-segment permission canonical schema ---
|
|
|
|
|
|
import re as _re
|
|
|
|
from app.core.permissions import _matches_permission
|
|
|
|
_THREE_SEGMENT = _re.compile(r'^[a-z_]+:[a-z_]+:[a-z_*]+$')
|
|
|
|
|
|
class TestMatchesPermissionCanonical:
|
|
def test_two_segment_wildcards(self):
|
|
assert _matches_permission('*:*', 'contacts:read') is True
|
|
assert _matches_permission('contacts:*', 'contacts:read') is True
|
|
assert _matches_permission('*:read', 'contacts:read') is True
|
|
assert _matches_permission('contacts:read', 'contacts:read') is True
|
|
|
|
def test_three_segment_grant_never_matches(self):
|
|
assert _matches_permission('core:*:read', 'contacts:read') is False
|
|
|
|
def test_segment_count_mismatch(self):
|
|
assert _matches_permission('contacts:read', 'contacts:read:extra') is False
|
|
|
|
|
|
class TestRouteLiteralsTwoSegment:
|
|
def test_no_three_segment_literals_in_routes(self):
|
|
import pathlib
|
|
|
|
routes_dir = pathlib.Path('/a0/usr/projects/leocrm/app/routes')
|
|
offenders = []
|
|
for py_file in sorted(routes_dir.glob('*.py')):
|
|
text = py_file.read_text()
|
|
for m in _re.finditer(r'require_permission\(([^)]*)\)', text):
|
|
arg = m.group(1).strip()
|
|
if not arg:
|
|
continue
|
|
value = arg.strip('\'"')
|
|
if ':' in value and value.count(':') != 1:
|
|
offenders.append(f'{py_file.name}: {value}')
|
|
assert offenders == [], f'3-segment permission literals found: {offenders}'
|
|
|
|
|
|
class TestManifestPermissionValidator:
|
|
def test_valid_permissions_accepted(self):
|
|
from app.plugins.manifest import PluginManifest
|
|
|
|
manifest = PluginManifest(
|
|
name='x',
|
|
version='1.0.0',
|
|
display_name='X',
|
|
permissions=['contacts:read', '*:*'],
|
|
)
|
|
assert manifest.permissions == ['contacts:read', '*:*']
|
|
|
|
def test_three_segment_rejected(self):
|
|
from pydantic import ValidationError
|
|
|
|
from app.plugins.manifest import PluginManifest
|
|
|
|
with pytest.raises(ValidationError):
|
|
PluginManifest(
|
|
name='x',
|
|
version='1.0.0',
|
|
display_name='X',
|
|
permissions=['core:contacts:read'],
|
|
)
|
|
|
|
|
|
# --- A2 deactivation cleanup: lifecycle symmetry ---
|
|
|
|
|
|
class TestServiceContainerRemove:
|
|
def test_remove_registered_service(self):
|
|
from app.core.service_container import ServiceContainer
|
|
|
|
container = ServiceContainer()
|
|
container.register('svc', object())
|
|
assert container.has('svc') is True
|
|
container.remove('svc')
|
|
assert container.has('svc') is False
|
|
|
|
def test_remove_absent_service_is_noop(self):
|
|
from app.core.service_container import ServiceContainer
|
|
|
|
container = ServiceContainer()
|
|
container.remove('never_registered') # must not raise
|
|
assert container.has('never_registered') is False
|
|
|
|
|
|
class TestKommunikationLifecycle:
|
|
@pytest.mark.asyncio
|
|
async def test_deactivate_removes_container_services(self):
|
|
from app.core.event_bus import EventBus
|
|
from app.core.service_container import ServiceContainer
|
|
from app.plugins.builtins.kommunikation.plugin import KommunikationPlugin
|
|
|
|
plugin = KommunikationPlugin()
|
|
container = ServiceContainer()
|
|
event_bus = EventBus()
|
|
|
|
await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
|
|
assert container.has('comm_websocket') is True
|
|
assert container.has('comm_miniapps') is True
|
|
|
|
await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
|
|
assert container.has('comm_websocket') is False
|
|
assert container.has('comm_miniapps') is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reactivate_no_duplicates(self):
|
|
from app.core.event_bus import EventBus
|
|
from app.core.service_container import ServiceContainer
|
|
from app.plugins.builtins.kommunikation.plugin import KommunikationPlugin
|
|
|
|
plugin = KommunikationPlugin()
|
|
container = ServiceContainer()
|
|
event_bus = EventBus()
|
|
|
|
await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
|
|
await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
|
|
await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
|
|
|
|
ws = container.get('comm_websocket')
|
|
miniapps = container.get('comm_miniapps')
|
|
assert ws is not None and miniapps is not None
|
|
assert len(miniapps._apps) == len(set(miniapps._apps.keys()))
|
|
|
|
|
|
class TestWikiProviderLifecycle:
|
|
@pytest.mark.asyncio
|
|
async def test_deactivate_unregisters_search_provider(self):
|
|
from app.core.event_bus import EventBus
|
|
from app.core.service_container import ServiceContainer
|
|
from app.plugins.builtins.contracts import get_contract
|
|
from app.plugins.builtins.wiki.plugin import WikiPlugin
|
|
|
|
plugin = WikiPlugin()
|
|
container = ServiceContainer()
|
|
event_bus = EventBus()
|
|
|
|
search_contract = get_contract('unified_search')
|
|
registry = search_contract.get_search_registry()
|
|
|
|
await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
|
|
assert registry.get('wiki_article') is not None
|
|
|
|
await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
|
|
assert registry.get('wiki_article') is None
|
|
|
|
|
|
# --- Gate A: Block A completion proof (docs/fix-plan-v3.md) ---
|
|
|
|
|
|
class TestGateAImports:
|
|
def test_every_app_module_imports(self):
|
|
import importlib
|
|
import pathlib
|
|
|
|
app_root = pathlib.Path('/a0/usr/projects/leocrm/app')
|
|
failures = []
|
|
seen = set()
|
|
for py_file in sorted(app_root.rglob('*.py')):
|
|
rel = py_file.relative_to(app_root.parent)
|
|
if py_file.name == '__init__.py':
|
|
mod_name = str(rel.parent).replace('/', '.')
|
|
else:
|
|
mod_name = str(rel.with_suffix('')).replace('/', '.')
|
|
if not mod_name or mod_name in seen:
|
|
continue
|
|
seen.add(mod_name)
|
|
try:
|
|
importlib.import_module(mod_name)
|
|
except Exception as exc:
|
|
failures.append(f'{mod_name}: {type(exc).__name__}: {exc}')
|
|
assert failures == [], (
|
|
f'{len(failures)} modules failed to import: ' + ' | '.join(failures)
|
|
)
|
|
|
|
|
|
def _build_gate_a_plugin():
|
|
from app.plugins.base import BasePlugin
|
|
from app.plugins.manifest import PluginManifest
|
|
|
|
class GateATestPlugin(BasePlugin):
|
|
manifest = PluginManifest(
|
|
name='gate_a_test',
|
|
version='1.0.0',
|
|
display_name='Gate A Test',
|
|
description='Proves lifecycle symmetry.',
|
|
dependencies=[],
|
|
routes=[],
|
|
events=['gate_a.event'],
|
|
migrations=[],
|
|
permissions=['gate_a_test:read'],
|
|
)
|
|
|
|
async def on_activate(self, db, service_container, event_bus) -> None:
|
|
await super().on_activate(db, service_container, event_bus)
|
|
service_container.register('gate_a_marker', object())
|
|
|
|
async def on_deactivate(self, db, service_container, event_bus) -> None:
|
|
if service_container.has('gate_a_marker'):
|
|
service_container.remove('gate_a_marker')
|
|
await super().on_deactivate(db, service_container, event_bus)
|
|
|
|
return GateATestPlugin()
|
|
|
|
|
|
class TestGateALifecycleSymmetry:
|
|
@pytest.mark.asyncio
|
|
async def test_activate_registers_deactivate_deregisters(self):
|
|
from app.core.event_bus import EventBus
|
|
from app.core.service_container import ServiceContainer
|
|
|
|
plugin = _build_gate_a_plugin()
|
|
container = ServiceContainer()
|
|
event_bus = EventBus()
|
|
|
|
await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
|
|
assert container.has('gate_a_marker') is True
|
|
assert len(plugin._event_handlers) == 1
|
|
|
|
await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
|
|
assert container.has('gate_a_marker') is False
|
|
assert plugin._event_handlers == {}
|
|
|
|
# re-activate: no duplicate handlers
|
|
await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
|
|
assert len(plugin._event_handlers) == 1
|
|
await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
|
|
|
|
|
|
class TestGateAActivateOnce:
|
|
def test_activation_outside_tenant_loop_in_main(self):
|
|
# ARCH-002 contract: main.py activates each plugin once per process,
|
|
# never inside a per-tenant loop.
|
|
import pathlib
|
|
|
|
main_py = pathlib.Path('/a0/usr/projects/leocrm/app/main.py')
|
|
lines = main_py.read_text().splitlines()
|
|
activate_lines = [i for i, line in enumerate(lines) if 'on_activate(' in line and 'def ' not in line]
|
|
assert activate_lines, 'no plugin on_activate call found in main.py'
|
|
for i in activate_lines:
|
|
# scan upward to the nearest loop header; it must be a plugin
|
|
# loop, never a tenant loop
|
|
for j in range(i - 1, max(0, i - 40), -1):
|
|
stripped = lines[j].strip()
|
|
if stripped.startswith('for ') or stripped.startswith('async for '):
|
|
assert 'tenant' not in stripped.split('in')[0], (
|
|
f'on_activate sits inside a tenant loop: main.py line {j + 1}'
|
|
)
|
|
break
|
|
|
|
|
|
class TestGateAContractRoundtrip:
|
|
def test_register_get_unregister_no_resurrect(self):
|
|
from app.plugins.builtins.contracts import (
|
|
get_contract,
|
|
get_contract_registry,
|
|
reset_contract_registry_for_testing,
|
|
)
|
|
|
|
reset_contract_registry_for_testing()
|
|
registry = get_contract_registry()
|
|
|
|
class Dummy:
|
|
contract_name = 'gate_a_dummy'
|
|
|
|
registry.register('gate_a_dummy', Dummy())
|
|
assert isinstance(get_contract('gate_a_dummy'), Dummy)
|
|
|
|
registry.unregister('gate_a_dummy')
|
|
assert get_contract('gate_a_dummy') is None
|
|
# second read must NOT lazily resurrect the unregistered contract
|
|
assert get_contract('gate_a_dummy') is None
|
|
|
|
reset_contract_registry_for_testing()
|
|
|
|
|
|
# --- Gate B: Block B completion proof (docs/fix-plan-v3.md) ---
|
|
|
|
|
|
def _install_inline_route_module():
|
|
"""Register a throwaway router module in sys.modules (no file on disk)."""
|
|
import sys
|
|
import types
|
|
|
|
from fastapi import APIRouter
|
|
|
|
mod_name = 'gate_b_inline_routes'
|
|
if mod_name in sys.modules:
|
|
return sys.modules[mod_name]
|
|
mod = types.ModuleType(mod_name)
|
|
router = APIRouter(prefix='/api/v1/gate-b-inline', tags=['gate-b-inline'])
|
|
|
|
@router.get('/ping')
|
|
async def ping():
|
|
return {'pong': 'gate-b'}
|
|
|
|
mod.router = router
|
|
sys.modules[mod_name] = mod
|
|
return mod
|
|
|
|
|
|
class TestGateBNewPluginNoCoreChanges:
|
|
def test_inline_plugin_route_mounted_and_entity_registered(self):
|
|
import uuid
|
|
|
|
from fastapi import Depends, FastAPI
|
|
from sqlalchemy import Column, Integer, String, Uuid
|
|
|
|
from app.core.db import Base
|
|
from app.deps import require_active_plugin
|
|
from app.plugins.base import BasePlugin
|
|
from app.plugins.manifest import PluginManifest, PluginRouteDef
|
|
from app.plugins.registry import get_registry
|
|
from app.services.entity_permission_service import (
|
|
ENTITY_MODELS,
|
|
register_entity_model,
|
|
unregister_entity_model,
|
|
)
|
|
|
|
_install_inline_route_module()
|
|
|
|
# Classic Column style: Mapped[] annotations cannot be resolved for
|
|
# classes defined inside a function (no module-level globals).
|
|
class GateBInlineThing(Base):
|
|
__tablename__ = 'gate_b_inline_things'
|
|
id = Column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
tenant_id = Column(Uuid(as_uuid=True), nullable=False)
|
|
name = Column(String(100))
|
|
sort_order = Column(Integer)
|
|
|
|
class GateBInlinePlugin(BasePlugin):
|
|
manifest = PluginManifest(
|
|
name='gate_b_inline',
|
|
version='1.0.0',
|
|
display_name='Gate B Inline',
|
|
description='Proves new-plugin capability without core changes.',
|
|
dependencies=[],
|
|
routes=[
|
|
PluginRouteDef(
|
|
path='/api/v1/gate-b-inline',
|
|
module='gate_b_inline_routes',
|
|
router_attr='router',
|
|
)
|
|
],
|
|
events=[],
|
|
migrations=[],
|
|
permissions=['gate_b_inline:read'],
|
|
)
|
|
|
|
def get_entity_models(self):
|
|
return {'gate_b_thing': GateBInlineThing}
|
|
|
|
plugin = GateBInlinePlugin()
|
|
registry = get_registry()
|
|
registry._plugins['gate_b_inline'] = plugin
|
|
|
|
try:
|
|
# 1) Route gets mounted by the plugin-route mechanism
|
|
app = FastAPI()
|
|
for route_def in plugin.manifest.routes:
|
|
import importlib
|
|
|
|
router_module = importlib.import_module(route_def.module)
|
|
router = getattr(router_module, route_def.router_attr)
|
|
app.include_router(
|
|
router, dependencies=[Depends(require_active_plugin('gate_b_inline'))]
|
|
)
|
|
paths = app.openapi()['paths']
|
|
assert '/api/v1/gate-b-inline/ping' in paths
|
|
|
|
# 2) Entity registers through the standard activation path
|
|
register_entity_model('gate_b_thing', GateBInlineThing)
|
|
try:
|
|
assert ENTITY_MODELS['gate_b_thing'] is GateBInlineThing
|
|
finally:
|
|
unregister_entity_model('gate_b_thing')
|
|
finally:
|
|
registry._plugins.pop('gate_b_inline', None)
|
|
|
|
|
|
class TestGateBDependencyBlockade:
|
|
@pytest.mark.asyncio
|
|
async def test_deactivate_blocked_while_dependent_active(self, db_session):
|
|
from app.models.plugin import Plugin as PluginModel
|
|
from app.plugins.base import BasePlugin
|
|
from app.plugins.manifest import PluginManifest
|
|
from app.plugins.registry import get_registry
|
|
|
|
class DepB(BasePlugin):
|
|
manifest = PluginManifest(
|
|
name='gate_b_dep_b',
|
|
version='1.0.0',
|
|
display_name='Dep B',
|
|
dependencies=[],
|
|
routes=[],
|
|
events=[],
|
|
migrations=[],
|
|
permissions=[],
|
|
)
|
|
|
|
class DepA(BasePlugin):
|
|
manifest = PluginManifest(
|
|
name='gate_b_dep_a',
|
|
version='1.0.0',
|
|
display_name='Dep A',
|
|
dependencies=['gate_b_dep_b'],
|
|
routes=[],
|
|
events=[],
|
|
migrations=[],
|
|
permissions=[],
|
|
)
|
|
|
|
registry = get_registry()
|
|
registry._plugins['gate_b_dep_b'] = DepB()
|
|
registry._plugins['gate_b_dep_a'] = DepA()
|
|
|
|
records = {}
|
|
try:
|
|
for name in ('gate_b_dep_b', 'gate_b_dep_a'):
|
|
rec = PluginModel(
|
|
name=name,
|
|
display_name=name,
|
|
version='1.0.0',
|
|
status='active',
|
|
installed=True,
|
|
active=True,
|
|
is_core=False,
|
|
)
|
|
db_session.add(rec)
|
|
records[name] = rec
|
|
await db_session.commit()
|
|
|
|
# Deactivating the dependency must be blocked because gate_b_dep_a
|
|
# is active and depends on it.
|
|
with pytest.raises(ValueError, match='depend'):
|
|
await registry.deactivate(db_session, 'gate_b_dep_b')
|
|
finally:
|
|
for rec in records.values():
|
|
await db_session.delete(rec)
|
|
await db_session.commit()
|
|
registry._plugins.pop('gate_b_dep_b', None)
|
|
registry._plugins.pop('gate_b_dep_a', None)
|