Leopoldadmin/leocrm
1
0
Fork 0
Code Issues 363 Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

feat(K): Phase K EU Compliance — AI Registry, DPIA, Incident Register, Retention Admin, Tests, Doku

Browse Source 
- K-REG: GET /api/v1/compliance/ai-registry — lists all agents with ai_use_case_metadata
- K-DPIA: GET /api/v1/compliance/dpia-template — pre-filled DPIA template export
- K-INC: ComplianceIncident model, Migration 0133 (RLS), CRUD routes (admin-only)
- K-RET: GET/PATCH /api/v1/compliance/retention-policies — 5 policies editable
- K-COMP-TEST: 12/12 integration tests pass
- K-DOC: docs/compliance.md — Betriebsdoku
- Frontend: ComplianceTab.tsx in SettingsAI.tsx (new tab)
- 13 files created/modified
This commit is contained in:
Agent Zero
2026-08-21 01:58:46 +02:00
parent fbcfbbced6
commit e59db34a6d
14 changed files with 1917 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+1
View File
@@ -39,6 +39,7 @@ from app.core.db import Base, close_engine, reset_engine_for_testing
from app.core.service_container import get_container # noqa: F401
from app.main import create_app
from app.models.ai_conversation import AIConversation, AIMessage # noqa: F401
from app.models.compliance import ComplianceIncident # noqa: F401
from app.models.contact import Contact, ContactPerson # noqa: F401
from app.models.contact_merge import ContactMergeHistory # noqa: F401
from app.models.plugin import Plugin, PluginMigration # noqa: F401
tests/test_phase_k_compliance.py
+502
View File
@@ -0,0 +1,502 @@
"""Phase K Compliance Tests — integration tests with real DB.
Covers:
- K-REG: AI registry lists all agents with metadata
- K-DPIA: DPIA template export
- K-INC: Incident CRUD (create, list, update)
- K-RET: Retention policies list
- Tenant isolation for compliance incidents
- Admin-only enforcement (non-admin gets 403)
Uses the real PostgreSQL test DB (conftest fixtures).
"""
from __future__ import annotations
import uuid
from datetime import UTC, datetime
from typing import Any
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
from app.core.db import close_engine, reset_engine_for_testing
from app.core.permission_registry import init_permission_registry
from app.core.service_container import get_container
from app.main import create_app
from app.models.compliance import ComplianceIncident
from app.models.system_settings import SystemSettings
from app.models.tenant import Tenant
from app.models.user import User
from app.plugins.builtins.automation.models import AgentDefinition
from app.plugins.registry import reset_registry_for_testing
from app.services.plugin_service import reset_plugin_service_for_testing
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
pytestmark = pytest.mark.asyncio
@pytest_asyncio.fixture(scope="session", autouse=True)
async def _ensure_compliance_tables(engine: AsyncEngine):
"""Create compliance_incidents table and ensure system_settings has required columns."""
from app.core.db import Base
from sqlalchemy import text
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Ensure system_settings has backup_enabled and retention_config columns
# (test DB may have been created from an older schema)
result = await conn.execute(text(
"SELECT column_name FROM information_schema.columns "
"WHERE table_name='system_settings' AND column_name='backup_enabled'"
))
if result.rowcount == 0:
await conn.execute(text(
"ALTER TABLE system_settings ADD COLUMN backup_enabled BOOLEAN NOT NULL DEFAULT false"
))
result = await conn.execute(text(
"SELECT column_name FROM information_schema.columns "
"WHERE table_name='system_settings' AND column_name='retention_config'"
))
if result.rowcount == 0:
await conn.execute(text(
"ALTER TABLE system_settings ADD COLUMN retention_config JSONB DEFAULT '{}'::jsonb"
))
yield
@pytest_asyncio.fixture(autouse=True)
async def _init_perms():
"""Ensure permission registry is initialized for every test."""
init_permission_registry(
active_plugin_names={
"permissions",
"automation",
}
)
yield
# ──────────────────────────────────────────────────────────────────────────
# API fixtures
# ──────────────────────────────────────────────────────────────────────────
@pytest_asyncio.fixture
async def compliance_app(engine: AsyncEngine, redis_client):
"""FastAPI app with automation + permissions plugins registered and active."""
reset_engine_for_testing(engine)
app = create_app()
registry = reset_registry_for_testing()
registry.initialize(engine, app)
init_permission_registry(
active_plugin_names={
"permissions",
"automation",
}
)
container = get_container()
await container.initialize()
from app.plugins.builtins.permissions.plugin import PermissionsPlugin
from app.plugins.builtins.automation.plugin import AutomationPlugin
registry.register_plugin(PermissionsPlugin())
registry.register_plugin(AutomationPlugin())
reset_plugin_service_for_testing(registry)
sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
async with sf() as session:
await registry.install(session, "permissions")
await registry.activate(session, "permissions")
await registry.install(session, "automation")
await registry.activate(session, "automation")
await session.commit()
yield app
await close_engine()
@pytest_asyncio.fixture
async def compliance_client(compliance_app) -> AsyncClient:
"""HTTP test client with compliance routes available."""
transport = ASGITransport(app=compliance_app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
@pytest_asyncio.fixture
async def admin_authed_client(
compliance_client: AsyncClient, db_session: AsyncSession
) -> tuple[AsyncClient, dict]:
"""Authenticated admin client with seeded data."""
seed = await seed_tenant_and_users(db_session)
# Grant is_system_admin so require_permission('system:admin') passes
await db_session.execute(
update(User).where(User.id == seed["admin_a"].id).values(is_system_admin=True)
)
await db_session.commit()
await login_client(compliance_client, "admin@tenanta.com")
return compliance_client, seed
@pytest_asyncio.fixture
async def viewer_authed_client(
compliance_client: AsyncClient, db_session: AsyncSession
) -> tuple[AsyncClient, dict]:
"""Authenticated non-admin (viewer) client."""
seed = await seed_tenant_and_users(db_session)
await login_client(compliance_client, "viewer@tenanta.com")
return compliance_client, seed
# ──────────────────────────────────────────────────────────────────────────
# Helpers
# ──────────────────────────────────────────────────────────────────────────
async def _create_agent(db, tenant_id, user_id, **overrides):
"""Create a real AgentDefinition in the DB and return it."""
agent = AgentDefinition(
tenant_id=tenant_id,
name=overrides.get("name", "Test Agent"),
description=overrides.get("description", "A test agent"),
system_prompt=overrides.get("system_prompt", "You are a helpful assistant."),
tool_ids=overrides.get("tool_ids", []),
temperature=overrides.get("temperature", 0.3),
max_tokens=overrides.get("max_tokens", 1000),
max_steps=overrides.get("max_steps", 20),
created_by=user_id,
ai_use_case_metadata=overrides.get("ai_use_case_metadata", {}),
)
db.add(agent)
await db.flush()
return agent
async def _create_system_settings(db, tenant_id):
"""Create minimal system settings for a tenant."""
settings = SystemSettings(
tenant_id=tenant_id,
company_name="Test Company",
company_street="Test St",
company_city="Test City",
company_zip="12345",
company_country="DE",
)
db.add(settings)
await db.flush()
return settings
# ──────────────────────────────────────────────────────────────────────────
# K-REG: AI Registry Tests
# ──────────────────────────────────────────────────────────────────────────
class TestAIRegistry:
async def test_ai_registry_lists_all_agents(self, admin_authed_client):
"""GET /api/v1/compliance/ai-registry returns all agents with metadata."""
client, seed = admin_authed_client
# Create an agent via the automation API with use-case metadata
resp = await client.post(
"/api/v1/agents/",
json={
"name": "Compliance Test Agent",
"description": "Test agent for compliance",
"system_prompt": "You are a test assistant.",
"ai_use_case_metadata": {
"intended_purpose": "Test purpose",
"owner": "admin@test.com",
"risk_class": "medium",
"oversight_policy": "on_high_risk",
"data_categories": ["contact_data"],
"human_review_required": True,
},
},
)
assert resp.status_code == 201, f"Agent creation failed: {resp.text}"
# Now query the AI registry
resp = await client.get("/api/v1/compliance/ai-registry")
assert resp.status_code == 200, f"AI registry failed: {resp.text}"
data = resp.json()
assert "items" in data
assert "total" in data
assert data["total"] >= 1
# Check the agent appears with metadata
found = False
for item in data["items"]:
if item["name"] == "Compliance Test Agent":
found = True
meta = item["ai_use_case_metadata"]
assert meta["intended_purpose"] == "Test purpose"
assert meta["owner"] == "admin@test.com"
assert meta["risk_class"] == "medium"
assert meta["oversight_policy"] == "on_high_risk"
assert "contact_data" in meta["data_categories"]
assert meta["human_review_required"] is True
assert "validation_warnings" in item
break
assert found, "Created agent not found in AI registry"
async def test_ai_registry_admin_only(self, viewer_authed_client):
"""Non-admin user gets 403 on AI registry."""
client, seed = viewer_authed_client
resp = await client.get("/api/v1/compliance/ai-registry")
assert resp.status_code == 403, f"Expected 403, got {resp.status_code}: {resp.text}"
# ──────────────────────────────────────────────────────────────────────────
# K-DPIA: DPIA Template Tests
# ──────────────────────────────────────────────────────────────────────────
class TestDPIATemplate:
async def test_dpia_template_export(self, admin_authed_client):
"""GET /api/v1/compliance/dpia-template returns pre-filled template."""
client, seed = admin_authed_client
# Create an agent with metadata
resp = await client.post(
"/api/v1/agents/",
json={
"name": "DPIA Test Agent",
"description": "Agent for DPIA test",
"system_prompt": "You are a test assistant.",
"ai_use_case_metadata": {
"intended_purpose": "Email summarization for CRM",
"owner": "dpo@test.com",
"risk_class": "high",
"oversight_policy": "always_required",
"data_categories": ["email_content", "contact_data"],
"allowed_actions": ["read", "summarize"],
"human_review_required": True,
},
},
)
assert resp.status_code == 201, f"Agent creation failed: {resp.text}"
agent_id = resp.json()["id"]
# Get DPIA template
resp = await client.get(
"/api/v1/compliance/dpia-template",
params={"agent_id": agent_id},
)
assert resp.status_code == 200, f"DPIA template failed: {resp.text}"
data = resp.json()
assert data["use_case_id"] == agent_id
assert data["agent_name"] == "DPIA Test Agent"
assert data["intended_purpose"] == "Email summarization for CRM"
assert data["owner"] == "dpo@test.com"
assert data["risk_class"] == "high"
assert data["oversight_policy"] == "always_required"
assert "email_content" in data["data_categories"]
assert "contact_data" in data["data_categories"]
assert data["human_review_required"] is True
assert "disclaimer" in data
assert "NOT" in data["disclaimer"] or "not" in data["disclaimer"].lower()
assert "validation_warnings" in data
async def test_dpia_template_not_found(self, admin_authed_client):
"""GET /api/v1/compliance/dpia-template with invalid agent_id returns 404."""
client, seed = admin_authed_client
fake_id = str(uuid.uuid4())
resp = await client.get(
"/api/v1/compliance/dpia-template",
params={"agent_id": fake_id},
)
assert resp.status_code == 404
# ──────────────────────────────────────────────────────────────────────────
# K-INC: Incident Register Tests
# ──────────────────────────────────────────────────────────────────────────
class TestIncidentCRUD:
async def test_incident_crud(self, admin_authed_client):
"""POST/GET/PATCH /api/v1/compliance/incidents full lifecycle."""
client, seed = admin_authed_client
# Create
resp = await client.post(
"/api/v1/compliance/incidents",
json={
"incident_type": "ai",
"title": "Test AI Incident",
"description": "An AI system produced biased output",
"provider": "openai",
"measures_taken": "Disabled agent, investigating",
"status": "open",
},
)
assert resp.status_code == 201, f"Incident creation failed: {resp.text}"
incident = resp.json()
assert incident["title"] == "Test AI Incident"
assert incident["incident_type"] == "ai"
assert incident["status"] == "open"
assert incident["id"] is not None
incident_id = incident["id"]
# List
resp = await client.get("/api/v1/compliance/incidents")
assert resp.status_code == 200
data = resp.json()
assert data["total"] >= 1
found = any(i["id"] == incident_id for i in data["items"])
assert found, "Created incident not found in list"
# Update (resolve)
resp = await client.patch(
f"/api/v1/compliance/incidents/{incident_id}",
json={"status": "resolved", "measures_taken": "Fixed by retraining"},
)
assert resp.status_code == 200, f"Incident update failed: {resp.text}"
updated = resp.json()
assert updated["status"] == "resolved"
assert updated["measures_taken"] == "Fixed by retraining"
assert updated["resolved_at"] is not None
assert updated["resolved_by"] is not None
async def test_incident_admin_only(self, viewer_authed_client):
"""Non-admin user gets 403 on incidents."""
client, seed = viewer_authed_client
# GET
resp = await client.get("/api/v1/compliance/incidents")
assert resp.status_code == 403, f"Expected 403, got {resp.status_code}: {resp.text}"
# POST
resp = await client.post(
"/api/v1/compliance/incidents",
json={"title": "Should Fail", "incident_type": "ai"},
)
assert resp.status_code == 403
async def test_incident_invalid_type(self, admin_authed_client):
"""POST with invalid incident_type returns 400."""
client, seed = admin_authed_client
resp = await client.post(
"/api/v1/compliance/incidents",
json={"title": "Bad Type", "incident_type": "invalid_type"},
)
assert resp.status_code == 400
async def test_incident_not_found(self, admin_authed_client):
"""PATCH with non-existent incident returns 404."""
client, seed = admin_authed_client
fake_id = str(uuid.uuid4())
resp = await client.patch(
f"/api/v1/compliance/incidents/{fake_id}",
json={"status": "resolved"},
)
assert resp.status_code == 404
class TestTenantIsolation:
async def test_tenant_isolation(self, admin_authed_client, db_session):
"""Compliance incidents are tenant-isolated."""
client, seed = admin_authed_client
# Create incident in tenant A
resp = await client.post(
"/api/v1/compliance/incidents",
json={"title": "Tenant A Incident", "incident_type": "ai"},
)
assert resp.status_code == 201
incident_a_id = resp.json()["id"]
# Create incident directly in tenant B via DB
incident_b = ComplianceIncident(
tenant_id=seed["tenant_b"].id,
incident_type="privacy",
title="Tenant B Incident",
)
db_session.add(incident_b)
await db_session.flush()
await db_session.commit()
# List incidents — should only see tenant A's
resp = await client.get("/api/v1/compliance/incidents")
assert resp.status_code == 200
data = resp.json()
titles = [i["title"] for i in data["items"]]
assert "Tenant A Incident" in titles
assert "Tenant B Incident" not in titles
# ──────────────────────────────────────────────────────────────────────────
# K-RET: Retention Policies Tests
# ──────────────────────────────────────────────────────────────────────────
class TestRetentionPolicies:
async def test_retention_policies_list(self, admin_authed_client, db_session):
"""GET /api/v1/compliance/retention-policies returns all policies."""
client, seed = admin_authed_client
# Create system settings for the tenant
await _create_system_settings(db_session, seed["tenant_a"].id)
await db_session.commit()
resp = await client.get("/api/v1/compliance/retention-policies")
assert resp.status_code == 200, f"Retention policies failed: {resp.text}"
data = resp.json()
assert "items" in data
assert data["total"] == 5 # audit_log, backup, trash, knowledge, agent_memory
keys = [p["key"] for p in data["items"]]
assert "audit_log" in keys
assert "backup" in keys
assert "trash" in keys
assert "knowledge" in keys
assert "agent_memory" in keys
# Check default values
for policy in data["items"]:
assert policy["current_days"] == policy["default_days"]
assert policy["editable"] is True
async def test_retention_policy_update(self, admin_authed_client, db_session):
"""PATCH /api/v1/compliance/retention-policies/{key} updates days."""
client, seed = admin_authed_client
# Create system settings
await _create_system_settings(db_session, seed["tenant_a"].id)
await db_session.commit()
resp = await client.patch(
"/api/v1/compliance/retention-policies/audit_log",
json={"days": 180},
)
assert resp.status_code == 200, f"Retention update failed: {resp.text}"
data = resp.json()
assert data["key"] == "audit_log"
assert data["days"] == 180
# Verify the update is reflected
resp = await client.get("/api/v1/compliance/retention-policies")
assert resp.status_code == 200
policies = resp.json()["items"]
audit_policy = next(p for p in policies if p["key"] == "audit_log")
assert audit_policy["current_days"] == 180
async def test_retention_policy_invalid_key(self, admin_authed_client, db_session):
"""PATCH with invalid key returns 400."""
client, seed = admin_authed_client
await _create_system_settings(db_session, seed["tenant_a"].id)
await db_session.commit()
resp = await client.patch(
"/api/v1/compliance/retention-policies/invalid_key",
json={"days": 30},
)
assert resp.status_code == 400