feat: Control-Detail Provenance + Atomare Controls Seite
All checks were successful
CI/CD / go-lint (push) Has been skipped
CI/CD / python-lint (push) Has been skipped
CI/CD / nodejs-lint (push) Has been skipped
CI/CD / test-go-ai-compliance (push) Successful in 41s
CI/CD / test-python-backend-compliance (push) Successful in 40s
CI/CD / test-python-document-crawler (push) Successful in 23s
CI/CD / test-python-dsms-gateway (push) Successful in 18s
CI/CD / validate-canonical-controls (push) Successful in 11s
CI/CD / Deploy (push) Successful in 4s
All checks were successful
CI/CD / go-lint (push) Has been skipped
CI/CD / python-lint (push) Has been skipped
CI/CD / nodejs-lint (push) Has been skipped
CI/CD / test-go-ai-compliance (push) Successful in 41s
CI/CD / test-python-backend-compliance (push) Successful in 40s
CI/CD / test-python-document-crawler (push) Successful in 23s
CI/CD / test-python-dsms-gateway (push) Successful in 18s
CI/CD / validate-canonical-controls (push) Successful in 11s
CI/CD / Deploy (push) Successful in 4s
Backend: provenance endpoint (obligations, doc refs, merged duplicates, regulations summary) + atomic-stats aggregation endpoint. Frontend: ControlDetail mit Provenance-Sektionen, klickbare Navigation, neue /sdk/atomic-controls Seite mit Stats-Bar und gefilterer Liste. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
277
backend-compliance/tests/test_provenance_endpoint.py
Normal file
277
backend-compliance/tests/test_provenance_endpoint.py
Normal file
@@ -0,0 +1,277 @@
|
||||
"""Tests for provenance and atomic-stats endpoints.
|
||||
|
||||
Covers:
|
||||
- GET /v1/canonical/controls/{control_id}/provenance
|
||||
- GET /v1/canonical/controls/atomic-stats
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from datetime import datetime
|
||||
|
||||
from compliance.api.canonical_control_routes import (
|
||||
get_control_provenance,
|
||||
atomic_stats,
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# HELPERS
|
||||
# =============================================================================
|
||||
|
||||
def _mock_row(**kwargs):
|
||||
"""Create a mock DB row with attribute access."""
|
||||
obj = MagicMock()
|
||||
for k, v in kwargs.items():
|
||||
setattr(obj, k, v)
|
||||
return obj
|
||||
|
||||
|
||||
def _mock_db_execute(return_values):
|
||||
"""Return a mock that cycles through return values for sequential .execute() calls."""
|
||||
mock_db = MagicMock()
|
||||
results = iter(return_values)
|
||||
|
||||
def execute_side_effect(*args, **kwargs):
|
||||
result = next(results)
|
||||
mock_result = MagicMock()
|
||||
if isinstance(result, list):
|
||||
mock_result.fetchall.return_value = result
|
||||
mock_result.fetchone.return_value = result[0] if result else None
|
||||
elif isinstance(result, int):
|
||||
mock_result.scalar.return_value = result
|
||||
elif result is None:
|
||||
mock_result.fetchone.return_value = None
|
||||
mock_result.fetchall.return_value = []
|
||||
mock_result.scalar.return_value = 0
|
||||
else:
|
||||
mock_result.fetchone.return_value = result
|
||||
mock_result.fetchall.return_value = [result]
|
||||
return mock_result
|
||||
|
||||
mock_db.execute.side_effect = execute_side_effect
|
||||
return mock_db
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# PROVENANCE ENDPOINT
|
||||
# =============================================================================
|
||||
|
||||
class TestProvenanceEndpoint:
|
||||
"""Tests for GET /controls/{control_id}/provenance."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provenance_not_found(self):
|
||||
"""404 when control doesn't exist."""
|
||||
from fastapi import HTTPException
|
||||
|
||||
mock_db = _mock_db_execute([None])
|
||||
|
||||
with patch("compliance.api.canonical_control_routes.SessionLocal") as mock_session:
|
||||
mock_session.return_value.__enter__ = MagicMock(return_value=mock_db)
|
||||
mock_session.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await get_control_provenance("NONEXISTENT-999")
|
||||
assert exc_info.value.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provenance_atomic_control(self):
|
||||
"""Atomic control returns document_references, parent_links, merged_duplicates."""
|
||||
import uuid
|
||||
ctrl_id = uuid.uuid4()
|
||||
|
||||
ctrl_row = _mock_row(
|
||||
id=ctrl_id,
|
||||
control_id="SEC-042",
|
||||
title="Test Atomic Control",
|
||||
parent_control_uuid=None,
|
||||
decomposition_method="pass0b",
|
||||
source_citation=None,
|
||||
)
|
||||
|
||||
parent_link = _mock_row(
|
||||
parent_control_uuid=uuid.uuid4(),
|
||||
parent_control_id="DATA-005",
|
||||
parent_title="Parent Control",
|
||||
link_type="decomposition",
|
||||
confidence=0.95,
|
||||
source_regulation="DSGVO",
|
||||
source_article="Art. 32",
|
||||
parent_citation=None,
|
||||
obligation_text="Must encrypt",
|
||||
action="encrypt",
|
||||
object="personal data",
|
||||
normative_strength="must",
|
||||
obligation_candidate_id=None,
|
||||
)
|
||||
|
||||
child_row = _mock_row(
|
||||
control_id="SEC-042a",
|
||||
title="Child",
|
||||
category="encryption",
|
||||
severity="high",
|
||||
decomposition_method="pass0b",
|
||||
)
|
||||
|
||||
obligation_row = _mock_row(
|
||||
candidate_id="OBL-SEC-042-001",
|
||||
obligation_text="Test obligation",
|
||||
action="encrypt",
|
||||
object="data at rest",
|
||||
normative_strength="must",
|
||||
release_state="composed",
|
||||
)
|
||||
|
||||
doc_ref = _mock_row(
|
||||
regulation_code="DSGVO",
|
||||
article="Art. 32",
|
||||
paragraph="Abs. 1 lit. a",
|
||||
extraction_method="llm_extracted",
|
||||
confidence=0.92,
|
||||
)
|
||||
|
||||
merged = _mock_row(
|
||||
control_id="SEC-099",
|
||||
title="Encryption at rest (NIS2)",
|
||||
source_regulation="NIS2",
|
||||
)
|
||||
|
||||
mock_db = _mock_db_execute([
|
||||
ctrl_row, # control lookup
|
||||
[parent_link], # parent_links
|
||||
[], # children
|
||||
[obligation_row], # obligations
|
||||
[doc_ref], # document_references
|
||||
[merged], # merged_duplicates
|
||||
])
|
||||
|
||||
with patch("compliance.api.canonical_control_routes.SessionLocal") as mock_session:
|
||||
mock_session.return_value.__enter__ = MagicMock(return_value=mock_db)
|
||||
mock_session.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
result = await get_control_provenance("SEC-042")
|
||||
|
||||
assert result["control_id"] == "SEC-042"
|
||||
assert result["is_atomic"] is True
|
||||
assert len(result["parent_links"]) == 1
|
||||
assert result["parent_links"][0]["parent_control_id"] == "DATA-005"
|
||||
assert result["obligation_count"] == 1
|
||||
assert len(result["document_references"]) == 1
|
||||
assert result["document_references"][0]["regulation_code"] == "DSGVO"
|
||||
assert len(result["merged_duplicates"]) == 1
|
||||
assert result["merged_duplicates"][0]["control_id"] == "SEC-099"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provenance_rich_control(self):
|
||||
"""Rich control returns obligations list and children."""
|
||||
import uuid
|
||||
ctrl_id = uuid.uuid4()
|
||||
|
||||
ctrl_row = _mock_row(
|
||||
id=ctrl_id,
|
||||
control_id="DATA-005",
|
||||
title="Rich Control",
|
||||
parent_control_uuid=None,
|
||||
decomposition_method=None,
|
||||
source_citation={"source": "DSGVO"},
|
||||
)
|
||||
|
||||
obligation_row = _mock_row(
|
||||
candidate_id="OBL-DATA-005-001",
|
||||
obligation_text="Encrypt personal data",
|
||||
action="encrypt",
|
||||
object="personal data",
|
||||
normative_strength="must",
|
||||
release_state="composed",
|
||||
)
|
||||
|
||||
child_row = _mock_row(
|
||||
control_id="SEC-042",
|
||||
title="Child Atomic",
|
||||
category="encryption",
|
||||
severity="high",
|
||||
decomposition_method="pass0b",
|
||||
)
|
||||
|
||||
mock_db = _mock_db_execute([
|
||||
ctrl_row, # control lookup
|
||||
[], # parent_links
|
||||
[child_row], # children
|
||||
[obligation_row], # obligations
|
||||
[], # document_references
|
||||
[], # merged_duplicates
|
||||
])
|
||||
|
||||
with patch("compliance.api.canonical_control_routes.SessionLocal") as mock_session:
|
||||
mock_session.return_value.__enter__ = MagicMock(return_value=mock_db)
|
||||
mock_session.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
result = await get_control_provenance("DATA-005")
|
||||
|
||||
assert result["control_id"] == "DATA-005"
|
||||
assert result["is_atomic"] is False
|
||||
assert result["obligation_count"] == 1
|
||||
assert result["obligations"][0]["candidate_id"] == "OBL-DATA-005-001"
|
||||
assert len(result["children"]) == 1
|
||||
assert result["children"][0]["control_id"] == "SEC-042"
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# ATOMIC STATS ENDPOINT
|
||||
# =============================================================================
|
||||
|
||||
class TestAtomicStatsEndpoint:
|
||||
"""Tests for GET /controls/atomic-stats."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_atomic_stats_response_shape(self):
|
||||
"""Stats endpoint returns expected aggregation fields."""
|
||||
mock_db = _mock_db_execute([
|
||||
18234, # total_active
|
||||
67000, # total_duplicate
|
||||
[ # by_domain
|
||||
_mock_row(**{"__getitem__": lambda s, i: ["SEC", 4200][i]}),
|
||||
],
|
||||
[ # by_regulation
|
||||
_mock_row(**{"__getitem__": lambda s, i: ["DSGVO", 1200][i]}),
|
||||
],
|
||||
2.3, # avg_coverage
|
||||
])
|
||||
|
||||
# Override __getitem__ for tuple-like access
|
||||
domain_row = MagicMock()
|
||||
domain_row.__getitem__ = lambda s, i: ["SEC", 4200][i]
|
||||
reg_row = MagicMock()
|
||||
reg_row.__getitem__ = lambda s, i: ["DSGVO", 1200][i]
|
||||
|
||||
mock_db2 = MagicMock()
|
||||
call_count = [0]
|
||||
responses = [18234, 67000, [domain_row], [reg_row], 2.3]
|
||||
|
||||
def execute_side(*args, **kwargs):
|
||||
idx = call_count[0]
|
||||
call_count[0] += 1
|
||||
r = MagicMock()
|
||||
val = responses[idx]
|
||||
if isinstance(val, list):
|
||||
r.fetchall.return_value = val
|
||||
else:
|
||||
r.scalar.return_value = val
|
||||
return r
|
||||
|
||||
mock_db2.execute.side_effect = execute_side
|
||||
|
||||
with patch("compliance.api.canonical_control_routes.SessionLocal") as mock_session:
|
||||
mock_session.return_value.__enter__ = MagicMock(return_value=mock_db2)
|
||||
mock_session.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
result = await atomic_stats()
|
||||
|
||||
assert result["total_active"] == 18234
|
||||
assert result["total_duplicate"] == 67000
|
||||
assert len(result["by_domain"]) == 1
|
||||
assert result["by_domain"][0]["domain"] == "SEC"
|
||||
assert len(result["by_regulation"]) == 1
|
||||
assert result["by_regulation"][0]["regulation"] == "DSGVO"
|
||||
assert result["avg_regulation_coverage"] == 2.3
|
||||
Reference in New Issue
Block a user