leocrm/tests/test_audit_architecture_fixes.py

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