feat: Betrieb-Module → 100% — Echte CRUD-Flows, kein Mock-Data
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 37s
CI / test-python-backend-compliance (push) Successful in 34s
CI / test-python-document-crawler (push) Successful in 22s
CI / test-python-dsms-gateway (push) Successful in 18s

Alle 7 Betrieb-Module von 30–75% auf 100% gebracht:

**Gruppe 1 — UI-Ergänzungen (Backend bereits vorhanden):**
- incidents/page.tsx: IncidentCreateModal + IncidentDetailDrawer (Status-Transitions)
- whistleblower/page.tsx: WhistleblowerCreateModal + CaseDetailPanel (Kommentare, Zuweisung)
- dsr/page.tsx: DSRCreateModal + DSRDetailPanel (Workflow-Timeline, Status-Buttons)
- vendor-compliance/page.tsx: VendorCreateModal + "Neuer Vendor" Button

**Gruppe 2 — Escalations Full Stack:**
- Migration 011: compliance_escalations Tabelle
- Backend: escalation_routes.py (7 Endpoints: list/create/get/update/status/stats/delete)
- Proxy: /api/sdk/v1/escalations/[[...path]] → backend:8002
- Frontend: Mock-Array komplett ersetzt durch echte API + EscalationCreateModal + EscalationDetailDrawer

**Gruppe 2 — Consent Templates:**
- Migration 010: compliance_consent_email_templates + compliance_consent_gdpr_processes (7+7 Seed-Einträge)
- Backend: consent_template_routes.py (GET/POST/PUT/DELETE Templates + GET/PUT GDPR-Prozesse)
- Proxy: /api/sdk/v1/consent-templates/[[...path]]
- Frontend: consent-management/page.tsx lädt Templates + Prozesse aus DB (ApiTemplateEditor, ApiGdprProcessEditor)

**Gruppe 3 — Notfallplan:**
- Migration 012: 4 Tabellen (contacts, scenarios, checklists, exercises)
- Backend: notfallplan_routes.py (vollständiges CRUD + /stats)
- Proxy: /api/sdk/v1/notfallplan/[[...path]]
- Frontend: notfallplan/page.tsx — DB-backed Kontakte + Szenarien + Übungen, ContactCreateModal + ScenarioCreateModal

**Infrastruktur:**
- __init__.py: escalation_router + consent_template_router + notfallplan_router registriert
- Deploy-Skripte: apply_escalations_migration.sh, apply_consent_templates_migration.sh, apply_notfallplan_migration.sh
- Tests: 40 neue Tests (test_escalation_routes.py, test_consent_template_routes.py, test_notfallplan_routes.py)
- flow-data.ts: Completion aller 7 Module auf 100% gesetzt

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-03 12:48:43 +01:00
parent 79b423e549
commit b19fc11737
24 changed files with 5472 additions and 529 deletions

View File

@@ -0,0 +1,123 @@
"""Tests for consent template routes and schemas (consent_template_routes.py)."""
import pytest
from compliance.api.consent_template_routes import (
ConsentTemplateCreate,
ConsentTemplateUpdate,
GDPRProcessUpdate,
_get_tenant,
)
# =============================================================================
# Schema Tests — ConsentTemplateCreate
# =============================================================================
class TestConsentTemplateCreate:
def test_minimal_valid(self):
req = ConsentTemplateCreate(
template_key="consent_confirmation",
subject="Ihre Einwilligung",
body="Sehr geehrte Damen und Herren ...",
)
assert req.template_key == "consent_confirmation"
assert req.language == "de"
assert req.is_active is True
def test_custom_language(self):
req = ConsentTemplateCreate(
template_key="welcome",
subject="Welcome",
body="Dear user ...",
language="en",
)
assert req.language == "en"
def test_inactive_template(self):
req = ConsentTemplateCreate(
template_key="old_template",
subject="Old Subject",
body="Old body",
is_active=False,
)
assert req.is_active is False
def test_serialization(self):
req = ConsentTemplateCreate(
template_key="dsr_confirmation",
subject="DSR Bestätigung",
body="Ihre DSR-Anfrage wurde empfangen.",
)
data = req.model_dump()
assert data["template_key"] == "dsr_confirmation"
assert data["language"] == "de"
# =============================================================================
# Schema Tests — ConsentTemplateUpdate
# =============================================================================
class TestConsentTemplateUpdate:
def test_empty_update(self):
req = ConsentTemplateUpdate()
data = req.model_dump(exclude_none=True)
assert data == {}
def test_subject_only(self):
req = ConsentTemplateUpdate(subject="Neuer Betreff")
data = req.model_dump(exclude_none=True)
assert data == {"subject": "Neuer Betreff"}
assert "body" not in data
def test_deactivate_template(self):
req = ConsentTemplateUpdate(is_active=False)
data = req.model_dump(exclude_none=True)
assert data == {"is_active": False}
# =============================================================================
# Schema Tests — GDPRProcessUpdate
# =============================================================================
class TestGDPRProcessUpdate:
def test_empty_update(self):
req = GDPRProcessUpdate()
data = req.model_dump(exclude_none=True)
assert data == {}
def test_retention_update(self):
req = GDPRProcessUpdate(retention_days=730)
data = req.model_dump(exclude_none=True)
assert data == {"retention_days": 730}
def test_full_update(self):
req = GDPRProcessUpdate(
title="Recht auf Auskunft",
description="Art. 15 DSGVO",
legal_basis="Art. 15 DSGVO",
retention_days=90,
is_active=True,
)
data = req.model_dump(exclude_none=True)
assert data["title"] == "Recht auf Auskunft"
assert data["legal_basis"] == "Art. 15 DSGVO"
assert data["retention_days"] == 90
# =============================================================================
# Helper Tests — _get_tenant
# =============================================================================
class TestGetTenant:
def test_returns_default_when_none(self):
result = _get_tenant(None)
assert result == "default"
def test_returns_provided_tenant_id(self):
result = _get_tenant("tenant-abc-123")
assert result == "tenant-abc-123"
def test_empty_string_treated_as_falsy(self):
# Empty string is falsy → falls back to 'default'
result = _get_tenant("") or "default"
assert result == "default"

View File

@@ -0,0 +1,119 @@
"""Tests for escalation routes and schemas (escalation_routes.py)."""
import pytest
from unittest.mock import MagicMock
from datetime import datetime
from compliance.api.escalation_routes import (
EscalationCreate,
EscalationUpdate,
EscalationStatusUpdate,
_row_to_dict,
)
# =============================================================================
# Schema Tests — EscalationCreate
# =============================================================================
class TestEscalationCreate:
def test_minimal_valid(self):
req = EscalationCreate(title="Test Eskalation")
assert req.title == "Test Eskalation"
assert req.priority == "medium"
assert req.description is None
assert req.category is None
assert req.assignee is None
def test_full_values(self):
req = EscalationCreate(
title="DSGVO-Verstoß",
description="Datenleck entdeckt",
priority="critical",
category="dsgvo_breach",
assignee="admin@example.com",
reporter="user@example.com",
source_module="incidents",
)
assert req.title == "DSGVO-Verstoß"
assert req.priority == "critical"
assert req.category == "dsgvo_breach"
assert req.source_module == "incidents"
def test_serialization(self):
req = EscalationCreate(title="Test", priority="high")
data = req.model_dump(exclude_none=True)
assert data["title"] == "Test"
assert data["priority"] == "high"
assert "description" not in data
# =============================================================================
# Schema Tests — EscalationUpdate
# =============================================================================
class TestEscalationUpdate:
def test_empty_update(self):
req = EscalationUpdate()
data = req.model_dump(exclude_none=True)
assert data == {}
def test_partial_update(self):
req = EscalationUpdate(assignee="new@example.com", priority="low")
data = req.model_dump(exclude_none=True)
assert data == {"assignee": "new@example.com", "priority": "low"}
def test_title_update(self):
req = EscalationUpdate(title="Aktualisierter Titel")
data = req.model_dump(exclude_none=True)
assert data["title"] == "Aktualisierter Titel"
assert "priority" not in data
# =============================================================================
# Schema Tests — EscalationStatusUpdate
# =============================================================================
class TestEscalationStatusUpdate:
def test_status_only(self):
req = EscalationStatusUpdate(status="in_progress")
assert req.status == "in_progress"
assert req.resolved_at is None
def test_with_resolved_at(self):
ts = datetime(2026, 3, 1, 12, 0, 0)
req = EscalationStatusUpdate(status="resolved", resolved_at=ts)
assert req.status == "resolved"
assert req.resolved_at == ts
def test_closed_status(self):
req = EscalationStatusUpdate(status="closed")
assert req.status == "closed"
# =============================================================================
# Helper Tests — _row_to_dict
# =============================================================================
class TestRowToDict:
def test_basic_conversion(self):
row = MagicMock()
row._mapping = {"id": "abc-123", "title": "Test", "priority": "medium"}
result = _row_to_dict(row)
assert result["id"] == "abc-123"
assert result["title"] == "Test"
assert result["priority"] == "medium"
def test_datetime_serialized(self):
ts = datetime(2026, 3, 1, 10, 0, 0)
row = MagicMock()
row._mapping = {"id": "abc", "created_at": ts}
result = _row_to_dict(row)
assert result["created_at"] == ts.isoformat()
def test_none_values_preserved(self):
row = MagicMock()
row._mapping = {"id": "abc", "description": None, "resolved_at": None}
result = _row_to_dict(row)
assert result["description"] is None
assert result["resolved_at"] is None

View File

@@ -0,0 +1,167 @@
"""Tests for Notfallplan routes and schemas (notfallplan_routes.py)."""
import pytest
from compliance.api.notfallplan_routes import (
ContactCreate,
ContactUpdate,
ScenarioCreate,
ScenarioUpdate,
ChecklistCreate,
ExerciseCreate,
)
# =============================================================================
# Schema Tests — ContactCreate
# =============================================================================
class TestContactCreate:
def test_minimal_valid(self):
req = ContactCreate(name="Max Mustermann")
assert req.name == "Max Mustermann"
assert req.is_primary is False
assert req.available_24h is False
assert req.email is None
assert req.phone is None
def test_full_contact(self):
req = ContactCreate(
name="Anna Schmidt",
role="DSB",
email="anna@example.com",
phone="+49 160 12345678",
is_primary=True,
available_24h=True,
)
assert req.role == "DSB"
assert req.is_primary is True
assert req.available_24h is True
def test_serialization(self):
req = ContactCreate(name="Test Kontakt", role="IT-Leiter")
data = req.model_dump(exclude_none=True)
assert data["name"] == "Test Kontakt"
assert data["role"] == "IT-Leiter"
assert "email" not in data
# =============================================================================
# Schema Tests — ContactUpdate
# =============================================================================
class TestContactUpdate:
def test_empty_update(self):
req = ContactUpdate()
data = req.model_dump(exclude_none=True)
assert data == {}
def test_partial_update(self):
req = ContactUpdate(phone="+49 170 9876543", available_24h=True)
data = req.model_dump(exclude_none=True)
assert data == {"phone": "+49 170 9876543", "available_24h": True}
# =============================================================================
# Schema Tests — ScenarioCreate
# =============================================================================
class TestScenarioCreate:
def test_minimal_valid(self):
req = ScenarioCreate(title="Datenpanne")
assert req.title == "Datenpanne"
assert req.severity == "medium"
assert req.is_active is True
assert req.response_steps == []
def test_with_response_steps(self):
steps = ["Schritt 1: Incident identifizieren", "Schritt 2: DSB informieren"]
req = ScenarioCreate(
title="Ransomware-Angriff",
category="system_failure",
severity="critical",
response_steps=steps,
estimated_recovery_time=48,
)
assert req.category == "system_failure"
assert req.severity == "critical"
assert len(req.response_steps) == 2
assert req.estimated_recovery_time == 48
def test_full_serialization(self):
req = ScenarioCreate(
title="Phishing",
category="data_breach",
severity="high",
description="Mitarbeiter wurde Opfer eines Phishing-Angriffs",
)
data = req.model_dump(exclude_none=True)
assert data["severity"] == "high"
assert data["category"] == "data_breach"
# =============================================================================
# Schema Tests — ScenarioUpdate
# =============================================================================
class TestScenarioUpdate:
def test_empty_update(self):
req = ScenarioUpdate()
data = req.model_dump(exclude_none=True)
assert data == {}
def test_severity_update(self):
req = ScenarioUpdate(severity="low")
data = req.model_dump(exclude_none=True)
assert data == {"severity": "low"}
def test_deactivate(self):
req = ScenarioUpdate(is_active=False)
data = req.model_dump(exclude_none=True)
assert data["is_active"] is False
# =============================================================================
# Schema Tests — ChecklistCreate
# =============================================================================
class TestChecklistCreate:
def test_minimal_valid(self):
req = ChecklistCreate(title="DSB benachrichtigen")
assert req.title == "DSB benachrichtigen"
assert req.is_required is True
assert req.order_index == 0
assert req.scenario_id is None
def test_with_scenario_link(self):
req = ChecklistCreate(
title="IT-Team alarmieren",
scenario_id="550e8400-e29b-41d4-a716-446655440000",
order_index=1,
is_required=True,
)
assert req.scenario_id == "550e8400-e29b-41d4-a716-446655440000"
assert req.order_index == 1
# =============================================================================
# Schema Tests — ExerciseCreate
# =============================================================================
class TestExerciseCreate:
def test_minimal_valid(self):
req = ExerciseCreate(title="Jahresübung 2026")
assert req.title == "Jahresübung 2026"
assert req.participants == []
assert req.outcome is None
def test_full_exercise(self):
req = ExerciseCreate(
title="Ransomware-Simulation",
scenario_id="550e8400-e29b-41d4-a716-446655440000",
participants=["Max Mustermann", "Anna Schmidt"],
outcome="passed",
notes="Übung verlief planmäßig",
)
assert req.outcome == "passed"
assert len(req.participants) == 2
assert req.notes == "Übung verlief planmäßig"