leocrm/tests/test_arch_block_a.py

# 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()