"""Regression tests for the external architecture audit (leocrm-full.zip, 2026-09). Covers the confirmed findings with live counterchecks (AGENTS.md rule — no finding is marked done without a measurement): - F1 list_workspaces() returns modules (WorkspaceManager overwrite bug) - F2 active-manifests respects tenant deactivation (signature level) - F4 ContractRegistry fails closed for DB-inactive plugins (restart edge case) and reopens on activation - F5 field definitions have a full lifecycle (register/unregister) - F9 permission catalog: every used key is registered - F10 contact field definitions are plugin-owned (not in core list) - F11 forgejo_error_reporter is not is_core - F12 dms is a platform core plugin (ADR-020) - F13 core worker has no contact import; contacts trash cleanup is a plugin job discovered via get_job_modules() - F14 the legacy dsgvo-export route delegates to the single DSAR collector - F15 get_entity_read_permission fails closed instead of contacts:read The frontend architecture items (static plugin routes, STATIC_COMPONENT_MAP, widgetRegistry) are tracked as Phase Q — not covered here. """ from __future__ import annotations import uuid import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.models.tenant import Tenant from app.models.user import User, UserTenant from app.services import workspace_service @pytest.fixture(autouse=True) def _restore_global_registries(): """Snapshot & restore global registries mutated by these tests. The F5 lifecycle test deactivates the real 'contacts' plugin through PluginService, which (since the audit fix) centrally unregisters its CONTRACT and clears permission-registry state. The conftest resets entity models but NOT the contract registry — without this restore, get_contract('contacts') stays fail-closed for every later test in the same process (suite-order dependence). """ from app.plugins.builtins.contracts import get_contract_registry from app.core.permission_registry import get_permission_registry from app.services.entity_permission_service import ( ENTITY_MODELS, ENTITY_PLUGIN_OWNERS, ) perm_reg = get_permission_registry() saved_active = set(perm_reg._active_plugins) saved_plugin_perms = {k: list(v) for k, v in perm_reg._plugin_permissions.items()} saved_field_defs = {k: list(v) for k, v in perm_reg._field_definitions.items()} cr = get_contract_registry() saved_contracts = dict(cr._contracts) saved_loaded = set(cr._loaded) saved_unreg = set(cr._unregistered) saved_db_inactive = set(cr._db_inactive) saved_entity_models = dict(ENTITY_MODELS) saved_owners = dict(ENTITY_PLUGIN_OWNERS) yield perm_reg._active_plugins.clear() perm_reg._active_plugins.update(saved_active) perm_reg._plugin_permissions.clear() perm_reg._plugin_permissions.update(saved_plugin_perms) perm_reg._field_definitions.clear() perm_reg._field_definitions.update(saved_field_defs) cr._contracts.clear() cr._contracts.update(saved_contracts) cr._loaded.clear() cr._loaded.update(saved_loaded) cr._unregistered.clear() cr._unregistered.update(saved_unreg) cr._db_inactive.clear() cr._db_inactive.update(saved_db_inactive) ENTITY_MODELS.clear() ENTITY_MODELS.update(saved_entity_models) ENTITY_PLUGIN_OWNERS.clear() ENTITY_PLUGIN_OWNERS.update(saved_owners) # ── F1: list_workspaces returns modules ───────────────────────────────────── async def _seed_tenant_and_user(db: AsyncSession) -> dict: """Seed a tenant and a user, return IDs (test_workspaces.py pattern).""" from app.core.auth import hash_password tenant = Tenant(name="Audit Tenant", slug=f"audit-{uuid.uuid4().hex[:8]}") db.add(tenant) await db.flush() user = User( email=f"audit-{uuid.uuid4().hex[:8]}@example.com", name="Audit User", password_hash=hash_password("TestPass123!"), is_active=True, preferences={}, ) db.add(user) await db.flush() db.add(UserTenant(user_id=user.id, tenant_id=tenant.id, is_default=True, role="admin")) await db.flush() return {"tenant": tenant, "user": user} @pytest.mark.asyncio async def test_f1_list_workspaces_returns_modules(db_session: AsyncSession): """Audit F1: list_workspaces must include configured modules. Previously every workspace came back with modules: [] and the WorkspaceManager editor overwrote existing config as all-hidden. """ seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="AuditWS", ) ws_id = uuid.UUID(created["id"]) await workspace_service.set_workspace_modules( db_session, seed["tenant"].id, ws_id, [ {"module_key": "contacts", "is_visible": True, "menu_order": 10, "config": {}}, {"module_key": "calendar", "is_visible": False, "menu_order": 20, "config": {}}, ], ) result = await workspace_service.list_workspaces(db_session, seed["tenant"].id) assert result["total"] == 1 listed = result["items"][0] assert listed["name"] == "AuditWS" # THE regression: modules must be present with their is_visible flags mods = {m["module_key"]: m["is_visible"] for m in listed["modules"]} assert mods == {"contacts": True, "calendar": False} # ── F2: active-manifests tenant filtering (signature level) ──────────────── def test_f2_get_active_manifests_accepts_tenant_id(): """Audit F2: the registry manifest API accepts and honours tenant_id.""" import inspect from app.plugins.registry import PluginRegistry sig = inspect.signature(PluginRegistry.get_active_manifests) assert "tenant_id" in sig.parameters, ( "get_active_manifests must accept tenant_id so tenant-deactivated " "plugins are hidden from the UI manifests" ) # Service passes it through from app.services.plugin_service import PluginService sig = inspect.signature(PluginService.get_active_manifests) assert "tenant_id" in sig.parameters @pytest.mark.asyncio async def test_f2_manifests_exclude_tenant_disabled_plugins(db_session: AsyncSession): """Audit F2: a plugin deactivated for the tenant is filtered out.""" from sqlalchemy import text seed = await _seed_tenant_and_user(db_session) # tenant_plugin_activation is created by migration 0066 only (no ORM # model) — test DBs built via create_all lack it, so create it if needed. await db_session.execute( text( "CREATE TABLE IF NOT EXISTS tenant_plugin_activation (" "id UUID PRIMARY KEY DEFAULT gen_random_uuid(), " "tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, " "plugin_name VARCHAR(100) NOT NULL, " "is_active BOOLEAN NOT NULL DEFAULT true, " "activated_by UUID, " "created_at TIMESTAMPTZ NOT NULL DEFAULT now(), " "updated_at TIMESTAMPTZ NOT NULL DEFAULT now())" ) ) await db_session.execute( text( "INSERT INTO tenant_plugin_activation (tenant_id, plugin_name, is_active) " "VALUES (:tid, 'forgejo_error_reporter', false)" ), {"tid": str(seed["tenant"].id)}, ) await db_session.flush() from app.models.plugin import Plugin as PluginModel from app.plugins.registry import get_registry registry = get_registry() if not registry.list_discovered(): registry.discover_builtins() # Seed a globally-ACTIVE plugin record so the manifest is served at all # (test DBs have no plugin rows unless installed through the registry). db_session.add(PluginModel( name="forgejo_error_reporter", display_name="Forgejo Error Reporter", version="1.0.0", status="active", installed=True, active=True, is_core=False, )) await db_session.flush() all_manifests = await registry.get_active_manifests(db_session, tenant_id=None) assert any(m["name"] == "forgejo_error_reporter" for m in all_manifests), ( "globally active plugin must appear in manifests (test premise)" ) filtered = await registry.get_active_manifests( db_session, tenant_id=seed["tenant"].id ) assert all(m["name"] != "forgejo_error_reporter" for m in filtered), ( "tenant-deactivated plugin must not appear in the UI manifests" ) # ── F4: contract registry restart guard ──────────────────────────────────── def test_f4_contract_registry_fails_closed_for_db_inactive(): """Audit F4: contracts of DB-inactive plugins are not resurrected.""" from app.plugins.builtins.contracts import get_contract, get_contract_registry reg = get_contract_registry() reg._reset_for_testing() try: reg.mark_db_inactive({"kommunikation"}) # Even after lazy-load attempts the contract must stay unavailable assert get_contract("kommunikation") is None finally: reg._reset_for_testing() def test_f4_contract_registry_reopens_on_activation(): """Audit F4: activating a plugin clears its inactive markers.""" from app.plugins.builtins.contracts import get_contract_registry reg = get_contract_registry() reg._reset_for_testing() try: reg.mark_db_inactive({"some_plugin"}) reg._unregistered.add("some_plugin") reg.mark_plugin_active("some_plugin") assert "some_plugin" not in reg._db_inactive assert "some_plugin" not in reg._unregistered finally: reg._reset_for_testing() # ── F5: field definitions lifecycle ───────────────────────────────────────── def test_f5_field_definitions_full_lifecycle(): """Audit F5: register → present, unregister → gone.""" from app.core.permission_registry import PermissionRegistry reg = PermissionRegistry() reg.initialize() defs = [{"module": "x", "field": "y", "label": "Y", "sensitivity": "normal"}] reg.register_field_definitions("audit_plugin", defs) assert any(d.get("plugin_key", d.get("field")) or True for d in defs) all_defs = reg.get_all_field_definitions() assert any(d.get("field") == "y" for d in all_defs) reg.unregister_field_definitions("audit_plugin") all_defs = reg.get_all_field_definitions() assert not any(d.get("field") == "y" and d.get("module") == "x" for d in all_defs) def test_f5_deactivate_plugin_unregisters_field_definitions(): """Audit F5: PluginService.deactivate_plugin removes field definitions.""" import asyncio from unittest.mock import AsyncMock, MagicMock from app.core.permission_registry import get_permission_registry, init_permission_registry from app.plugins.builtins.contacts.plugin import ContactsPlugin from app.services.plugin_service import PluginService init_permission_registry(set()) perm_reg = get_permission_registry() perm_reg._active_plugins.add("contacts") perm_reg.register_field_definitions("contacts", ContactsPlugin().get_field_definitions()) assert any( d.get("module") == "contacts" for d in perm_reg.get_all_field_definitions() ) service = PluginService() # deactivate_plugin resolves the record via registry._get_plugin_record — # a MagicMock plugin record with active=True triggers the cleanup path. # NOTE: service._registry IS the global singleton — save & restore the # patched methods so the conftest bootstrap stays intact for later tests # (order-dependence guard). mock_record = MagicMock() mock_record.active = True mock_record.status = "active" _orig_get_record = service._registry._get_plugin_record _orig_deactivate = service._registry.deactivate _orig_get_plugin = service._registry.get_plugin service._registry._get_plugin_record = AsyncMock(return_value=mock_record) service._registry.deactivate = AsyncMock(return_value=mock_record) service._registry.get_plugin = MagicMock(return_value=ContactsPlugin()) try: asyncio.get_event_loop().run_until_complete( service.deactivate_plugin(AsyncMock(), "contacts") ) finally: service._registry._get_plugin_record = _orig_get_record service._registry.deactivate = _orig_deactivate service._registry.get_plugin = _orig_get_plugin assert not any( d.get("module") == "contacts" for d in perm_reg.get_all_field_definitions() ), "deactivation must unregister the plugin's field definitions" # ── F9: permission catalog completeness ───────────────────────────────────── def test_f9_all_audit_permission_keys_registered(): """Audit F9: the 12 previously missing keys are grantable now.""" from app.core.permission_registry import ( CORE_PERMISSIONS, get_permission_registry, init_permission_registry, ) init_permission_registry(set()) # Register from the REAL manifests (not hand-picked lists) — this catches # kwargs-level corruption like the bug where permissions=[...] had landed # inside PluginRouteDef kwargs instead of the manifest level. from app.plugins.builtins.forgejo_error_reporter.plugin import ( ForgejoErrorReporterPlugin, ) from app.plugins.builtins.permissions.plugin import PermissionsPlugin assert PermissionsPlugin().manifest.permissions == [ "permissions:read", "permissions:admin", ] assert ForgejoErrorReporterPlugin().manifest.permissions == ["system:read"], ( "system:read must be declared at MANIFEST level (not PluginRouteDef kwargs)" ) get_permission_registry().register_plugin_permissions( "permissions", PermissionsPlugin().manifest.permissions ) get_permission_registry().register_plugin_permissions( "forgejo_error_reporter", ForgejoErrorReporterPlugin().manifest.permissions ) known = get_permission_registry().get_all() keys = {p["key"] for p in known} core_expected = [ "automation:admin", "bank-accounts:read", "bank-accounts:write", "delegations:read", "delegations:write", "policies:read", "policies:write", "templates:read", "templates:write", ] for key in core_expected: assert key in {p["key"] for p in CORE_PERMISSIONS}, f"{key} missing in CORE_PERMISSIONS" for key in ["permissions:read", "permissions:admin", "system:read"]: assert key in keys, f"{key} must be grantable" assert get_permission_registry().is_valid("delegations:read") # ── F10: contact field definitions are plugin-owned ──────────────────────── def test_f10_contact_field_definitions_plugin_owned(): """Audit F10: contacts:* field defs come from the plugin manifest.""" from app.core.permission_registry import CORE_FIELD_DEFINITIONS from app.plugins.builtins.contacts.plugin import ContactsPlugin core_modules = {fd["module"] for fd in CORE_FIELD_DEFINITIONS} assert "contacts" not in core_modules, "core must not own contacts field definitions" plugin_defs = ContactsPlugin().get_field_definitions() assert len(plugin_defs) >= 30, "contacts plugin should carry its ~39 field definitions" assert all(fd["module"] == "contacts" for fd in plugin_defs) # Spot-check sensitive fields are still classified sensitive after the move by_field = {fd["field"]: fd for fd in plugin_defs} assert by_field["mobilephone"]["sensitivity"] == "sensitive" assert by_field["bic"]["sensitivity"] == "sensitive" # ── F11/F12: plugin classification ───────────────────────────────────────── def test_f11_forgejo_error_reporter_not_core(): """Audit F11: the test/staging-only reporter is deactivatable.""" from app.plugins.builtins.forgejo_error_reporter.plugin import ForgejoErrorReporterPlugin assert ForgejoErrorReporterPlugin().manifest.is_core is False def test_f12_dms_is_platform_core_plugin(): """Audit F12/ADR-020: the core schema builds on the DMS files table.""" from app.plugins.builtins.dms.plugin import DmsPlugin assert DmsPlugin().manifest.is_core is True # ── F13: worker decoupled from contacts ──────────────────────────────────── def test_f13_core_worker_has_no_contact_import(): """Audit F13: app/core/worker.py must not import contact models.""" import inspect import app.core.worker as worker src = inspect.getsource(worker) assert "from app.models.contact import" not in src def test_f13_contacts_trash_job_discovered_via_job_modules(): """Audit F13: contacts plugin owns its trash cleanup job.""" from app.core.job_registry import get_job from app.plugins.builtins.contacts.jobs import cleanup_contacts_trash_job from app.plugins.builtins.contacts.plugin import ContactsPlugin modules = ContactsPlugin().get_job_modules() assert "app.plugins.builtins.contacts.jobs" in modules import importlib importlib.import_module("app.plugins.builtins.contacts.jobs") assert get_job("cleanup_contacts_trash") is cleanup_contacts_trash_job # ── F14: single DSAR export path ─────────────────────────────────────────── def test_f14_dsgvo_route_delegates_to_dsar_collector(): """Audit F14: the legacy export route delegates, it does not reimplement.""" import inspect import app.routes.system_settings as ss src = inspect.getsource(ss) assert "_dsar_collect_user_data" in src, "route must delegate to the DSAR collector" assert "from app.models.contact import" not in src, ( "the export route must not know contacts internals" ) # ── F15: entity permission fallback fails closed ─────────────────────────── def test_f15_entity_read_permission_fails_closed(): """Audit F15: unmapped entities no longer default to contacts:read.""" from app.services.entity_permission_service import ( ENTITY_PLUGIN_OWNERS, get_entity_read_permission, ) saved = ENTITY_PLUGIN_OWNERS.pop("zzz_audit_unknown", None) try: perm = get_entity_read_permission("zzz_audit_unknown") assert perm != "contacts:read" assert perm == "__unmapped__:read" finally: if saved is not None: ENTITY_PLUGIN_OWNERS["zzz_audit_unknown"] = saved def test_f15_contact_folder_registered_by_plugin(): """Audit F15: contact_folder comes from ContactsPlugin, not core.""" from app.plugins.builtins.contacts.plugin import ContactsPlugin models = ContactsPlugin().get_entity_models() assert "contact_folder" in models from app.services.entity_permission_service import get_entity_read_permission assert get_entity_read_permission("contact_folder") == "contacts:read" # ── Uninstall lifecycle (audit P1) ───────────────────────────────────────── def test_f3_uninstall_deactivates_via_service_first(): """Audit P1: uninstall_plugin runs the full service deactivation first.""" import inspect from app.services.plugin_service import PluginService src = inspect.getsource(PluginService.uninstall_plugin) assert "self.deactivate_plugin(" in src, ( "uninstall must run the full service-level deactivation (permissions, " "active set, entity models) before registry.uninstall()" )