a4df3201db
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 35s
CI / test-python-backend-compliance (push) Successful in 38s
CI / test-python-document-crawler (push) Successful in 25s
CI / test-python-dsms-gateway (push) Successful in 21s
- Backend: compliance_obligations Tabelle (Migration 013) - Backend: obligation_routes.py — GET/POST/PUT/DELETE + Stats-Endpoint - Backend: obligation_router in __init__.py registriert - Frontend: obligations/page.tsx — ObligationModal, ObligationDetail, ObligationCard, alle Buttons verdrahtet - Proxy: PATCH-Methode in compliance catch-all route ergänzt - Tests: 39/39 Obligation-Tests (Schemas, Helpers, Business Logic) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
326 lines
12 KiB
Python
326 lines
12 KiB
Python
"""Tests for compliance obligation routes and schemas (obligation_routes.py)."""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
from datetime import datetime
|
|
|
|
from compliance.api.obligation_routes import (
|
|
ObligationCreate,
|
|
ObligationUpdate,
|
|
ObligationStatusUpdate,
|
|
_row_to_dict,
|
|
_get_tenant_id,
|
|
DEFAULT_TENANT_ID,
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Tests — ObligationCreate
|
|
# =============================================================================
|
|
|
|
class TestObligationCreate:
|
|
def test_minimal_valid(self):
|
|
req = ObligationCreate(title="Art. 30 VVT führen")
|
|
assert req.title == "Art. 30 VVT führen"
|
|
assert req.source == "DSGVO"
|
|
assert req.status == "pending"
|
|
assert req.priority == "medium"
|
|
assert req.description is None
|
|
assert req.source_article is None
|
|
assert req.deadline is None
|
|
assert req.responsible is None
|
|
assert req.linked_systems is None
|
|
assert req.assessment_id is None
|
|
assert req.rule_code is None
|
|
assert req.notes is None
|
|
|
|
def test_full_values(self):
|
|
deadline = datetime(2026, 6, 1, 0, 0, 0)
|
|
req = ObligationCreate(
|
|
title="DSFA durchführen",
|
|
description="Pflicht nach Art. 35 DSGVO",
|
|
source="DSGVO",
|
|
source_article="Art. 35",
|
|
deadline=deadline,
|
|
status="in-progress",
|
|
priority="critical",
|
|
responsible="Datenschutzbeauftragter",
|
|
linked_systems=["CRM", "ERP"],
|
|
assessment_id="abc123",
|
|
rule_code="RULE-DSFA-001",
|
|
notes="Frist ist bindend",
|
|
)
|
|
assert req.title == "DSFA durchführen"
|
|
assert req.source_article == "Art. 35"
|
|
assert req.priority == "critical"
|
|
assert req.responsible == "Datenschutzbeauftragter"
|
|
assert req.linked_systems == ["CRM", "ERP"]
|
|
assert req.rule_code == "RULE-DSFA-001"
|
|
|
|
def test_ai_act_source(self):
|
|
req = ObligationCreate(title="Risikoklasse bestimmen", source="AI Act")
|
|
assert req.source == "AI Act"
|
|
assert req.status == "pending"
|
|
|
|
def test_nis2_source(self):
|
|
req = ObligationCreate(title="Meldepflicht einrichten", source="NIS2")
|
|
assert req.source == "NIS2"
|
|
|
|
def test_serialization_excludes_none(self):
|
|
req = ObligationCreate(title="Test", priority="high")
|
|
data = req.model_dump(exclude_none=True)
|
|
assert data["title"] == "Test"
|
|
assert data["priority"] == "high"
|
|
assert "description" not in data
|
|
assert "deadline" not in data
|
|
|
|
def test_serialization_includes_set_fields(self):
|
|
req = ObligationCreate(title="Test", status="overdue", responsible="admin")
|
|
data = req.model_dump()
|
|
assert data["status"] == "overdue"
|
|
assert data["responsible"] == "admin"
|
|
assert data["source"] == "DSGVO"
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Tests — ObligationUpdate
|
|
# =============================================================================
|
|
|
|
class TestObligationUpdate:
|
|
def test_empty_update(self):
|
|
req = ObligationUpdate()
|
|
data = req.model_dump(exclude_unset=True)
|
|
assert data == {}
|
|
|
|
def test_partial_update_title(self):
|
|
req = ObligationUpdate(title="Neuer Titel")
|
|
data = req.model_dump(exclude_unset=True)
|
|
assert data == {"title": "Neuer Titel"}
|
|
|
|
def test_partial_update_status_priority(self):
|
|
req = ObligationUpdate(status="completed", priority="low")
|
|
data = req.model_dump(exclude_unset=True)
|
|
assert data["status"] == "completed"
|
|
assert data["priority"] == "low"
|
|
assert "title" not in data
|
|
|
|
def test_update_linked_systems(self):
|
|
req = ObligationUpdate(linked_systems=["CRM"])
|
|
data = req.model_dump(exclude_unset=True)
|
|
assert data["linked_systems"] == ["CRM"]
|
|
|
|
def test_update_clears_linked_systems(self):
|
|
req = ObligationUpdate(linked_systems=[])
|
|
data = req.model_dump(exclude_unset=True)
|
|
assert data["linked_systems"] == []
|
|
|
|
def test_update_deadline(self):
|
|
dl = datetime(2026, 12, 31)
|
|
req = ObligationUpdate(deadline=dl)
|
|
data = req.model_dump(exclude_unset=True)
|
|
assert data["deadline"] == dl
|
|
|
|
def test_full_update(self):
|
|
req = ObligationUpdate(
|
|
title="Updated",
|
|
description="Neue Beschreibung",
|
|
source="NIS2",
|
|
source_article="Art. 21",
|
|
status="in-progress",
|
|
priority="high",
|
|
responsible="CISO",
|
|
notes="Jetzt eilt es",
|
|
)
|
|
data = req.model_dump(exclude_unset=True)
|
|
assert len(data) == 8
|
|
assert data["responsible"] == "CISO"
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Tests — ObligationStatusUpdate
|
|
# =============================================================================
|
|
|
|
class TestObligationStatusUpdate:
|
|
def test_pending(self):
|
|
req = ObligationStatusUpdate(status="pending")
|
|
assert req.status == "pending"
|
|
|
|
def test_in_progress(self):
|
|
req = ObligationStatusUpdate(status="in-progress")
|
|
assert req.status == "in-progress"
|
|
|
|
def test_completed(self):
|
|
req = ObligationStatusUpdate(status="completed")
|
|
assert req.status == "completed"
|
|
|
|
def test_overdue(self):
|
|
req = ObligationStatusUpdate(status="overdue")
|
|
assert req.status == "overdue"
|
|
|
|
def test_serialization(self):
|
|
req = ObligationStatusUpdate(status="completed")
|
|
data = req.model_dump()
|
|
assert data == {"status": "completed"}
|
|
|
|
|
|
# =============================================================================
|
|
# Helper Tests — _row_to_dict
|
|
# =============================================================================
|
|
|
|
class TestRowToDict:
|
|
def test_basic_conversion(self):
|
|
row = MagicMock()
|
|
row._mapping = {"id": "abc-123", "title": "Test Pflicht", "priority": "medium"}
|
|
result = _row_to_dict(row)
|
|
assert result["id"] == "abc-123"
|
|
assert result["title"] == "Test Pflicht"
|
|
assert result["priority"] == "medium"
|
|
|
|
def test_datetime_serialized(self):
|
|
ts = datetime(2026, 6, 1, 12, 0, 0)
|
|
row = MagicMock()
|
|
row._mapping = {"id": "abc", "created_at": ts, "updated_at": ts}
|
|
result = _row_to_dict(row)
|
|
assert result["created_at"] == ts.isoformat()
|
|
assert result["updated_at"] == ts.isoformat()
|
|
|
|
def test_deadline_serialized(self):
|
|
dl = datetime(2026, 12, 31, 23, 59, 59)
|
|
row = MagicMock()
|
|
row._mapping = {"id": "abc", "deadline": dl}
|
|
result = _row_to_dict(row)
|
|
assert result["deadline"] == dl.isoformat()
|
|
|
|
def test_none_values_preserved(self):
|
|
row = MagicMock()
|
|
row._mapping = {
|
|
"id": "abc",
|
|
"description": None,
|
|
"deadline": None,
|
|
"responsible": None,
|
|
"notes": None,
|
|
}
|
|
result = _row_to_dict(row)
|
|
assert result["description"] is None
|
|
assert result["deadline"] is None
|
|
assert result["responsible"] is None
|
|
assert result["notes"] is None
|
|
|
|
def test_uuid_converted_to_string(self):
|
|
import uuid
|
|
uid = uuid.UUID("9282a473-5c95-4b3a-bf78-0ecc0ec71d3e")
|
|
row = MagicMock()
|
|
row._mapping = {"id": uid, "tenant_id": uid}
|
|
result = _row_to_dict(row)
|
|
assert result["id"] == str(uid)
|
|
assert result["tenant_id"] == str(uid)
|
|
|
|
def test_string_fields_unchanged(self):
|
|
row = MagicMock()
|
|
row._mapping = {
|
|
"title": "DSFA durchführen",
|
|
"status": "pending",
|
|
"source": "DSGVO",
|
|
"source_article": "Art. 35",
|
|
"priority": "critical",
|
|
}
|
|
result = _row_to_dict(row)
|
|
assert result["title"] == "DSFA durchführen"
|
|
assert result["status"] == "pending"
|
|
assert result["source"] == "DSGVO"
|
|
assert result["priority"] == "critical"
|
|
|
|
def test_int_and_bool_unchanged(self):
|
|
row = MagicMock()
|
|
row._mapping = {"count": 42, "active": True, "flag": False}
|
|
result = _row_to_dict(row)
|
|
assert result["count"] == 42
|
|
assert result["active"] is True
|
|
assert result["flag"] is False
|
|
|
|
|
|
# =============================================================================
|
|
# Helper Tests — _get_tenant_id
|
|
# =============================================================================
|
|
|
|
class TestGetTenantId:
|
|
def test_valid_uuid_returned(self):
|
|
tenant_id = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
|
|
result = _get_tenant_id(x_tenant_id=tenant_id)
|
|
assert result == tenant_id
|
|
|
|
def test_different_valid_uuid(self):
|
|
tenant_id = "12345678-1234-1234-1234-123456789abc"
|
|
result = _get_tenant_id(x_tenant_id=tenant_id)
|
|
assert result == tenant_id
|
|
|
|
def test_none_returns_default(self):
|
|
result = _get_tenant_id(x_tenant_id=None)
|
|
assert result == DEFAULT_TENANT_ID
|
|
|
|
def test_invalid_uuid_returns_default(self):
|
|
result = _get_tenant_id(x_tenant_id="not-a-valid-uuid")
|
|
assert result == DEFAULT_TENANT_ID
|
|
|
|
def test_empty_string_returns_default(self):
|
|
result = _get_tenant_id(x_tenant_id="")
|
|
assert result == DEFAULT_TENANT_ID
|
|
|
|
def test_partial_uuid_returns_default(self):
|
|
result = _get_tenant_id(x_tenant_id="9282a473-5c95-4b3a")
|
|
assert result == DEFAULT_TENANT_ID
|
|
|
|
|
|
# =============================================================================
|
|
# Business Logic Tests
|
|
# =============================================================================
|
|
|
|
class TestObligationBusinessLogic:
|
|
def test_default_tenant_id_is_valid_uuid(self):
|
|
import uuid
|
|
# Should not raise
|
|
parsed = uuid.UUID(DEFAULT_TENANT_ID)
|
|
assert str(parsed) == DEFAULT_TENANT_ID
|
|
|
|
def test_valid_statuses(self):
|
|
valid = {"pending", "in-progress", "completed", "overdue"}
|
|
# Each status should be a valid string, matching what the route validates
|
|
assert "pending" in valid
|
|
assert "in-progress" in valid
|
|
assert "completed" in valid
|
|
assert "overdue" in valid
|
|
|
|
def test_valid_priorities(self):
|
|
valid = {"critical", "high", "medium", "low"}
|
|
req = ObligationCreate(title="Test", priority="critical")
|
|
assert req.priority in valid
|
|
req2 = ObligationCreate(title="Test", priority="low")
|
|
assert req2.priority in valid
|
|
|
|
def test_priority_order_correctness(self):
|
|
"""Ensure priority values match the SQL CASE ordering in the route."""
|
|
priorities_ordered = ["critical", "high", "medium", "low"]
|
|
for i, p in enumerate(priorities_ordered):
|
|
req = ObligationCreate(title=f"Test {p}", priority=p)
|
|
assert req.priority == p
|
|
|
|
def test_linked_systems_defaults_to_none(self):
|
|
req = ObligationCreate(title="Test")
|
|
assert req.linked_systems is None
|
|
|
|
def test_linked_systems_can_be_empty_list(self):
|
|
req = ObligationCreate(title="Test", linked_systems=[])
|
|
assert req.linked_systems == []
|
|
|
|
def test_linked_systems_multiple_items(self):
|
|
systems = ["CRM", "ERP", "HR-System", "Buchhaltung"]
|
|
req = ObligationCreate(title="Test", linked_systems=systems)
|
|
assert len(req.linked_systems) == 4
|
|
assert "ERP" in req.linked_systems
|
|
|
|
def test_source_defaults(self):
|
|
"""Verify all common DSGVO/AI Act sources can be stored."""
|
|
for source in ["DSGVO", "AI Act", "NIS2", "BDSG", "ISO 27001"]:
|
|
req = ObligationCreate(title="Test", source=source)
|
|
assert req.source == source
|