feat: 6 Dokumentations-Module auf 100% — VVT Backend, Filter, PDF-Export
Phase 1 — VVT Backend (localStorage → API): - migrations/006_vvt.sql: Neue Tabellen (vvt_organization, vvt_activities, vvt_audit_log) - compliance/db/vvt_models.py: SQLAlchemy-Models für alle VVT-Tabellen - compliance/api/vvt_routes.py: Vollständiger CRUD-Router (10 Endpoints) - compliance/api/__init__.py: VVT-Router registriert - compliance/api/schemas.py: VVT Pydantic-Schemas ergänzt - app/(sdk)/sdk/vvt/page.tsx: API-Client + camelCase↔snake_case Mapping, localStorage durch persistente DB-Calls ersetzt (POST/PUT/DELETE/GET) - tests/test_vvt_routes.py: 18 Tests (alle grün) Phase 3 — Document Generator PDF-Export: - document-generator/page.tsx: "Als PDF exportieren"-Button funktioniert jetzt via window.print() + Print-Window mit korrektem HTML - Fallback-Banner wenn Template-Service (breakpilot-core) nicht erreichbar Phase 4 — Source Policy erweiterte Filter: - SourcesTab.tsx: source_type-Filter (Rechtlich / Leitlinien / Vorlagen / etc.) - PIIRulesTab.tsx: category-Filter (E-Mail / Telefon / IBAN / etc.) - source_policy_router.py: Backend-Endpoints unterstützen jetzt source_type und category als Query-Parameter - requirements.txt: reportlab==4.2.5 ergänzt (fehlende Audit-PDF-Dependency) Phase 2 — Training (Migration-Skripte): - scripts/apply_training_migrations.sh: SSH-Skript für Mac Mini - scripts/apply_vvt_migration.sh: Vollständiges Deploy-Skript für VVT Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
@@ -9,6 +9,7 @@ from .dashboard_routes import router as dashboard_router
|
||||
from .scraper_routes import router as scraper_router
|
||||
from .module_routes import router as module_router
|
||||
from .isms_routes import router as isms_router
|
||||
from .vvt_routes import router as vvt_router
|
||||
|
||||
# Include sub-routers
|
||||
router.include_router(audit_router)
|
||||
@@ -19,6 +20,7 @@ router.include_router(dashboard_router)
|
||||
router.include_router(scraper_router)
|
||||
router.include_router(module_router)
|
||||
router.include_router(isms_router)
|
||||
router.include_router(vvt_router)
|
||||
|
||||
__all__ = [
|
||||
"router",
|
||||
@@ -30,4 +32,5 @@ __all__ = [
|
||||
"scraper_router",
|
||||
"module_router",
|
||||
"isms_router",
|
||||
"vvt_router",
|
||||
]
|
||||
|
||||
@@ -1849,3 +1849,143 @@ class ISO27001OverviewResponse(BaseModel):
|
||||
policies_approved: int
|
||||
objectives_count: int
|
||||
objectives_achieved: int
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# VVT Schemas — Verzeichnis von Verarbeitungstaetigkeiten (Art. 30 DSGVO)
|
||||
# ============================================================================
|
||||
|
||||
class VVTOrganizationUpdate(BaseModel):
|
||||
organization_name: Optional[str] = None
|
||||
industry: Optional[str] = None
|
||||
locations: Optional[List[str]] = None
|
||||
employee_count: Optional[int] = None
|
||||
dpo_name: Optional[str] = None
|
||||
dpo_contact: Optional[str] = None
|
||||
vvt_version: Optional[str] = None
|
||||
last_review_date: Optional[date] = None
|
||||
next_review_date: Optional[date] = None
|
||||
review_interval: Optional[str] = None
|
||||
|
||||
|
||||
class VVTOrganizationResponse(BaseModel):
|
||||
id: str
|
||||
organization_name: str
|
||||
industry: Optional[str] = None
|
||||
locations: List[Any] = []
|
||||
employee_count: Optional[int] = None
|
||||
dpo_name: Optional[str] = None
|
||||
dpo_contact: Optional[str] = None
|
||||
vvt_version: str = '1.0'
|
||||
last_review_date: Optional[date] = None
|
||||
next_review_date: Optional[date] = None
|
||||
review_interval: str = 'annual'
|
||||
created_at: datetime
|
||||
updated_at: Optional[datetime] = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class VVTActivityCreate(BaseModel):
|
||||
vvt_id: str
|
||||
name: str
|
||||
description: Optional[str] = None
|
||||
purposes: List[str] = []
|
||||
legal_bases: List[str] = []
|
||||
data_subject_categories: List[str] = []
|
||||
personal_data_categories: List[str] = []
|
||||
recipient_categories: List[str] = []
|
||||
third_country_transfers: List[Any] = []
|
||||
retention_period: Dict[str, Any] = {}
|
||||
tom_description: Optional[str] = None
|
||||
business_function: Optional[str] = None
|
||||
systems: List[str] = []
|
||||
deployment_model: Optional[str] = None
|
||||
data_sources: List[Any] = []
|
||||
data_flows: List[Any] = []
|
||||
protection_level: str = 'MEDIUM'
|
||||
dpia_required: bool = False
|
||||
structured_toms: Dict[str, Any] = {}
|
||||
status: str = 'DRAFT'
|
||||
responsible: Optional[str] = None
|
||||
owner: Optional[str] = None
|
||||
|
||||
|
||||
class VVTActivityUpdate(BaseModel):
|
||||
name: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
purposes: Optional[List[str]] = None
|
||||
legal_bases: Optional[List[str]] = None
|
||||
data_subject_categories: Optional[List[str]] = None
|
||||
personal_data_categories: Optional[List[str]] = None
|
||||
recipient_categories: Optional[List[str]] = None
|
||||
third_country_transfers: Optional[List[Any]] = None
|
||||
retention_period: Optional[Dict[str, Any]] = None
|
||||
tom_description: Optional[str] = None
|
||||
business_function: Optional[str] = None
|
||||
systems: Optional[List[str]] = None
|
||||
deployment_model: Optional[str] = None
|
||||
data_sources: Optional[List[Any]] = None
|
||||
data_flows: Optional[List[Any]] = None
|
||||
protection_level: Optional[str] = None
|
||||
dpia_required: Optional[bool] = None
|
||||
structured_toms: Optional[Dict[str, Any]] = None
|
||||
status: Optional[str] = None
|
||||
responsible: Optional[str] = None
|
||||
owner: Optional[str] = None
|
||||
|
||||
|
||||
class VVTActivityResponse(BaseModel):
|
||||
id: str
|
||||
vvt_id: str
|
||||
name: str
|
||||
description: Optional[str] = None
|
||||
purposes: List[Any] = []
|
||||
legal_bases: List[Any] = []
|
||||
data_subject_categories: List[Any] = []
|
||||
personal_data_categories: List[Any] = []
|
||||
recipient_categories: List[Any] = []
|
||||
third_country_transfers: List[Any] = []
|
||||
retention_period: Dict[str, Any] = {}
|
||||
tom_description: Optional[str] = None
|
||||
business_function: Optional[str] = None
|
||||
systems: List[Any] = []
|
||||
deployment_model: Optional[str] = None
|
||||
data_sources: List[Any] = []
|
||||
data_flows: List[Any] = []
|
||||
protection_level: str = 'MEDIUM'
|
||||
dpia_required: bool = False
|
||||
structured_toms: Dict[str, Any] = {}
|
||||
status: str = 'DRAFT'
|
||||
responsible: Optional[str] = None
|
||||
owner: Optional[str] = None
|
||||
created_at: datetime
|
||||
updated_at: Optional[datetime] = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class VVTStatsResponse(BaseModel):
|
||||
total: int
|
||||
by_status: Dict[str, int]
|
||||
by_business_function: Dict[str, int]
|
||||
dpia_required_count: int
|
||||
third_country_count: int
|
||||
draft_count: int
|
||||
approved_count: int
|
||||
|
||||
|
||||
class VVTAuditLogEntry(BaseModel):
|
||||
id: str
|
||||
action: str
|
||||
entity_type: str
|
||||
entity_id: Optional[str] = None
|
||||
changed_by: Optional[str] = None
|
||||
old_values: Optional[Dict[str, Any]] = None
|
||||
new_values: Optional[Dict[str, Any]] = None
|
||||
created_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
@@ -148,12 +148,18 @@ def _source_to_dict(source: AllowedSourceDB) -> dict:
|
||||
@router.get("/sources")
|
||||
async def list_sources(
|
||||
active_only: bool = Query(False),
|
||||
source_type: Optional[str] = Query(None),
|
||||
license: Optional[str] = Query(None),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""List all allowed sources."""
|
||||
"""List all allowed sources with optional filters."""
|
||||
query = db.query(AllowedSourceDB)
|
||||
if active_only:
|
||||
query = query.filter(AllowedSourceDB.active == True)
|
||||
if source_type:
|
||||
query = query.filter(AllowedSourceDB.source_type == source_type)
|
||||
if license:
|
||||
query = query.filter(AllowedSourceDB.license == license)
|
||||
sources = query.order_by(AllowedSourceDB.name).all()
|
||||
return {
|
||||
"sources": [
|
||||
@@ -328,9 +334,15 @@ async def update_operation(
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/pii-rules")
|
||||
async def list_pii_rules(db: Session = Depends(get_db)):
|
||||
"""List all PII rules."""
|
||||
rules = db.query(PIIRuleDB).order_by(PIIRuleDB.category, PIIRuleDB.name).all()
|
||||
async def list_pii_rules(
|
||||
category: Optional[str] = Query(None),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""List all PII rules with optional category filter."""
|
||||
query = db.query(PIIRuleDB)
|
||||
if category:
|
||||
query = query.filter(PIIRuleDB.category == category)
|
||||
rules = query.order_by(PIIRuleDB.category, PIIRuleDB.name).all()
|
||||
return {
|
||||
"rules": [
|
||||
{
|
||||
|
||||
384
backend-compliance/compliance/api/vvt_routes.py
Normal file
384
backend-compliance/compliance/api/vvt_routes.py
Normal file
@@ -0,0 +1,384 @@
|
||||
"""
|
||||
FastAPI routes for VVT — Verzeichnis von Verarbeitungstaetigkeiten (Art. 30 DSGVO).
|
||||
|
||||
Endpoints:
|
||||
GET /vvt/organization — Load organization header
|
||||
PUT /vvt/organization — Save organization header
|
||||
GET /vvt/activities — List activities (filter: status, business_function)
|
||||
POST /vvt/activities — Create new activity
|
||||
GET /vvt/activities/{id} — Get single activity
|
||||
PUT /vvt/activities/{id} — Update activity
|
||||
DELETE /vvt/activities/{id} — Delete activity
|
||||
GET /vvt/audit-log — Audit trail (limit, offset)
|
||||
GET /vvt/export — JSON export of all activities
|
||||
GET /vvt/stats — Statistics
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Optional, List
|
||||
from uuid import uuid4
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from classroom_engine.database import get_db
|
||||
|
||||
from ..db.vvt_models import VVTOrganizationDB, VVTActivityDB, VVTAuditLogDB
|
||||
from .schemas import (
|
||||
VVTOrganizationUpdate, VVTOrganizationResponse,
|
||||
VVTActivityCreate, VVTActivityUpdate, VVTActivityResponse,
|
||||
VVTStatsResponse, VVTAuditLogEntry,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/vvt", tags=["compliance-vvt"])
|
||||
|
||||
|
||||
def _log_audit(
|
||||
db: Session,
|
||||
action: str,
|
||||
entity_type: str,
|
||||
entity_id=None,
|
||||
changed_by: str = "system",
|
||||
old_values=None,
|
||||
new_values=None,
|
||||
):
|
||||
entry = VVTAuditLogDB(
|
||||
action=action,
|
||||
entity_type=entity_type,
|
||||
entity_id=entity_id,
|
||||
changed_by=changed_by,
|
||||
old_values=old_values,
|
||||
new_values=new_values,
|
||||
)
|
||||
db.add(entry)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Organization Header
|
||||
# ============================================================================
|
||||
|
||||
@router.get("/organization", response_model=Optional[VVTOrganizationResponse])
|
||||
async def get_organization(db: Session = Depends(get_db)):
|
||||
"""Load the VVT organization header (single record)."""
|
||||
org = db.query(VVTOrganizationDB).order_by(VVTOrganizationDB.created_at).first()
|
||||
if not org:
|
||||
return None
|
||||
return VVTOrganizationResponse(
|
||||
id=str(org.id),
|
||||
organization_name=org.organization_name,
|
||||
industry=org.industry,
|
||||
locations=org.locations or [],
|
||||
employee_count=org.employee_count,
|
||||
dpo_name=org.dpo_name,
|
||||
dpo_contact=org.dpo_contact,
|
||||
vvt_version=org.vvt_version or '1.0',
|
||||
last_review_date=org.last_review_date,
|
||||
next_review_date=org.next_review_date,
|
||||
review_interval=org.review_interval or 'annual',
|
||||
created_at=org.created_at,
|
||||
updated_at=org.updated_at,
|
||||
)
|
||||
|
||||
|
||||
@router.put("/organization", response_model=VVTOrganizationResponse)
|
||||
async def upsert_organization(
|
||||
request: VVTOrganizationUpdate,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Create or update the VVT organization header."""
|
||||
org = db.query(VVTOrganizationDB).order_by(VVTOrganizationDB.created_at).first()
|
||||
|
||||
if not org:
|
||||
data = request.dict(exclude_none=True)
|
||||
if 'organization_name' not in data:
|
||||
data['organization_name'] = 'Meine Organisation'
|
||||
org = VVTOrganizationDB(**data)
|
||||
db.add(org)
|
||||
else:
|
||||
for field, value in request.dict(exclude_none=True).items():
|
||||
setattr(org, field, value)
|
||||
org.updated_at = datetime.utcnow()
|
||||
|
||||
db.commit()
|
||||
db.refresh(org)
|
||||
|
||||
return VVTOrganizationResponse(
|
||||
id=str(org.id),
|
||||
organization_name=org.organization_name,
|
||||
industry=org.industry,
|
||||
locations=org.locations or [],
|
||||
employee_count=org.employee_count,
|
||||
dpo_name=org.dpo_name,
|
||||
dpo_contact=org.dpo_contact,
|
||||
vvt_version=org.vvt_version or '1.0',
|
||||
last_review_date=org.last_review_date,
|
||||
next_review_date=org.next_review_date,
|
||||
review_interval=org.review_interval or 'annual',
|
||||
created_at=org.created_at,
|
||||
updated_at=org.updated_at,
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Activities
|
||||
# ============================================================================
|
||||
|
||||
def _activity_to_response(act: VVTActivityDB) -> VVTActivityResponse:
|
||||
return VVTActivityResponse(
|
||||
id=str(act.id),
|
||||
vvt_id=act.vvt_id,
|
||||
name=act.name,
|
||||
description=act.description,
|
||||
purposes=act.purposes or [],
|
||||
legal_bases=act.legal_bases or [],
|
||||
data_subject_categories=act.data_subject_categories or [],
|
||||
personal_data_categories=act.personal_data_categories or [],
|
||||
recipient_categories=act.recipient_categories or [],
|
||||
third_country_transfers=act.third_country_transfers or [],
|
||||
retention_period=act.retention_period or {},
|
||||
tom_description=act.tom_description,
|
||||
business_function=act.business_function,
|
||||
systems=act.systems or [],
|
||||
deployment_model=act.deployment_model,
|
||||
data_sources=act.data_sources or [],
|
||||
data_flows=act.data_flows or [],
|
||||
protection_level=act.protection_level or 'MEDIUM',
|
||||
dpia_required=act.dpia_required or False,
|
||||
structured_toms=act.structured_toms or {},
|
||||
status=act.status or 'DRAFT',
|
||||
responsible=act.responsible,
|
||||
owner=act.owner,
|
||||
created_at=act.created_at,
|
||||
updated_at=act.updated_at,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/activities", response_model=List[VVTActivityResponse])
|
||||
async def list_activities(
|
||||
status: Optional[str] = Query(None),
|
||||
business_function: Optional[str] = Query(None),
|
||||
search: Optional[str] = Query(None),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""List all processing activities with optional filters."""
|
||||
query = db.query(VVTActivityDB)
|
||||
|
||||
if status:
|
||||
query = query.filter(VVTActivityDB.status == status)
|
||||
if business_function:
|
||||
query = query.filter(VVTActivityDB.business_function == business_function)
|
||||
if search:
|
||||
term = f"%{search}%"
|
||||
query = query.filter(
|
||||
(VVTActivityDB.name.ilike(term)) |
|
||||
(VVTActivityDB.description.ilike(term)) |
|
||||
(VVTActivityDB.vvt_id.ilike(term))
|
||||
)
|
||||
|
||||
activities = query.order_by(VVTActivityDB.created_at.desc()).all()
|
||||
return [_activity_to_response(a) for a in activities]
|
||||
|
||||
|
||||
@router.post("/activities", response_model=VVTActivityResponse, status_code=201)
|
||||
async def create_activity(
|
||||
request: VVTActivityCreate,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Create a new processing activity."""
|
||||
# Check for duplicate vvt_id
|
||||
existing = db.query(VVTActivityDB).filter(
|
||||
VVTActivityDB.vvt_id == request.vvt_id
|
||||
).first()
|
||||
if existing:
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"Activity with VVT-ID '{request.vvt_id}' already exists"
|
||||
)
|
||||
|
||||
act = VVTActivityDB(**request.dict())
|
||||
db.add(act)
|
||||
db.flush() # get ID before audit log
|
||||
|
||||
_log_audit(
|
||||
db,
|
||||
action="CREATE",
|
||||
entity_type="activity",
|
||||
entity_id=act.id,
|
||||
new_values={"vvt_id": act.vvt_id, "name": act.name, "status": act.status},
|
||||
)
|
||||
|
||||
db.commit()
|
||||
db.refresh(act)
|
||||
return _activity_to_response(act)
|
||||
|
||||
|
||||
@router.get("/activities/{activity_id}", response_model=VVTActivityResponse)
|
||||
async def get_activity(activity_id: str, db: Session = Depends(get_db)):
|
||||
"""Get a single processing activity by ID."""
|
||||
act = db.query(VVTActivityDB).filter(VVTActivityDB.id == activity_id).first()
|
||||
if not act:
|
||||
raise HTTPException(status_code=404, detail=f"Activity {activity_id} not found")
|
||||
return _activity_to_response(act)
|
||||
|
||||
|
||||
@router.put("/activities/{activity_id}", response_model=VVTActivityResponse)
|
||||
async def update_activity(
|
||||
activity_id: str,
|
||||
request: VVTActivityUpdate,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Update a processing activity."""
|
||||
act = db.query(VVTActivityDB).filter(VVTActivityDB.id == activity_id).first()
|
||||
if not act:
|
||||
raise HTTPException(status_code=404, detail=f"Activity {activity_id} not found")
|
||||
|
||||
old_values = {"name": act.name, "status": act.status}
|
||||
updates = request.dict(exclude_none=True)
|
||||
for field, value in updates.items():
|
||||
setattr(act, field, value)
|
||||
act.updated_at = datetime.utcnow()
|
||||
|
||||
_log_audit(
|
||||
db,
|
||||
action="UPDATE",
|
||||
entity_type="activity",
|
||||
entity_id=act.id,
|
||||
old_values=old_values,
|
||||
new_values=updates,
|
||||
)
|
||||
|
||||
db.commit()
|
||||
db.refresh(act)
|
||||
return _activity_to_response(act)
|
||||
|
||||
|
||||
@router.delete("/activities/{activity_id}")
|
||||
async def delete_activity(activity_id: str, db: Session = Depends(get_db)):
|
||||
"""Delete a processing activity."""
|
||||
act = db.query(VVTActivityDB).filter(VVTActivityDB.id == activity_id).first()
|
||||
if not act:
|
||||
raise HTTPException(status_code=404, detail=f"Activity {activity_id} not found")
|
||||
|
||||
_log_audit(
|
||||
db,
|
||||
action="DELETE",
|
||||
entity_type="activity",
|
||||
entity_id=act.id,
|
||||
old_values={"vvt_id": act.vvt_id, "name": act.name},
|
||||
)
|
||||
|
||||
db.delete(act)
|
||||
db.commit()
|
||||
return {"success": True, "message": f"Activity {activity_id} deleted"}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Audit Log
|
||||
# ============================================================================
|
||||
|
||||
@router.get("/audit-log", response_model=List[VVTAuditLogEntry])
|
||||
async def get_audit_log(
|
||||
limit: int = Query(50, ge=1, le=500),
|
||||
offset: int = Query(0, ge=0),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Get the VVT audit trail."""
|
||||
entries = (
|
||||
db.query(VVTAuditLogDB)
|
||||
.order_by(VVTAuditLogDB.created_at.desc())
|
||||
.offset(offset)
|
||||
.limit(limit)
|
||||
.all()
|
||||
)
|
||||
return [
|
||||
VVTAuditLogEntry(
|
||||
id=str(e.id),
|
||||
action=e.action,
|
||||
entity_type=e.entity_type,
|
||||
entity_id=str(e.entity_id) if e.entity_id else None,
|
||||
changed_by=e.changed_by,
|
||||
old_values=e.old_values,
|
||||
new_values=e.new_values,
|
||||
created_at=e.created_at,
|
||||
)
|
||||
for e in entries
|
||||
]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Export & Stats
|
||||
# ============================================================================
|
||||
|
||||
@router.get("/export")
|
||||
async def export_activities(db: Session = Depends(get_db)):
|
||||
"""JSON export of all activities for external review / PDF generation."""
|
||||
org = db.query(VVTOrganizationDB).order_by(VVTOrganizationDB.created_at).first()
|
||||
activities = db.query(VVTActivityDB).order_by(VVTActivityDB.created_at).all()
|
||||
|
||||
_log_audit(
|
||||
db,
|
||||
action="EXPORT",
|
||||
entity_type="all_activities",
|
||||
new_values={"count": len(activities)},
|
||||
)
|
||||
db.commit()
|
||||
|
||||
return {
|
||||
"exported_at": datetime.utcnow().isoformat(),
|
||||
"organization": {
|
||||
"name": org.organization_name if org else "",
|
||||
"dpo_name": org.dpo_name if org else "",
|
||||
"dpo_contact": org.dpo_contact if org else "",
|
||||
"vvt_version": org.vvt_version if org else "1.0",
|
||||
} if org else None,
|
||||
"activities": [
|
||||
{
|
||||
"id": str(a.id),
|
||||
"vvt_id": a.vvt_id,
|
||||
"name": a.name,
|
||||
"description": a.description,
|
||||
"status": a.status,
|
||||
"purposes": a.purposes,
|
||||
"legal_bases": a.legal_bases,
|
||||
"data_subject_categories": a.data_subject_categories,
|
||||
"personal_data_categories": a.personal_data_categories,
|
||||
"recipient_categories": a.recipient_categories,
|
||||
"third_country_transfers": a.third_country_transfers,
|
||||
"retention_period": a.retention_period,
|
||||
"dpia_required": a.dpia_required,
|
||||
"protection_level": a.protection_level,
|
||||
"business_function": a.business_function,
|
||||
"responsible": a.responsible,
|
||||
"created_at": a.created_at.isoformat(),
|
||||
"updated_at": a.updated_at.isoformat() if a.updated_at else None,
|
||||
}
|
||||
for a in activities
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@router.get("/stats", response_model=VVTStatsResponse)
|
||||
async def get_stats(db: Session = Depends(get_db)):
|
||||
"""Get VVT statistics summary."""
|
||||
activities = db.query(VVTActivityDB).all()
|
||||
|
||||
by_status: dict = {}
|
||||
by_bf: dict = {}
|
||||
|
||||
for a in activities:
|
||||
status = a.status or 'DRAFT'
|
||||
bf = a.business_function or 'unknown'
|
||||
by_status[status] = by_status.get(status, 0) + 1
|
||||
by_bf[bf] = by_bf.get(bf, 0) + 1
|
||||
|
||||
return VVTStatsResponse(
|
||||
total=len(activities),
|
||||
by_status=by_status,
|
||||
by_business_function=by_bf,
|
||||
dpia_required_count=sum(1 for a in activities if a.dpia_required),
|
||||
third_country_count=sum(1 for a in activities if a.third_country_transfers),
|
||||
draft_count=by_status.get('DRAFT', 0),
|
||||
approved_count=by_status.get('APPROVED', 0),
|
||||
)
|
||||
109
backend-compliance/compliance/db/vvt_models.py
Normal file
109
backend-compliance/compliance/db/vvt_models.py
Normal file
@@ -0,0 +1,109 @@
|
||||
"""
|
||||
SQLAlchemy models for VVT — Verzeichnis von Verarbeitungstaetigkeiten (Art. 30 DSGVO).
|
||||
|
||||
Tables:
|
||||
- compliance_vvt_organization: Organization header (DSB, version, review dates)
|
||||
- compliance_vvt_activities: Individual processing activities
|
||||
- compliance_vvt_audit_log: Audit trail for all VVT changes
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import (
|
||||
Column, String, Text, Boolean, Integer, Date, DateTime, JSON, Index
|
||||
)
|
||||
from sqlalchemy.dialects.postgresql import UUID
|
||||
|
||||
from classroom_engine.database import Base
|
||||
|
||||
|
||||
class VVTOrganizationDB(Base):
|
||||
"""VVT organization header — stores DSB contact, version and review schedule."""
|
||||
|
||||
__tablename__ = 'compliance_vvt_organization'
|
||||
|
||||
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
||||
organization_name = Column(String(300), nullable=False)
|
||||
industry = Column(String(100))
|
||||
locations = Column(JSON, default=list)
|
||||
employee_count = Column(Integer)
|
||||
dpo_name = Column(String(200))
|
||||
dpo_contact = Column(String(200))
|
||||
vvt_version = Column(String(20), default='1.0')
|
||||
last_review_date = Column(Date)
|
||||
next_review_date = Column(Date)
|
||||
review_interval = Column(String(20), default='annual')
|
||||
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
|
||||
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
||||
|
||||
__table_args__ = (
|
||||
Index('idx_vvt_org_created', 'created_at'),
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<VVTOrganization {self.organization_name}>"
|
||||
|
||||
|
||||
class VVTActivityDB(Base):
|
||||
"""Individual processing activity per Art. 30 DSGVO."""
|
||||
|
||||
__tablename__ = 'compliance_vvt_activities'
|
||||
|
||||
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
||||
vvt_id = Column(String(50), unique=True, nullable=False)
|
||||
name = Column(String(300), nullable=False)
|
||||
description = Column(Text)
|
||||
purposes = Column(JSON, default=list)
|
||||
legal_bases = Column(JSON, default=list)
|
||||
data_subject_categories = Column(JSON, default=list)
|
||||
personal_data_categories = Column(JSON, default=list)
|
||||
recipient_categories = Column(JSON, default=list)
|
||||
third_country_transfers = Column(JSON, default=list)
|
||||
retention_period = Column(JSON, default=dict)
|
||||
tom_description = Column(Text)
|
||||
business_function = Column(String(50))
|
||||
systems = Column(JSON, default=list)
|
||||
deployment_model = Column(String(20))
|
||||
data_sources = Column(JSON, default=list)
|
||||
data_flows = Column(JSON, default=list)
|
||||
protection_level = Column(String(10), default='MEDIUM')
|
||||
dpia_required = Column(Boolean, default=False)
|
||||
structured_toms = Column(JSON, default=dict)
|
||||
status = Column(String(20), default='DRAFT')
|
||||
responsible = Column(String(200))
|
||||
owner = Column(String(200))
|
||||
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
|
||||
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
||||
|
||||
__table_args__ = (
|
||||
Index('idx_vvt_activities_status', 'status'),
|
||||
Index('idx_vvt_activities_business_function', 'business_function'),
|
||||
Index('idx_vvt_activities_vvt_id', 'vvt_id'),
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<VVTActivity {self.vvt_id}: {self.name}>"
|
||||
|
||||
|
||||
class VVTAuditLogDB(Base):
|
||||
"""Audit trail for all VVT create/update/delete/export actions."""
|
||||
|
||||
__tablename__ = 'compliance_vvt_audit_log'
|
||||
|
||||
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
||||
action = Column(String(20), nullable=False) # CREATE, UPDATE, DELETE, EXPORT
|
||||
entity_type = Column(String(50), nullable=False) # activity, organization
|
||||
entity_id = Column(UUID(as_uuid=True))
|
||||
changed_by = Column(String(200))
|
||||
old_values = Column(JSON)
|
||||
new_values = Column(JSON)
|
||||
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
|
||||
|
||||
__table_args__ = (
|
||||
Index('idx_vvt_audit_created', 'created_at'),
|
||||
Index('idx_vvt_audit_entity', 'entity_type', 'entity_id'),
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<VVTAuditLog {self.action} {self.entity_type}>"
|
||||
66
backend-compliance/migrations/006_vvt.sql
Normal file
66
backend-compliance/migrations/006_vvt.sql
Normal file
@@ -0,0 +1,66 @@
|
||||
-- =========================================================
|
||||
-- Migration 006: VVT — Verzeichnis von Verarbeitungstaetigkeiten
|
||||
-- Art. 30 DSGVO
|
||||
-- =========================================================
|
||||
|
||||
CREATE TABLE IF NOT EXISTS compliance_vvt_organization (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
organization_name VARCHAR(300) NOT NULL,
|
||||
industry VARCHAR(100),
|
||||
locations JSONB DEFAULT '[]',
|
||||
employee_count INT,
|
||||
dpo_name VARCHAR(200),
|
||||
dpo_contact VARCHAR(200),
|
||||
vvt_version VARCHAR(20) DEFAULT '1.0',
|
||||
last_review_date DATE,
|
||||
next_review_date DATE,
|
||||
review_interval VARCHAR(20) DEFAULT 'annual',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS compliance_vvt_activities (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
vvt_id VARCHAR(50) UNIQUE NOT NULL,
|
||||
name VARCHAR(300) NOT NULL,
|
||||
description TEXT,
|
||||
purposes JSONB DEFAULT '[]',
|
||||
legal_bases JSONB DEFAULT '[]',
|
||||
data_subject_categories JSONB DEFAULT '[]',
|
||||
personal_data_categories JSONB DEFAULT '[]',
|
||||
recipient_categories JSONB DEFAULT '[]',
|
||||
third_country_transfers JSONB DEFAULT '[]',
|
||||
retention_period JSONB DEFAULT '{}',
|
||||
tom_description TEXT,
|
||||
business_function VARCHAR(50),
|
||||
systems JSONB DEFAULT '[]',
|
||||
deployment_model VARCHAR(20),
|
||||
data_sources JSONB DEFAULT '[]',
|
||||
data_flows JSONB DEFAULT '[]',
|
||||
protection_level VARCHAR(10) DEFAULT 'MEDIUM',
|
||||
dpia_required BOOLEAN DEFAULT FALSE,
|
||||
structured_toms JSONB DEFAULT '{}',
|
||||
status VARCHAR(20) DEFAULT 'DRAFT',
|
||||
responsible VARCHAR(200),
|
||||
owner VARCHAR(200),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_vvt_activities_status ON compliance_vvt_activities(status);
|
||||
CREATE INDEX IF NOT EXISTS idx_vvt_activities_business_function ON compliance_vvt_activities(business_function);
|
||||
CREATE INDEX IF NOT EXISTS idx_vvt_activities_vvt_id ON compliance_vvt_activities(vvt_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS compliance_vvt_audit_log (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
action VARCHAR(20) NOT NULL,
|
||||
entity_type VARCHAR(50) NOT NULL,
|
||||
entity_id UUID,
|
||||
changed_by VARCHAR(200),
|
||||
old_values JSONB,
|
||||
new_values JSONB,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_vvt_audit_created ON compliance_vvt_audit_log(created_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_vvt_audit_entity ON compliance_vvt_audit_log(entity_type, entity_id);
|
||||
@@ -24,6 +24,7 @@ anthropic==0.75.0
|
||||
|
||||
# PDF Generation (GDPR export, audit reports)
|
||||
weasyprint==66.0
|
||||
reportlab==4.2.5
|
||||
Jinja2==3.1.6
|
||||
|
||||
# Document Processing (Word import for consent admin)
|
||||
|
||||
222
backend-compliance/tests/test_vvt_routes.py
Normal file
222
backend-compliance/tests/test_vvt_routes.py
Normal file
@@ -0,0 +1,222 @@
|
||||
"""Tests for VVT routes and schemas (vvt_routes.py, vvt_models.py)."""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from datetime import datetime, date
|
||||
import uuid
|
||||
|
||||
from compliance.api.schemas import (
|
||||
VVTActivityCreate,
|
||||
VVTActivityUpdate,
|
||||
VVTOrganizationUpdate,
|
||||
VVTStatsResponse,
|
||||
)
|
||||
from compliance.api.vvt_routes import _activity_to_response, _log_audit
|
||||
from compliance.db.vvt_models import VVTActivityDB, VVTOrganizationDB, VVTAuditLogDB
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Schema Tests
|
||||
# =============================================================================
|
||||
|
||||
class TestVVTActivityCreate:
|
||||
def test_default_values(self):
|
||||
req = VVTActivityCreate(vvt_id="VVT-001", name="Test Verarbeitung")
|
||||
assert req.vvt_id == "VVT-001"
|
||||
assert req.name == "Test Verarbeitung"
|
||||
assert req.status == "DRAFT"
|
||||
assert req.protection_level == "MEDIUM"
|
||||
assert req.dpia_required is False
|
||||
assert req.purposes == []
|
||||
assert req.legal_bases == []
|
||||
|
||||
def test_full_values(self):
|
||||
req = VVTActivityCreate(
|
||||
vvt_id="VVT-002",
|
||||
name="Gehaltsabrechnung",
|
||||
description="Verarbeitung von Gehaltsabrechnungsdaten",
|
||||
purposes=["Vertragserfuellung"],
|
||||
legal_bases=["Art. 6 Abs. 1b DSGVO"],
|
||||
data_subject_categories=["Mitarbeiter"],
|
||||
personal_data_categories=["Bankdaten", "Steuer-ID"],
|
||||
status="APPROVED",
|
||||
dpia_required=False,
|
||||
)
|
||||
assert req.vvt_id == "VVT-002"
|
||||
assert req.status == "APPROVED"
|
||||
assert len(req.purposes) == 1
|
||||
assert len(req.personal_data_categories) == 2
|
||||
|
||||
def test_serialization(self):
|
||||
req = VVTActivityCreate(vvt_id="VVT-003", name="Test")
|
||||
data = req.model_dump()
|
||||
assert data["vvt_id"] == "VVT-003"
|
||||
assert isinstance(data["purposes"], list)
|
||||
assert isinstance(data["retention_period"], dict)
|
||||
|
||||
|
||||
class TestVVTActivityUpdate:
|
||||
def test_partial_update(self):
|
||||
req = VVTActivityUpdate(status="APPROVED")
|
||||
data = req.model_dump(exclude_none=True)
|
||||
assert data == {"status": "APPROVED"}
|
||||
|
||||
def test_empty_update(self):
|
||||
req = VVTActivityUpdate()
|
||||
data = req.model_dump(exclude_none=True)
|
||||
assert data == {}
|
||||
|
||||
def test_multi_field_update(self):
|
||||
req = VVTActivityUpdate(
|
||||
name="Updated Name",
|
||||
dpia_required=True,
|
||||
protection_level="HIGH",
|
||||
)
|
||||
data = req.model_dump(exclude_none=True)
|
||||
assert data["name"] == "Updated Name"
|
||||
assert data["dpia_required"] is True
|
||||
assert data["protection_level"] == "HIGH"
|
||||
|
||||
|
||||
class TestVVTOrganizationUpdate:
|
||||
def test_defaults(self):
|
||||
req = VVTOrganizationUpdate()
|
||||
data = req.model_dump(exclude_none=True)
|
||||
assert data == {}
|
||||
|
||||
def test_partial_update(self):
|
||||
req = VVTOrganizationUpdate(
|
||||
organization_name="BreakPilot GmbH",
|
||||
dpo_name="Max Mustermann",
|
||||
)
|
||||
data = req.model_dump(exclude_none=True)
|
||||
assert data["organization_name"] == "BreakPilot GmbH"
|
||||
assert data["dpo_name"] == "Max Mustermann"
|
||||
|
||||
|
||||
class TestVVTStatsResponse:
|
||||
def test_stats_response(self):
|
||||
stats = VVTStatsResponse(
|
||||
total=5,
|
||||
by_status={"DRAFT": 3, "APPROVED": 2},
|
||||
by_business_function={"HR": 2, "IT": 3},
|
||||
dpia_required_count=1,
|
||||
third_country_count=0,
|
||||
draft_count=3,
|
||||
approved_count=2,
|
||||
)
|
||||
assert stats.total == 5
|
||||
assert stats.by_status["DRAFT"] == 3
|
||||
assert stats.dpia_required_count == 1
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DB Model Tests
|
||||
# =============================================================================
|
||||
|
||||
class TestVVTModels:
|
||||
def test_activity_defaults(self):
|
||||
act = VVTActivityDB()
|
||||
assert act.status is None or act.status == 'DRAFT'
|
||||
assert act.dpia_required is False or act.dpia_required is None
|
||||
|
||||
def test_activity_repr(self):
|
||||
act = VVTActivityDB()
|
||||
act.vvt_id = "VVT-001"
|
||||
act.name = "Test"
|
||||
assert "VVT-001" in repr(act)
|
||||
|
||||
def test_organization_repr(self):
|
||||
org = VVTOrganizationDB()
|
||||
org.organization_name = "Test GmbH"
|
||||
assert "Test GmbH" in repr(org)
|
||||
|
||||
def test_audit_log_repr(self):
|
||||
log = VVTAuditLogDB()
|
||||
log.action = "CREATE"
|
||||
log.entity_type = "activity"
|
||||
assert "CREATE" in repr(log)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Helper Function Tests
|
||||
# =============================================================================
|
||||
|
||||
class TestActivityToResponse:
|
||||
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 = kwargs.get("description", None)
|
||||
act.purposes = kwargs.get("purposes", [])
|
||||
act.legal_bases = kwargs.get("legal_bases", [])
|
||||
act.data_subject_categories = kwargs.get("data_subject_categories", [])
|
||||
act.personal_data_categories = kwargs.get("personal_data_categories", [])
|
||||
act.recipient_categories = kwargs.get("recipient_categories", [])
|
||||
act.third_country_transfers = kwargs.get("third_country_transfers", [])
|
||||
act.retention_period = kwargs.get("retention_period", {})
|
||||
act.tom_description = kwargs.get("tom_description", None)
|
||||
act.business_function = kwargs.get("business_function", None)
|
||||
act.systems = kwargs.get("systems", [])
|
||||
act.deployment_model = kwargs.get("deployment_model", None)
|
||||
act.data_sources = kwargs.get("data_sources", [])
|
||||
act.data_flows = kwargs.get("data_flows", [])
|
||||
act.protection_level = kwargs.get("protection_level", "MEDIUM")
|
||||
act.dpia_required = kwargs.get("dpia_required", False)
|
||||
act.structured_toms = kwargs.get("structured_toms", {})
|
||||
act.status = kwargs.get("status", "DRAFT")
|
||||
act.responsible = kwargs.get("responsible", None)
|
||||
act.owner = kwargs.get("owner", None)
|
||||
act.created_at = datetime.utcnow()
|
||||
act.updated_at = None
|
||||
return act
|
||||
|
||||
def test_basic_conversion(self):
|
||||
act = self._make_activity(vvt_id="VVT-001", name="Kundendaten")
|
||||
response = _activity_to_response(act)
|
||||
assert response.vvt_id == "VVT-001"
|
||||
assert response.name == "Kundendaten"
|
||||
assert response.status == "DRAFT"
|
||||
assert response.protection_level == "MEDIUM"
|
||||
|
||||
def test_null_lists_become_empty(self):
|
||||
act = self._make_activity()
|
||||
act.purposes = None
|
||||
act.legal_bases = None
|
||||
response = _activity_to_response(act)
|
||||
assert response.purposes == []
|
||||
assert response.legal_bases == []
|
||||
|
||||
def test_null_dicts_become_empty(self):
|
||||
act = self._make_activity()
|
||||
act.retention_period = None
|
||||
act.structured_toms = None
|
||||
response = _activity_to_response(act)
|
||||
assert response.retention_period == {}
|
||||
assert response.structured_toms == {}
|
||||
|
||||
|
||||
class TestLogAudit:
|
||||
def test_creates_audit_entry(self):
|
||||
mock_db = MagicMock()
|
||||
act_id = uuid.uuid4()
|
||||
_log_audit(
|
||||
db=mock_db,
|
||||
action="CREATE",
|
||||
entity_type="activity",
|
||||
entity_id=act_id,
|
||||
changed_by="test_user",
|
||||
new_values={"name": "Test"},
|
||||
)
|
||||
mock_db.add.assert_called_once()
|
||||
added = mock_db.add.call_args[0][0]
|
||||
assert added.action == "CREATE"
|
||||
assert added.entity_type == "activity"
|
||||
assert added.entity_id == act_id
|
||||
|
||||
def test_defaults_changed_by(self):
|
||||
mock_db = MagicMock()
|
||||
_log_audit(mock_db, "DELETE", "activity")
|
||||
added = mock_db.add.call_args[0][0]
|
||||
assert added.changed_by == "system"
|
||||
Reference in New Issue
Block a user