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

feat: control_parent_links population + traceability API + frontend

Browse Source 
- _write_atomic_control() now uses RETURNING id and inserts into
  control_parent_links (M:N) with source_regulation, source_article,
  and obligation_candidate_id parsed from parent's source_citation
- New _parse_citation() helper for JSONB source_citation extraction
- New GET /controls/{id}/traceability endpoint returning full chain:
  parent links with obligations, child controls, source_count
- Backend: control_type filter (atomic/rich) for controls + count
- Frontend: Rechtsgrundlagen section in ControlDetail showing all
  parent links per source regulation with obligation text + strength
- Frontend: Atomic/Rich filter dropdown in Control Library list
- Frontend: GenerationStrategyBadge recognizes 'pass0b' strategy
- Tests: 3 new tests for parent_link creation + citation parsing,
  existing batch test mock updated for RETURNING clause

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-23 08:14:29 +01:00
parent 0027f78fc5
commit ac6134ce6d
7 changed files with 511 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
backend-compliance/tests/test_decomposition_pass.py
View File

@@ -1118,10 +1118,15 @@ class TestDecompositionPassAnthropicBatch:
call_count[0] += 1
if call_count[0] == 1:
return mock_rows # SELECT candidates
# _next_atomic_seq calls (every 3rd after first: 2, 5, 8, ...)
if call_count[0] in (2, 5):
# _next_atomic_seq calls: call 2 (control 1), call 6 (control 2)
if call_count[0] in (2, 6):
return mock_seq
return MagicMock() # INSERT/UPDATE
# INSERT RETURNING calls: call 3 (control 1), call 7 (control 2)
if call_count[0] in (3, 7):
mock_insert = MagicMock()
mock_insert.fetchone.return_value = (f"new-uuid-{call_count[0]}",)
return mock_insert
return MagicMock() # parent_links INSERT / UPDATE
mock_db.execute.side_effect = side_effect
batched_response = json.dumps({
@@ -1608,12 +1613,16 @@ class TestPass0bWithEnrichment:
mock_db = MagicMock()
mock_seq = MagicMock()
mock_seq.fetchone.return_value = (0,)
mock_insert = MagicMock()
mock_insert.fetchone.return_value = ("new-uuid-1",)
call_count = [0]
def side_effect(*args, **kwargs):
call_count[0] += 1
if call_count[0] == 1:
return mock_seq # _next_atomic_seq
if call_count[0] == 2:
return mock_insert # INSERT RETURNING id
return MagicMock()
mock_db.execute.side_effect = side_effect
@@ -1623,12 +1632,20 @@ class TestPass0bWithEnrichment:
decomp._process_pass0b_control(obl, parsed, stats)
)
# _write_atomic_control is call #2: db.execute(text(...), {params})
# _write_atomic_control INSERT is call #2: db.execute(text(...), {params})
insert_call = mock_db.execute.call_args_list[1]
# positional args: (text_obj, params_dict)
insert_params = insert_call[0][1]
assert insert_params["severity"] == "medium"
# parent_link INSERT is call #3
link_call = mock_db.execute.call_args_list[2]
link_query = str(link_call[0][0])
assert "control_parent_links" in link_query
link_params = link_call[0][1]
assert link_params["cu"] == "new-uuid-1"
assert link_params["pu"] == "p-uuid"
def test_test_obligation_gets_testing_category(self):
"""Test obligations should get category='testing'."""
obl = {
@@ -1664,12 +1681,16 @@ class TestPass0bWithEnrichment:
mock_db = MagicMock()
mock_seq = MagicMock()
mock_seq.fetchone.return_value = (0,)
mock_insert = MagicMock()
mock_insert.fetchone.return_value = ("new-uuid-2",)
call_count = [0]
def side_effect(*args, **kwargs):
call_count[0] += 1
if call_count[0] == 1:
return mock_seq
if call_count[0] == 2:
return mock_insert # INSERT RETURNING id
return MagicMock()
mock_db.execute.side_effect = side_effect
@@ -1679,7 +1700,99 @@ class TestPass0bWithEnrichment:
decomp._process_pass0b_control(obl, parsed, stats)
)
# _write_atomic_control is call #2: db.execute(text(...), {params})
# _write_atomic_control INSERT is call #2: db.execute(text(...), {params})
insert_call = mock_db.execute.call_args_list[1]
insert_params = insert_call[0][1]
assert insert_params["category"] == "testing"
def test_parent_link_created_with_source_citation(self):
"""_write_atomic_control inserts a row into control_parent_links
with source_regulation and source_article parsed from parent_citation."""
import json as _json
obl = {
"oc_id": "oc-link-1",
"candidate_id": "OC-DSGVO-01",
"parent_uuid": "p-uuid-dsgvo",
"obligation_text": "Daten minimieren",
"action": "minimieren",
"object": "personenbezogene Daten",
"is_test": False,
"is_reporting": False,
"parent_title": "Datenminimierung",
"parent_category": "privacy",
"parent_citation": _json.dumps({
"source": "DSGVO",
"article": "Art. 5 Abs. 1 lit. c",
"paragraph": "",
}),
"parent_severity": "high",
"parent_control_id": "PRIV-001",
"source_ref": "DSGVO Art. 5 Abs. 1 lit. c",
"trigger_type": "continuous",
"is_implementation_specific": False,
}
parsed = {
"title": "Personenbezogene Daten minimieren",
"objective": "Nur erforderliche Daten erheben",
"requirements": ["Datenminimierung"],
"test_procedure": ["Audit"],
"evidence": ["Protokoll"],
"severity": "high",
"category": "privacy",
}
stats = {"controls_created": 0, "candidates_processed": 0,
"llm_failures": 0, "dedup_linked": 0, "dedup_review": 0}
mock_db = MagicMock()
mock_seq = MagicMock()
mock_seq.fetchone.return_value = (0,)
mock_insert = MagicMock()
mock_insert.fetchone.return_value = ("new-uuid-dsgvo",)
call_count = [0]
def side_effect(*args, **kwargs):
call_count[0] += 1
if call_count[0] == 1:
return mock_seq
if call_count[0] == 2:
return mock_insert
return MagicMock()
mock_db.execute.side_effect = side_effect
import asyncio
decomp = DecompositionPass(db=mock_db)
asyncio.get_event_loop().run_until_complete(
decomp._process_pass0b_control(obl, parsed, stats)
)
# Call #3 is the parent_link INSERT
link_call = mock_db.execute.call_args_list[2]
link_query = str(link_call[0][0])
assert "control_parent_links" in link_query
link_params = link_call[0][1]
assert link_params["cu"] == "new-uuid-dsgvo"
assert link_params["pu"] == "p-uuid-dsgvo"
assert link_params["sr"] == "DSGVO"
assert link_params["sa"] == "Art. 5 Abs. 1 lit. c"
assert link_params["oci"] == "oc-link-1"
def test_parse_citation_handles_formats(self):
"""_parse_citation handles JSON string, dict, empty, and invalid."""
import json as _json
from compliance.services.decomposition_pass import _parse_citation
# JSON string
result = _parse_citation(_json.dumps({"source": "NIS2", "article": "Art. 21"}))
assert result["source"] == "NIS2"
assert result["article"] == "Art. 21"
# Already a dict
result = _parse_citation({"source": "DSGVO", "article": "Art. 5"})
assert result["source"] == "DSGVO"
# Empty / None
assert _parse_citation("") == {}
assert _parse_citation(None) == {}
# Invalid JSON
assert _parse_citation("not json") == {}