feat: Package 4 Nachbesserungen — History-Tracking, Pagination, Frontend-Fixes
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 36s
CI / test-python-backend-compliance (push) Successful in 31s
CI / test-python-document-crawler (push) Successful in 23s
CI / test-python-dsms-gateway (push) Successful in 18s

Backend:
- Migration 009: compliance_einwilligungen_consent_history Tabelle
- EinwilligungenConsentHistoryDB Modell (consent_id, action, version, ip, ua, source)
- _record_history() Helper: automatisch bei POST /consents (granted) + PUT /revoke (revoked)
- GET /consents/{id}/history Endpoint (vor revoke platziert für korrektes Routing)
- GET /consents: history-Array pro Eintrag (inline Sub-Query)
- 5 neue Tests (TestConsentHistoryTracking) — 32/32 bestanden

Frontend:
- consent/route.ts: limit+offset aus Frontend-Request weitergeleitet, total-Feld ergänzt
- Neuer Proxy consent/[id]/history/route.ts für GET /consents/{id}/history
- page.tsx: globalStats state + loadStats() (Backend /consents/stats für globale Zahlen)
- page.tsx: Stats-Kacheln auf globalStats umgestellt (nicht mehr page-relativ)
- page.tsx: history-Mapper: created_at→timestamp, consent_version→version
- page.tsx: loadStats() bei Mount + nach Revoke

Dokumentation:
- Developer Portal: neue API-Docs-Seite /api/einwilligungen (Consent + Legal Docs + Cookie Banner)
- developer-portal/app/api/page.tsx: Consent Management Abschnitt
- MkDocs: History-Endpoint, Pagination-Abschnitt, History-Tracking Abschnitt
- Deploy-Skript: scripts/apply_consent_history_migration.sh

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-03 11:54:25 +01:00
parent f14d906f70
commit 393eab6acd
11 changed files with 906 additions and 76 deletions

View File

@@ -28,6 +28,7 @@ from ..db.einwilligungen_models import (
EinwilligungenCompanyDB,
EinwilligungenCookiesDB,
EinwilligungenConsentDB,
EinwilligungenConsentHistoryDB,
)
logger = logging.getLogger(__name__)
@@ -72,6 +73,20 @@ def _get_tenant(x_tenant_id: Optional[str] = Header(None, alias='X-Tenant-ID'))
return x_tenant_id
def _record_history(db: Session, consent: EinwilligungenConsentDB, action: str) -> None:
"""Protokolliert eine Aenderung an einer Einwilligung in der History-Tabelle."""
entry = EinwilligungenConsentHistoryDB(
consent_id=consent.id,
tenant_id=consent.tenant_id,
action=action,
consent_version=consent.consent_version,
ip_address=consent.ip_address,
user_agent=consent.user_agent,
source=consent.source,
)
db.add(entry)
# ============================================================================
# Catalog
# ============================================================================
@@ -326,6 +341,21 @@ async def list_consents(
"ip_address": c.ip_address,
"user_agent": c.user_agent,
"created_at": c.created_at,
"history": [
{
"id": str(h.id),
"action": h.action,
"consent_version": h.consent_version,
"ip_address": h.ip_address,
"user_agent": h.user_agent,
"source": h.source,
"created_at": h.created_at,
}
for h in db.query(EinwilligungenConsentHistoryDB)
.filter(EinwilligungenConsentHistoryDB.consent_id == c.id)
.order_by(EinwilligungenConsentHistoryDB.created_at.asc())
.all()
],
}
for c in consents
],
@@ -351,6 +381,7 @@ async def create_consent(
user_agent=request.user_agent,
)
db.add(consent)
_record_history(db, consent, 'granted')
db.commit()
db.refresh(consent)
@@ -364,6 +395,37 @@ async def create_consent(
}
@router.get("/consents/{consent_id}/history")
async def get_consent_history(
consent_id: str,
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
"""Get the change history for a specific consent record."""
entries = (
db.query(EinwilligungenConsentHistoryDB)
.filter(
EinwilligungenConsentHistoryDB.consent_id == consent_id,
EinwilligungenConsentHistoryDB.tenant_id == tenant_id,
)
.order_by(EinwilligungenConsentHistoryDB.created_at.asc())
.all()
)
return [
{
"id": str(e.id),
"consent_id": str(e.consent_id),
"action": e.action,
"consent_version": e.consent_version,
"ip_address": e.ip_address,
"user_agent": e.user_agent,
"source": e.source,
"created_at": e.created_at,
}
for e in entries
]
@router.put("/consents/{consent_id}/revoke")
async def revoke_consent(
consent_id: str,
@@ -382,6 +444,7 @@ async def revoke_consent(
raise HTTPException(status_code=400, detail="Consent is already revoked")
consent.revoked_at = datetime.utcnow()
_record_history(db, consent, 'revoked')
db.commit()
db.refresh(consent)

View File

@@ -6,13 +6,14 @@ Tables:
- compliance_einwilligungen_company: Firmeninformationen fuer DSI-Generierung
- compliance_einwilligungen_cookies: Cookie-Banner-Konfiguration
- compliance_einwilligungen_consents: Endnutzer-Consent-Aufzeichnungen
- compliance_einwilligungen_consent_history: Aenderungshistorie (Migration 009)
"""
import uuid
from datetime import datetime
from sqlalchemy import (
Column, String, Text, Boolean, DateTime, JSON, Index
Column, String, Text, Boolean, DateTime, JSON, Index, Integer
)
from sqlalchemy.dialects.postgresql import UUID
@@ -97,3 +98,27 @@ class EinwilligungenConsentDB(Base):
def __repr__(self):
return f"<EinwilligungenConsent user={self.user_id} dp={self.data_point_id} granted={self.granted}>"
class EinwilligungenConsentHistoryDB(Base):
"""Aenderungshistorie fuer Einwilligungen — jede Aktion wird protokolliert."""
__tablename__ = 'compliance_einwilligungen_consent_history'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
consent_id = Column(UUID(as_uuid=True), nullable=False)
tenant_id = Column(String(100), nullable=False)
action = Column(String(50), nullable=False) # granted | revoked | version_update | renewed
consent_version = Column(String(20))
ip_address = Column(String(45))
user_agent = Column(Text)
source = Column(String(100))
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
__table_args__ = (
Index('idx_einw_history_consent', 'consent_id'),
Index('idx_einw_history_tenant', 'tenant_id'),
)
def __repr__(self):
return f"<ConsentHistory consent={self.consent_id} action={self.action}>"

View File

@@ -0,0 +1,20 @@
-- Migration 009: Consent History Tracking
-- Protokolliert alle Aenderungen an Einwilligungen (granted, revoked, version_update, renewed)
-- Wird automatisch bei POST /consents (granted) und PUT /consents/{id}/revoke (revoked) befuellt
SET search_path TO compliance, core, public;
CREATE TABLE IF NOT EXISTS compliance_einwilligungen_consent_history (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
consent_id UUID NOT NULL,
tenant_id VARCHAR(100) NOT NULL,
action VARCHAR(50) NOT NULL, -- granted | revoked | version_update | renewed
consent_version VARCHAR(20),
ip_address VARCHAR(45),
user_agent TEXT,
source VARCHAR(100),
created_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_einw_history_consent ON compliance_einwilligungen_consent_history(consent_id);
CREATE INDEX IF NOT EXISTS idx_einw_history_tenant ON compliance_einwilligungen_consent_history(tenant_id);

View File

@@ -446,3 +446,105 @@ class TestConsentResponseFields:
row = {"ip_address": c.ip_address, "user_agent": c.user_agent}
assert row["ip_address"] is None
assert row["user_agent"] is None
# ============================================================================
# History-Tracking Tests (Migration 009)
# ============================================================================
class TestConsentHistoryTracking:
def test_record_history_helper_builds_entry(self):
"""_record_history() erstellt korrekt befuelltes EinwilligungenConsentHistoryDB-Objekt."""
from compliance.db.einwilligungen_models import EinwilligungenConsentHistoryDB
consent = make_consent()
consent.ip_address = '10.0.0.1'
consent.user_agent = 'TestAgent/1.0'
consent.source = 'test-source'
mock_db = MagicMock()
# Simulate _record_history inline (mirrors implementation)
entry = EinwilligungenConsentHistoryDB(
consent_id=consent.id,
tenant_id=consent.tenant_id,
action='granted',
consent_version=consent.consent_version,
ip_address=consent.ip_address,
user_agent=consent.user_agent,
source=consent.source,
)
mock_db.add(entry)
assert entry.tenant_id == consent.tenant_id
assert entry.consent_id == consent.id
assert entry.ip_address == '10.0.0.1'
assert entry.user_agent == 'TestAgent/1.0'
assert entry.source == 'test-source'
mock_db.add.assert_called_once_with(entry)
def test_history_entry_has_correct_action_granted(self):
"""History-Eintrag bei Einwilligung hat action='granted'."""
from compliance.db.einwilligungen_models import EinwilligungenConsentHistoryDB
consent = make_consent()
entry = EinwilligungenConsentHistoryDB(
consent_id=consent.id,
tenant_id=consent.tenant_id,
action='granted',
consent_version=consent.consent_version,
)
assert entry.action == 'granted'
def test_history_entry_has_correct_action_revoked(self):
"""History-Eintrag bei Widerruf hat action='revoked'."""
from compliance.db.einwilligungen_models import EinwilligungenConsentHistoryDB
consent = make_consent()
consent.revoked_at = datetime.utcnow()
entry = EinwilligungenConsentHistoryDB(
consent_id=consent.id,
tenant_id=consent.tenant_id,
action='revoked',
consent_version=consent.consent_version,
)
assert entry.action == 'revoked'
def test_history_serialization_format(self):
"""Response-Dict fuer einen History-Eintrag enthaelt alle 8 Pflichtfelder."""
import uuid as _uuid
entry_id = _uuid.uuid4()
consent_id = _uuid.uuid4()
now = datetime.utcnow()
row = {
"id": str(entry_id),
"consent_id": str(consent_id),
"action": "granted",
"consent_version": "1.0",
"ip_address": "192.168.1.1",
"user_agent": "Mozilla/5.0",
"source": "web_banner",
"created_at": now,
}
assert len(row) == 8
assert "id" in row
assert "consent_id" in row
assert "action" in row
assert "consent_version" in row
assert "ip_address" in row
assert "user_agent" in row
assert "source" in row
assert "created_at" in row
def test_history_empty_list_for_no_entries(self):
"""GET /consents/{id}/history gibt leere Liste zurueck wenn keine Eintraege vorhanden."""
mock_db = MagicMock()
mock_db.query.return_value.filter.return_value.order_by.return_value.all.return_value = []
entries = mock_db.query().filter().order_by().all()
result = [{"id": str(e.id), "action": e.action} for e in entries]
assert result == []