feat: Package 4 Rechtliche Texte — DB-Persistenz fuer Legal Documents, Einwilligungen und Cookie Banner
All checks were successful
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / test-go-ai-compliance (push) Successful in 46s
CI / test-python-backend-compliance (push) Successful in 32s
CI / test-python-document-crawler (push) Successful in 22s
CI / test-python-dsms-gateway (push) Successful in 17s

- Migration 007: compliance_legal_documents, _versions, _approvals (Approval-Workflow)
- Migration 008: compliance_einwilligungen_catalog, _company, _cookies, _consents
- Backend: legal_document_routes.py (11 Endpoints + draft→review→approved→published Workflow)
- Backend: einwilligungen_routes.py (10 Endpoints inkl. Stats, Pagination, Revoke)
- Frontend: /api/admin/consent/[[...path]] Catch-All-Proxy fuer Legal Documents
- Frontend: catalog/consent/cookie-banner routes von In-Memory auf DB-Proxy umgestellt
- Frontend: einwilligungen/page.tsx + cookie-banner/page.tsx laden jetzt via API (kein Mock)
- Tests: 44/44 pass (test_legal_document_routes.py + test_einwilligungen_routes.py)
- Deploy-Scripts: apply_legal_docs_migration.sh + apply_einwilligungen_migration.sh

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-03 08:25:13 +01:00
parent 799668e472
commit 113ecdfa77
17 changed files with 2501 additions and 664 deletions

View File

@@ -0,0 +1,389 @@
"""
Tests for Einwilligungen Routes — 008_einwilligungen migration.
Tests: Catalog Upsert, Company Info, Cookie-Config, Consent erfassen,
Consent widerrufen, Statistiken.
"""
import pytest
from unittest.mock import MagicMock, patch
from datetime import datetime
import uuid
# ============================================================================
# Shared Fixtures
# ============================================================================
def make_uuid():
return str(uuid.uuid4())
def make_catalog(tenant_id='test-tenant'):
rec = MagicMock()
rec.id = uuid.uuid4()
rec.tenant_id = tenant_id
rec.selected_data_point_ids = ['dp-001', 'dp-002']
rec.custom_data_points = []
rec.updated_at = datetime.utcnow()
return rec
def make_company(tenant_id='test-tenant'):
rec = MagicMock()
rec.id = uuid.uuid4()
rec.tenant_id = tenant_id
rec.data = {'company_name': 'Test GmbH', 'email': 'datenschutz@test.de'}
rec.updated_at = datetime.utcnow()
return rec
def make_cookies(tenant_id='test-tenant'):
rec = MagicMock()
rec.id = uuid.uuid4()
rec.tenant_id = tenant_id
rec.categories = [
{'id': 'necessary', 'name': 'Notwendig', 'isRequired': True, 'defaultEnabled': True},
{'id': 'analytics', 'name': 'Analyse', 'isRequired': False, 'defaultEnabled': False},
]
rec.config = {'position': 'bottom', 'style': 'bar'}
rec.updated_at = datetime.utcnow()
return rec
def make_consent(tenant_id='test-tenant', user_id='user-001', data_point_id='dp-001', granted=True):
rec = MagicMock()
rec.id = uuid.uuid4()
rec.tenant_id = tenant_id
rec.user_id = user_id
rec.data_point_id = data_point_id
rec.granted = granted
rec.granted_at = datetime.utcnow()
rec.revoked_at = None
rec.consent_version = '1.0'
rec.source = 'website'
rec.ip_address = None
rec.user_agent = None
rec.created_at = datetime.utcnow()
return rec
# ============================================================================
# Pydantic Schema Tests
# ============================================================================
class TestCatalogUpsert:
def test_catalog_upsert_defaults(self):
from compliance.api.einwilligungen_routes import CatalogUpsert
data = CatalogUpsert()
assert data.selected_data_point_ids == []
assert data.custom_data_points == []
def test_catalog_upsert_with_data(self):
from compliance.api.einwilligungen_routes import CatalogUpsert
data = CatalogUpsert(
selected_data_point_ids=['dp-001', 'dp-002', 'dp-003'],
custom_data_points=[{'id': 'custom-1', 'name': 'Eigener Datenpunkt'}],
)
assert len(data.selected_data_point_ids) == 3
assert len(data.custom_data_points) == 1
class TestCompanyUpsert:
def test_company_upsert_empty(self):
from compliance.api.einwilligungen_routes import CompanyUpsert
data = CompanyUpsert()
assert data.data == {}
def test_company_upsert_with_data(self):
from compliance.api.einwilligungen_routes import CompanyUpsert
data = CompanyUpsert(data={
'company_name': 'Test GmbH',
'address': 'Musterstraße 1, 12345 Berlin',
'email': 'datenschutz@test.de',
'dpo_name': 'Max Mustermann',
})
assert data.data['company_name'] == 'Test GmbH'
assert data.data['dpo_name'] == 'Max Mustermann'
class TestCookiesUpsert:
def test_cookies_upsert_defaults(self):
from compliance.api.einwilligungen_routes import CookiesUpsert
data = CookiesUpsert()
assert data.categories == []
assert data.config == {}
def test_cookies_upsert_with_categories(self):
from compliance.api.einwilligungen_routes import CookiesUpsert
data = CookiesUpsert(
categories=[
{'id': 'necessary', 'name': 'Notwendig', 'isRequired': True},
{'id': 'analytics', 'name': 'Analyse', 'isRequired': False},
],
config={'position': 'bottom'},
)
assert len(data.categories) == 2
assert data.config['position'] == 'bottom'
class TestConsentCreate:
def test_consent_create_valid(self):
from compliance.api.einwilligungen_routes import ConsentCreate
data = ConsentCreate(
user_id='user-001',
data_point_id='dp-marketing',
granted=True,
)
assert data.user_id == 'user-001'
assert data.granted is True
assert data.consent_version == '1.0'
assert data.source is None
def test_consent_create_revoke(self):
from compliance.api.einwilligungen_routes import ConsentCreate
data = ConsentCreate(
user_id='user-001',
data_point_id='dp-analytics',
granted=False,
consent_version='2.0',
source='cookie-banner',
)
assert data.granted is False
assert data.consent_version == '2.0'
# ============================================================================
# Catalog Upsert Tests
# ============================================================================
class TestCatalogDB:
def test_catalog_returns_empty_when_not_found(self):
"""GET catalog should return empty defaults when no record exists."""
from compliance.db.einwilligungen_models import EinwilligungenCatalogDB
mock_db = MagicMock()
mock_db.query.return_value.filter.return_value.first.return_value = None
result = mock_db.query(EinwilligungenCatalogDB).filter().first()
assert result is None
def test_catalog_upsert_creates_new(self):
"""PUT catalog should create a new record if none exists."""
from compliance.db.einwilligungen_models import EinwilligungenCatalogDB
mock_db = MagicMock()
mock_db.query.return_value.filter.return_value.first.return_value = None
new_record = EinwilligungenCatalogDB(
tenant_id='test-tenant',
selected_data_point_ids=['dp-001'],
custom_data_points=[],
)
assert new_record.tenant_id == 'test-tenant'
assert new_record.selected_data_point_ids == ['dp-001']
def test_catalog_upsert_updates_existing(self):
"""PUT catalog should update existing record."""
existing = make_catalog()
existing.selected_data_point_ids = ['dp-001', 'dp-002', 'dp-003']
existing.custom_data_points = [{'id': 'custom-1'}]
assert len(existing.selected_data_point_ids) == 3
assert len(existing.custom_data_points) == 1
# ============================================================================
# Cookie Config Tests
# ============================================================================
class TestCookieConfig:
def test_cookie_config_returns_empty_when_not_found(self):
"""GET cookies should return empty defaults for new tenant."""
mock_db = MagicMock()
mock_db.query.return_value.filter.return_value.first.return_value = None
result = mock_db.query().filter().first()
assert result is None
def test_cookie_config_upsert_with_categories(self):
"""PUT cookies should store categories and config."""
from compliance.db.einwilligungen_models import EinwilligungenCookiesDB
categories = [
{'id': 'necessary', 'name': 'Notwendig', 'isRequired': True, 'defaultEnabled': True},
]
config = {'position': 'bottom', 'primaryColor': '#6366f1'}
rec = EinwilligungenCookiesDB(
tenant_id='test-tenant',
categories=categories,
config=config,
)
assert rec.categories[0]['id'] == 'necessary'
assert rec.config['position'] == 'bottom'
def test_essential_cookies_cannot_be_disabled(self):
"""Category with isRequired=True should not allow enabled=False."""
categories = [
{'id': 'necessary', 'name': 'Notwendig', 'isRequired': True, 'defaultEnabled': True},
{'id': 'analytics', 'name': 'Analyse', 'isRequired': False, 'defaultEnabled': True},
]
# Simulate the toggle logic
category_id = 'necessary'
enabled = False
updated = []
for cat in categories:
if cat['id'] == category_id:
if cat.get('isRequired') and not enabled:
updated.append(cat) # Not changed
else:
updated.append({**cat, 'defaultEnabled': enabled})
else:
updated.append(cat)
necessary_cat = next(c for c in updated if c['id'] == 'necessary')
assert necessary_cat['defaultEnabled'] is True # Not changed
# ============================================================================
# Consent Tests
# ============================================================================
class TestConsentDB:
def test_consent_record_creation(self):
"""Consent record should store all required fields."""
from compliance.db.einwilligungen_models import EinwilligungenConsentDB
consent = EinwilligungenConsentDB(
tenant_id='test-tenant',
user_id='user-001',
data_point_id='dp-marketing',
granted=True,
granted_at=datetime.utcnow(),
consent_version='1.0',
source='website',
)
assert consent.tenant_id == 'test-tenant'
assert consent.granted is True
assert consent.revoked_at is None
def test_consent_revoke_sets_revoked_at(self):
"""Revoking a consent should set revoked_at timestamp."""
consent = make_consent()
assert consent.revoked_at is None
consent.revoked_at = datetime.utcnow()
assert consent.revoked_at is not None
def test_cannot_revoke_already_revoked(self):
"""Should not be possible to revoke an already revoked consent."""
consent = make_consent()
consent.revoked_at = datetime.utcnow()
# Simulate the guard logic from the route
already_revoked = consent.revoked_at is not None
assert already_revoked is True # Route would raise HTTPException 400
# ============================================================================
# Statistics Tests
# ============================================================================
class TestConsentStats:
def test_stats_empty_tenant(self):
"""Stats for tenant with no consents should return zeros."""
consents = []
total = len(consents)
active = sum(1 for c in consents if c.granted and not c.revoked_at)
revoked = sum(1 for c in consents if c.revoked_at)
unique_users = len(set(c.user_id for c in consents))
assert total == 0
assert active == 0
assert revoked == 0
assert unique_users == 0
def test_stats_with_mixed_consents(self):
"""Stats should correctly count active and revoked consents."""
consents = [
make_consent(user_id='user-1', data_point_id='dp-1', granted=True),
make_consent(user_id='user-1', data_point_id='dp-2', granted=True),
make_consent(user_id='user-2', data_point_id='dp-1', granted=True),
]
# Revoke one
consents[1].revoked_at = datetime.utcnow()
total = len(consents)
active = sum(1 for c in consents if c.granted and not c.revoked_at)
revoked = sum(1 for c in consents if c.revoked_at)
unique_users = len(set(c.user_id for c in consents))
assert total == 3
assert active == 2
assert revoked == 1
assert unique_users == 2
def test_stats_conversion_rate(self):
"""Conversion rate = users with active consent / total unique users."""
consents = [
make_consent(user_id='user-1', granted=True),
make_consent(user_id='user-2', granted=True),
make_consent(user_id='user-3', granted=True),
]
consents[2].revoked_at = datetime.utcnow() # user-3 revoked
unique_users = len(set(c.user_id for c in consents))
users_with_active = len(set(c.user_id for c in consents if c.granted and not c.revoked_at))
rate = round((users_with_active / unique_users * 100), 1) if unique_users > 0 else 0.0
assert unique_users == 3
assert users_with_active == 2
assert rate == pytest.approx(66.7, 0.1)
def test_stats_by_data_point(self):
"""Stats should group consents by data_point_id."""
consents = [
make_consent(data_point_id='dp-marketing', granted=True),
make_consent(data_point_id='dp-marketing', granted=True),
make_consent(data_point_id='dp-analytics', granted=True),
]
by_dp: dict = {}
for c in consents:
dp = c.data_point_id
if dp not in by_dp:
by_dp[dp] = {'total': 0, 'active': 0}
by_dp[dp]['total'] += 1
if c.granted and not c.revoked_at:
by_dp[dp]['active'] += 1
assert by_dp['dp-marketing']['total'] == 2
assert by_dp['dp-analytics']['total'] == 1
# ============================================================================
# Model Repr Tests
# ============================================================================
class TestModelReprs:
def test_catalog_repr(self):
from compliance.db.einwilligungen_models import EinwilligungenCatalogDB
rec = EinwilligungenCatalogDB(tenant_id='my-tenant')
assert 'my-tenant' in repr(rec)
def test_cookies_repr(self):
from compliance.db.einwilligungen_models import EinwilligungenCookiesDB
rec = EinwilligungenCookiesDB(tenant_id='my-tenant')
assert 'my-tenant' in repr(rec)
def test_consent_repr(self):
from compliance.db.einwilligungen_models import EinwilligungenConsentDB
rec = EinwilligungenConsentDB(
tenant_id='t1', user_id='u1', data_point_id='dp1', granted=True
)
assert 'u1' in repr(rec)
assert 'dp1' in repr(rec)

View File

@@ -0,0 +1,313 @@
"""
Tests for Legal Document Routes — 007_legal_documents migration.
Tests: Document CRUD, Version creation, Approval-Workflow (submit→approve→publish),
Rejection-Flow, approval history.
"""
import pytest
from unittest.mock import MagicMock, patch
from datetime import datetime
import uuid
# ============================================================================
# Shared Fixtures
# ============================================================================
def make_uuid():
return str(uuid.uuid4())
def make_document(type='privacy_policy', name='Datenschutzerklärung', tenant_id='test-tenant'):
doc = MagicMock()
doc.id = uuid.uuid4()
doc.tenant_id = tenant_id
doc.type = type
doc.name = name
doc.description = 'Test description'
doc.mandatory = False
doc.created_at = datetime.utcnow()
doc.updated_at = None
return doc
def make_version(document_id=None, version='1.0', status='draft', title='Test Version'):
v = MagicMock()
v.id = uuid.uuid4()
v.document_id = document_id or uuid.uuid4()
v.version = version
v.language = 'de'
v.title = title
v.content = '<p>Inhalt der Datenschutzerklärung</p>'
v.summary = 'Kurzzusammenfassung'
v.status = status
v.created_by = 'admin@test.de'
v.approved_by = None
v.approved_at = None
v.rejection_reason = None
v.created_at = datetime.utcnow()
v.updated_at = None
return v
def make_approval(version_id=None, action='created'):
a = MagicMock()
a.id = uuid.uuid4()
a.version_id = version_id or uuid.uuid4()
a.action = action
a.approver = 'admin@test.de'
a.comment = None
a.created_at = datetime.utcnow()
return a
# ============================================================================
# Pydantic Schema Tests
# ============================================================================
class TestDocumentCreate:
def test_document_create_valid(self):
from compliance.api.legal_document_routes import DocumentCreate
doc = DocumentCreate(
type='privacy_policy',
name='Datenschutzerklärung',
description='DSE für Webseite',
mandatory=True,
tenant_id='tenant-abc',
)
assert doc.type == 'privacy_policy'
assert doc.mandatory is True
assert doc.tenant_id == 'tenant-abc'
def test_document_create_minimal(self):
from compliance.api.legal_document_routes import DocumentCreate
doc = DocumentCreate(type='terms', name='AGB')
assert doc.mandatory is False
assert doc.tenant_id is None
assert doc.description is None
def test_document_create_all_types(self):
from compliance.api.legal_document_routes import DocumentCreate
for doc_type in ['privacy_policy', 'terms', 'cookie_policy', 'imprint', 'dpa']:
doc = DocumentCreate(type=doc_type, name=f'{doc_type} document')
assert doc.type == doc_type
class TestVersionCreate:
def test_version_create_valid(self):
from compliance.api.legal_document_routes import VersionCreate
doc_id = make_uuid()
v = VersionCreate(
document_id=doc_id,
version='1.0',
title='DSE Version 1.0',
content='<p>Inhalt</p>',
summary='Zusammenfassung',
created_by='admin@test.de',
)
assert v.version == '1.0'
assert v.language == 'de'
assert v.document_id == doc_id
def test_version_create_defaults(self):
from compliance.api.legal_document_routes import VersionCreate
v = VersionCreate(
document_id=make_uuid(),
version='2.0',
title='Version 2',
content='Content',
)
assert v.language == 'de'
assert v.created_by is None
assert v.summary is None
def test_version_update_partial(self):
from compliance.api.legal_document_routes import VersionUpdate
update = VersionUpdate(title='Neuer Titel', content='Neuer Inhalt')
data = update.dict(exclude_none=True)
assert 'title' in data
assert 'content' in data
assert 'language' not in data
class TestActionRequest:
def test_action_request_defaults(self):
from compliance.api.legal_document_routes import ActionRequest
req = ActionRequest()
assert req.approver is None
assert req.comment is None
def test_action_request_with_data(self):
from compliance.api.legal_document_routes import ActionRequest
req = ActionRequest(approver='dpo@company.de', comment='Alles korrekt')
assert req.approver == 'dpo@company.de'
assert req.comment == 'Alles korrekt'
# ============================================================================
# Helper Function Tests
# ============================================================================
class TestDocToResponse:
def test_doc_to_response(self):
from compliance.api.legal_document_routes import _doc_to_response
doc = make_document()
resp = _doc_to_response(doc)
assert resp.id == str(doc.id)
assert resp.type == 'privacy_policy'
assert resp.mandatory is False
def test_doc_to_response_mandatory(self):
from compliance.api.legal_document_routes import _doc_to_response
doc = make_document()
doc.mandatory = True
resp = _doc_to_response(doc)
assert resp.mandatory is True
class TestVersionToResponse:
def test_version_to_response_draft(self):
from compliance.api.legal_document_routes import _version_to_response
v = make_version(status='draft')
resp = _version_to_response(v)
assert resp.status == 'draft'
assert resp.approved_by is None
assert resp.rejection_reason is None
def test_version_to_response_approved(self):
from compliance.api.legal_document_routes import _version_to_response
v = make_version(status='approved')
v.approved_by = 'dpo@company.de'
v.approved_at = datetime.utcnow()
resp = _version_to_response(v)
assert resp.status == 'approved'
assert resp.approved_by == 'dpo@company.de'
def test_version_to_response_rejected(self):
from compliance.api.legal_document_routes import _version_to_response
v = make_version(status='rejected')
v.rejection_reason = 'Inhalt unvollständig'
resp = _version_to_response(v)
assert resp.status == 'rejected'
assert resp.rejection_reason == 'Inhalt unvollständig'
# ============================================================================
# Approval Workflow Transition Tests
# ============================================================================
class TestApprovalWorkflow:
def test_transition_raises_on_wrong_status(self):
"""_transition should raise HTTPException if version is in wrong status."""
from compliance.api.legal_document_routes import _transition
from fastapi import HTTPException
mock_db = MagicMock()
v = make_version(status='draft')
mock_db.query.return_value.filter.return_value.first.return_value = v
with pytest.raises(HTTPException) as exc_info:
_transition(mock_db, str(v.id), ['review'], 'approved', 'approved', None, None)
assert exc_info.value.status_code == 400
assert 'draft' in exc_info.value.detail
def test_transition_raises_on_not_found(self):
"""_transition should raise 404 if version not found."""
from compliance.api.legal_document_routes import _transition
from fastapi import HTTPException
mock_db = MagicMock()
mock_db.query.return_value.filter.return_value.first.return_value = None
with pytest.raises(HTTPException) as exc_info:
_transition(mock_db, make_uuid(), ['draft'], 'review', 'submitted', None, None)
assert exc_info.value.status_code == 404
def test_transition_success(self):
"""_transition should change status and log approval."""
from compliance.api.legal_document_routes import _transition
mock_db = MagicMock()
v = make_version(status='draft')
mock_db.query.return_value.filter.return_value.first.return_value = v
result = _transition(mock_db, str(v.id), ['draft'], 'review', 'submitted', 'admin', None)
assert v.status == 'review'
mock_db.commit.assert_called_once()
def test_full_workflow_draft_to_published(self):
"""Simulate the full approval workflow: draft → review → approved → published."""
from compliance.api.legal_document_routes import _transition
mock_db = MagicMock()
v = make_version(status='draft')
mock_db.query.return_value.filter.return_value.first.return_value = v
# Step 1: Submit for review
_transition(mock_db, str(v.id), ['draft'], 'review', 'submitted', 'author', None)
assert v.status == 'review'
# Step 2: Approve
mock_db.reset_mock()
_transition(mock_db, str(v.id), ['review'], 'approved', 'approved', 'dpo', 'Korrekt',
extra_updates={'approved_by': 'dpo', 'approved_at': datetime.utcnow()})
assert v.status == 'approved'
# Step 3: Publish
mock_db.reset_mock()
_transition(mock_db, str(v.id), ['approved'], 'published', 'published', 'dpo', None)
assert v.status == 'published'
def test_rejection_flow(self):
"""Review → Rejected → draft (re-edit) → review again."""
from compliance.api.legal_document_routes import _transition
mock_db = MagicMock()
v = make_version(status='review')
mock_db.query.return_value.filter.return_value.first.return_value = v
# Reject
_transition(mock_db, str(v.id), ['review'], 'rejected', 'rejected', 'dpo', 'Überarbeitung nötig',
extra_updates={'rejection_reason': 'Überarbeitung nötig'})
assert v.status == 'rejected'
# After rejection, version is editable again (draft/rejected allowed)
# Re-submit for review
_transition(mock_db, str(v.id), ['draft', 'rejected'], 'review', 'submitted', 'author', None)
assert v.status == 'review'
# ============================================================================
# Log Approval Tests
# ============================================================================
class TestLogApproval:
def test_log_approval_creates_entry(self):
from compliance.api.legal_document_routes import _log_approval
from compliance.db.legal_document_models import LegalDocumentApprovalDB
mock_db = MagicMock()
version_id = uuid.uuid4()
entry = _log_approval(mock_db, version_id, 'approved', 'dpo@test.de', 'Gut')
mock_db.add.assert_called_once()
added = mock_db.add.call_args[0][0]
assert isinstance(added, LegalDocumentApprovalDB)
assert added.action == 'approved'
assert added.approver == 'dpo@test.de'
def test_log_approval_without_approver(self):
from compliance.api.legal_document_routes import _log_approval
from compliance.db.legal_document_models import LegalDocumentApprovalDB
mock_db = MagicMock()
_log_approval(mock_db, uuid.uuid4(), 'created')
added = mock_db.add.call_args[0][0]
assert added.approver is None
assert added.comment is None