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

feat(sdk): vendor-compliance cross-module integration — VVT, obligations, TOM, loeschfristen

Browse Source 
Integrate the vendor-compliance module with four DSGVO modules to eliminate
data silos and resolve the VVT processor tab's ephemeral state problem.

- Reposition vendor-compliance sidebar from seq 4200 to 2500 (after VVT)
- VVT: replace ephemeral ProcessorRecord state with Vendor-API fetch (read-only)
- Obligations: add linked_vendor_ids (JSONB) + compliance check #12 MISSING_VENDOR_LINK
- TOM: add vendor TOM-controls cross-reference table in overview tab
- Loeschfristen: add linked_vendor_ids (JSONB) + vendor picker + document section
- Migrations: 069_obligations_vendor_link.sql, 070_loeschfristen_vendor_link.sql
- Tests: 12 new backend tests (125 total pass)
- Docs: update obligations.md + vendors.md with cross-module integration

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-19 13:59:43 +01:00
parent 4b1eede45b
commit c3afa628ed
19 changed files with 2852 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend-compliance/compliance/api/loeschfristen_routes.py
View File

@@ -56,6 +56,7 @@ class LoeschfristCreate(BaseModel):
responsible_person: Optional[str] = None
release_process: Optional[str] = None
linked_vvt_activity_ids: Optional[List[Any]] = None
linked_vendor_ids: Optional[List[Any]] = None
status: str = "DRAFT"
last_review_date: Optional[datetime] = None
next_review_date: Optional[datetime] = None
@@ -86,6 +87,7 @@ class LoeschfristUpdate(BaseModel):
responsible_person: Optional[str] = None
release_process: Optional[str] = None
linked_vvt_activity_ids: Optional[List[Any]] = None
linked_vendor_ids: Optional[List[Any]] = None
status: Optional[str] = None
last_review_date: Optional[datetime] = None
next_review_date: Optional[datetime] = None
@@ -100,7 +102,7 @@ class StatusUpdate(BaseModel):
# JSONB fields that need CAST
JSONB_FIELDS = {
"affected_groups", "data_categories", "legal_holds",
"storage_locations", "linked_vvt_activity_ids", "tags"
"storage_locations", "linked_vvt_activity_ids", "linked_vendor_ids", "tags"
}

11
backend-compliance/compliance/api/obligation_routes.py
View File

@@ -42,6 +42,7 @@ class ObligationCreate(BaseModel):
priority: str = "medium"
responsible: Optional[str] = None
linked_systems: Optional[List[str]] = None
linked_vendor_ids: Optional[List[str]] = None
assessment_id: Optional[str] = None
rule_code: Optional[str] = None
notes: Optional[str] = None
@@ -57,6 +58,7 @@ class ObligationUpdate(BaseModel):
priority: Optional[str] = None
responsible: Optional[str] = None
linked_systems: Optional[List[str]] = None
linked_vendor_ids: Optional[List[str]] = None
notes: Optional[str] = None
@@ -173,14 +175,15 @@ async def create_obligation(
import json
linked_systems = json.dumps(payload.linked_systems or [])
linked_vendor_ids = json.dumps(payload.linked_vendor_ids or [])
row = db.execute(text("""
INSERT INTO compliance_obligations
(tenant_id, title, description, source, source_article, deadline,
status, priority, responsible, linked_systems, assessment_id, rule_code, notes)
status, priority, responsible, linked_systems, linked_vendor_ids, assessment_id, rule_code, notes)
VALUES
(:tenant_id, :title, :description, :source, :source_article, :deadline,
:status, :priority, :responsible, CAST(:linked_systems AS jsonb), :assessment_id, :rule_code, :notes)
:status, :priority, :responsible, CAST(:linked_systems AS jsonb), CAST(:linked_vendor_ids AS jsonb), :assessment_id, :rule_code, :notes)
RETURNING *
"""), {
"tenant_id": tenant_id,
@@ -193,6 +196,7 @@ async def create_obligation(
"priority": payload.priority,
"responsible": payload.responsible,
"linked_systems": linked_systems,
"linked_vendor_ids": linked_vendor_ids,
"assessment_id": payload.assessment_id,
"rule_code": payload.rule_code,
"notes": payload.notes,
@@ -235,6 +239,9 @@ async def update_obligation(
if field == "linked_systems":
updates["linked_systems"] = json.dumps(value or [])
set_clauses.append("linked_systems = CAST(:linked_systems AS jsonb)")
elif field == "linked_vendor_ids":
updates["linked_vendor_ids"] = json.dumps(value or [])
set_clauses.append("linked_vendor_ids = CAST(:linked_vendor_ids AS jsonb)")
else:
updates[field] = value
set_clauses.append(f"{field} = :{field}")

3
backend-compliance/migrations/069_obligations_vendor_link.sql Normal file
View File

@@ -0,0 +1,3 @@
-- Obligations: Vendor-Verknuepfung fuer Art. 28 DSGVO
ALTER TABLE compliance_obligations
ADD COLUMN IF NOT EXISTS linked_vendor_ids JSONB DEFAULT '[]'::jsonb;

3
backend-compliance/migrations/070_loeschfristen_vendor_link.sql Normal file
View File

@@ -0,0 +1,3 @@
-- Loeschfristen: Vendor-Verknuepfung
ALTER TABLE compliance_loeschfristen
ADD COLUMN IF NOT EXISTS linked_vendor_ids JSONB DEFAULT '[]'::jsonb;

68
backend-compliance/tests/test_loeschfristen_routes.py
View File

@@ -56,6 +56,7 @@ def make_policy_row(overrides=None):
"responsible_person": None,
"release_process": None,
"linked_vvt_activity_ids": [],
"linked_vendor_ids": [],
"status": "DRAFT",
"last_review_date": None,
"next_review_date": None,
@@ -132,7 +133,7 @@ class TestRowToDict:
class TestJsonbFields:
def test_jsonb_fields_set(self):
expected = {"affected_groups", "data_categories", "legal_holds",
"storage_locations", "linked_vvt_activity_ids", "tags"}
"storage_locations", "linked_vvt_activity_ids", "linked_vendor_ids", "tags"}
assert JSONB_FIELDS == expected
@@ -618,3 +619,68 @@ class TestDeleteLoeschfrist:
)
call_params = mock_db.execute.call_args[0][1]
assert call_params["tenant_id"] == "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
# =============================================================================
# Linked Vendor IDs (Vendor-Compliance Integration)
# =============================================================================
class TestLinkedVendorIds:
def test_create_with_linked_vendor_ids(self, mock_db):
row = make_policy_row({"linked_vendor_ids": ["vendor-1"]})
mock_db.execute.return_value.fetchone.return_value = row
resp = client.post("/loeschfristen", json={
"data_object_name": "Vendor-Daten",
"linked_vendor_ids": ["vendor-1"],
})
assert resp.status_code == 201
import json
call_params = mock_db.execute.call_args[0][1]
assert json.loads(call_params["linked_vendor_ids"]) == ["vendor-1"]
def test_create_without_linked_vendor_ids_defaults_empty(self, mock_db):
row = make_policy_row()
mock_db.execute.return_value.fetchone.return_value = row
resp = client.post("/loeschfristen", json={
"data_object_name": "Ohne Vendor",
})
assert resp.status_code == 201
import json
call_params = mock_db.execute.call_args[0][1]
assert json.loads(call_params["linked_vendor_ids"]) == []
def test_update_linked_vendor_ids(self, mock_db):
updated_row = make_policy_row({"linked_vendor_ids": ["v1"]})
mock_db.execute.return_value.fetchone.return_value = updated_row
resp = client.put(f"/loeschfristen/{POLICY_ID}", json={
"linked_vendor_ids": ["v1"],
})
assert resp.status_code == 200
import json
call_params = mock_db.execute.call_args[0][1]
assert json.loads(call_params["linked_vendor_ids"]) == ["v1"]
def test_update_clears_linked_vendor_ids(self, mock_db):
updated_row = make_policy_row({"linked_vendor_ids": []})
mock_db.execute.return_value.fetchone.return_value = updated_row
resp = client.put(f"/loeschfristen/{POLICY_ID}", json={
"linked_vendor_ids": [],
})
assert resp.status_code == 200
import json
call_params = mock_db.execute.call_args[0][1]
assert json.loads(call_params["linked_vendor_ids"]) == []
def test_schema_includes_linked_vendor_ids(self):
create_obj = LoeschfristCreate(
data_object_name="Test",
linked_vendor_ids=["vendor-a", "vendor-b"],
)
assert create_obj.linked_vendor_ids == ["vendor-a", "vendor-b"]
update_obj = LoeschfristUpdate(linked_vendor_ids=["vendor-c"])
data = update_obj.model_dump(exclude_unset=True)
assert data["linked_vendor_ids"] == ["vendor-c"]
def test_jsonb_fields_contains_linked_vendor_ids(self):
assert "linked_vendor_ids" in JSONB_FIELDS

58
backend-compliance/tests/test_obligation_routes.py
View File

@@ -52,6 +52,7 @@ def _make_obligation_row(overrides=None):
"priority": "medium",
"responsible": None,
"linked_systems": [],
"linked_vendor_ids": [],
"assessment_id": None,
"rule_code": None,
"notes": None,
@@ -607,3 +608,60 @@ class TestObligationSearchRoute:
resp = client.get("/obligations?source=AI Act")
assert resp.status_code == 200
assert resp.json()["obligations"][0]["source"] == "AI Act"
# =============================================================================
# Linked Vendor IDs Tests (Art. 28 DSGVO)
# =============================================================================
class TestLinkedVendorIds:
def test_create_with_linked_vendor_ids(self, client, mock_db):
row = _make_obligation_row({"linked_vendor_ids": ["vendor-1", "vendor-2"]})
mock_db.execute.return_value = MagicMock(fetchone=MagicMock(return_value=row))
resp = client.post("/obligations", json={
"title": "Vendor-Prüfung Art. 28",
"linked_vendor_ids": ["vendor-1", "vendor-2"],
})
assert resp.status_code == 201
assert resp.json()["linked_vendor_ids"] == ["vendor-1", "vendor-2"]
def test_create_without_linked_vendor_ids_defaults_empty(self, client, mock_db):
row = _make_obligation_row()
mock_db.execute.return_value = MagicMock(fetchone=MagicMock(return_value=row))
resp = client.post("/obligations", json={"title": "Ohne Vendor"})
assert resp.status_code == 201
# Schema allows it — linked_vendor_ids defaults to None in the schema
schema = ObligationCreate(title="Ohne Vendor")
assert schema.linked_vendor_ids is None
def test_update_linked_vendor_ids(self, client, mock_db):
updated = _make_obligation_row({"linked_vendor_ids": ["v1"]})
mock_db.execute.return_value = MagicMock(fetchone=MagicMock(return_value=updated))
resp = client.put(f"/obligations/{OBLIGATION_ID}", json={
"linked_vendor_ids": ["v1"],
})
assert resp.status_code == 200
assert resp.json()["linked_vendor_ids"] == ["v1"]
def test_update_clears_linked_vendor_ids(self, client, mock_db):
updated = _make_obligation_row({"linked_vendor_ids": []})
mock_db.execute.return_value = MagicMock(fetchone=MagicMock(return_value=updated))
resp = client.put(f"/obligations/{OBLIGATION_ID}", json={
"linked_vendor_ids": [],
})
assert resp.status_code == 200
assert resp.json()["linked_vendor_ids"] == []
def test_schema_create_includes_linked_vendor_ids(self):
schema = ObligationCreate(
title="Test Vendor Link",
linked_vendor_ids=["a", "b"],
)
assert schema.linked_vendor_ids == ["a", "b"]
data = schema.model_dump()
assert data["linked_vendor_ids"] == ["a", "b"]
def test_schema_update_includes_linked_vendor_ids(self):
schema = ObligationUpdate(linked_vendor_ids=["a"])
data = schema.model_dump(exclude_unset=True)
assert data["linked_vendor_ids"] == ["a"]