Agent Zero
282 lines
9.8 KiB
Python
282 lines
9.8 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
|