Benjamin_Boenisch/breakpilot-compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(vvt): Go-Features nach Python portieren (Source of Truth)

Browse Source 
Review-Daten (last_reviewed_at, next_review_at), created_by, DSFA-Link,
CSV-Export mit Semikolon-Trennung, overdue_review_count in Stats.
Go-VVT-Handler als DEPRECATED markiert. 32 Tests bestanden.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-06 17:14:38 +01:00
parent 885b97d422
commit 4cbfea5c1d
7 changed files with 330 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
backend-compliance/tests/test_vvt_routes.py
View File

@@ -2,7 +2,7 @@
import pytest
from unittest.mock import MagicMock, patch
from datetime import datetime, date
from datetime import datetime, date, timedelta, timezone
import uuid
from compliance.api.schemas import (
@@ -104,10 +104,24 @@ class TestVVTStatsResponse:
third_country_count=0,
draft_count=3,
approved_count=2,
overdue_review_count=1,
)
assert stats.total == 5
assert stats.by_status["DRAFT"] == 3
assert stats.dpia_required_count == 1
assert stats.overdue_review_count == 1
def test_stats_overdue_default_zero(self):
stats = VVTStatsResponse(
total=0,
by_status={},
by_business_function={},
dpia_required_count=0,
third_country_count=0,
draft_count=0,
approved_count=0,
)
assert stats.overdue_review_count == 0
# =============================================================================
@@ -168,6 +182,10 @@ class TestActivityToResponse:
act.status = kwargs.get("status", "DRAFT")
act.responsible = kwargs.get("responsible", None)
act.owner = kwargs.get("owner", None)
act.last_reviewed_at = kwargs.get("last_reviewed_at", None)
act.next_review_at = kwargs.get("next_review_at", None)
act.created_by = kwargs.get("created_by", None)
act.dsfa_id = kwargs.get("dsfa_id", None)
act.created_at = datetime.utcnow()
act.updated_at = None
return act
@@ -220,3 +238,196 @@ class TestLogAudit:
_log_audit(mock_db, "DELETE", "activity")
added = mock_db.add.call_args[0][0]
assert added.changed_by == "system"
# =============================================================================
# Consolidation Tests (Go → Python feature parity)
# =============================================================================
class TestVVTConsolidationSchemas:
"""Tests for new fields ported from Go: review dates, created_by, dsfa_id."""
def test_activity_create_with_review_dates(self):
now = datetime.now(timezone.utc)
future = now + timedelta(days=365)
req = VVTActivityCreate(
vvt_id="VVT-REV-001",
name="Review-Test",
last_reviewed_at=now,
next_review_at=future,
)
assert req.last_reviewed_at == now
assert req.next_review_at == future
def test_activity_create_sets_created_by(self):
req = VVTActivityCreate(
vvt_id="VVT-CB-001",
name="Created-By Test",
created_by="admin@example.com",
)
assert req.created_by == "admin@example.com"
def test_activity_create_created_by_defaults_none(self):
req = VVTActivityCreate(vvt_id="VVT-CB-002", name="Default Test")
assert req.created_by is None
def test_activity_create_with_dsfa_id(self):
dsfa_uuid = str(uuid.uuid4())
req = VVTActivityCreate(
vvt_id="VVT-DSFA-001",
name="DSFA-Link Test",
dsfa_id=dsfa_uuid,
)
assert req.dsfa_id == dsfa_uuid
def test_activity_update_review_dates(self):
now = datetime.now(timezone.utc)
req = VVTActivityUpdate(
last_reviewed_at=now,
next_review_at=now + timedelta(days=180),
)
data = req.model_dump(exclude_none=True)
assert "last_reviewed_at" in data
assert "next_review_at" in data
def test_activity_update_dsfa_id(self):
dsfa_uuid = str(uuid.uuid4())
req = VVTActivityUpdate(dsfa_id=dsfa_uuid)
data = req.model_dump(exclude_none=True)
assert data["dsfa_id"] == dsfa_uuid
class TestVVTConsolidationResponse:
"""Tests for new fields in response mapping."""
def _make_activity(self, **kwargs) -> VVTActivityDB:
act = VVTActivityDB()
act.id = uuid.uuid4()
act.vvt_id = kwargs.get("vvt_id", "VVT-001")
act.name = kwargs.get("name", "Test")
act.description = None
act.purposes = []
act.legal_bases = []
act.data_subject_categories = []
act.personal_data_categories = []
act.recipient_categories = []
act.third_country_transfers = []
act.retention_period = {}
act.tom_description = None
act.business_function = None
act.systems = []
act.deployment_model = None
act.data_sources = []
act.data_flows = []
act.protection_level = "MEDIUM"
act.dpia_required = False
act.structured_toms = {}
act.status = "DRAFT"
act.responsible = None
act.owner = None
act.last_reviewed_at = kwargs.get("last_reviewed_at", None)
act.next_review_at = kwargs.get("next_review_at", None)
act.created_by = kwargs.get("created_by", None)
act.dsfa_id = kwargs.get("dsfa_id", None)
act.created_at = datetime.utcnow()
act.updated_at = None
return act
def test_response_includes_review_dates(self):
now = datetime.now(timezone.utc)
future = now + timedelta(days=365)
act = self._make_activity(last_reviewed_at=now, next_review_at=future)
resp = _activity_to_response(act)
assert resp.last_reviewed_at == now
assert resp.next_review_at == future
def test_response_includes_created_by(self):
act = self._make_activity(created_by="admin@example.com")
resp = _activity_to_response(act)
assert resp.created_by == "admin@example.com"
def test_response_includes_dsfa_id(self):
dsfa_uuid = uuid.uuid4()
act = self._make_activity(dsfa_id=dsfa_uuid)
resp = _activity_to_response(act)
assert resp.dsfa_id == str(dsfa_uuid)
def test_response_null_new_fields(self):
act = self._make_activity()
resp = _activity_to_response(act)
assert resp.last_reviewed_at is None
assert resp.next_review_at is None
assert resp.created_by is None
assert resp.dsfa_id is None
class TestVVTCsvExport:
"""Tests for CSV export functionality."""
def _collect_csv_body(self, response) -> str:
"""Extract text from StreamingResponse (async generator)."""
import asyncio
async def _read():
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk)
return ''.join(chunks)
return asyncio.get_event_loop().run_until_complete(_read())
def test_export_csv_format(self):
from compliance.api.vvt_routes import _export_csv
act = VVTActivityDB()
act.id = uuid.uuid4()
act.vvt_id = "VVT-CSV-001"
act.name = "CSV Test"
act.purposes = ["Zweck A", "Zweck B"]
act.legal_bases = ["Art. 6 Abs. 1b"]
act.personal_data_categories = ["Email"]
act.data_subject_categories = ["Kunden"]
act.recipient_categories = ["IT-Dienstleister"]
act.third_country_transfers = ["USA"]
act.retention_period = {"duration": "3 Jahre"}
act.status = "APPROVED"
act.responsible = "DSB"
act.created_by = "admin"
act.created_at = datetime(2026, 1, 15, 10, 30)
act.updated_at = None
response = _export_csv([act])
text = self._collect_csv_body(response)
assert 'VVT-CSV-001' in text
assert 'CSV Test' in text
assert 'APPROVED' in text
def test_export_csv_semicolon_separator(self):
from compliance.api.vvt_routes import _export_csv
act = VVTActivityDB()
act.id = uuid.uuid4()
act.vvt_id = "VVT-SEP-001"
act.name = "Separator Test"
act.purposes = []
act.legal_bases = []
act.personal_data_categories = []
act.data_subject_categories = []
act.recipient_categories = []
act.third_country_transfers = []
act.retention_period = {}
act.status = "DRAFT"
act.responsible = ""
act.created_by = "system"
act.created_at = datetime(2026, 3, 1, 12, 0)
act.updated_at = None
response = _export_csv([act])
text = self._collect_csv_body(response)
lines = text.strip().split('\n')
header = lines[0]
assert ';' in header
assert 'ID;VVT-ID;Name' in header.replace('\ufeff', '')
def test_export_csv_empty_list(self):
from compliance.api.vvt_routes import _export_csv
response = _export_csv([])
text = self._collect_csv_body(response)
lines = text.strip().split('\n')
assert len(lines) == 1