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'],
            )