feat: add compliance modules 2-5 (dashboard, security templates, process manager, evidence collector)
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 32s
CI/CD / test-python-backend-compliance (push) Successful in 34s
CI/CD / test-python-document-crawler (push) Successful in 23s
CI/CD / test-python-dsms-gateway (push) Successful in 21s
CI/CD / validate-canonical-controls (push) Successful in 11s
CI/CD / Deploy (push) Successful in 2s
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 32s
CI/CD / test-python-backend-compliance (push) Successful in 34s
CI/CD / test-python-document-crawler (push) Successful in 23s
CI/CD / test-python-dsms-gateway (push) Successful in 21s
CI/CD / validate-canonical-controls (push) Successful in 11s
CI/CD / Deploy (push) Successful in 2s
Module 2: Extended Compliance Dashboard with roadmap, module-status, next-actions, snapshots, score-history Module 3: 7 German security document templates (IT-Sicherheitskonzept, Datenschutz, Backup, Logging, Incident-Response, Zugriff, Risikomanagement) Module 4: Compliance Process Manager with CRUD, complete/skip/seed, ~50 seed tasks, 3-tab UI Module 5: Evidence Collector Extended with automated checks, control-mapping, coverage report, 4-tab UI Also includes: canonical control library enhancements (verification method, categories, dedup), control generator improvements, RAG client extensions 52 tests pass, frontend builds clean. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
374
backend-compliance/tests/test_evidence_check_routes.py
Normal file
374
backend-compliance/tests/test_evidence_check_routes.py
Normal file
@@ -0,0 +1,374 @@
|
||||
"""Tests for Evidence Check routes (evidence_check_routes.py)."""
|
||||
|
||||
import json
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch, AsyncMock
|
||||
from datetime import datetime
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from compliance.api.evidence_check_routes import router, VALID_CHECK_TYPES
|
||||
from classroom_engine.database import get_db
|
||||
from compliance.api.tenant_utils import get_tenant_id
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# App setup with mocked DB dependency
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
app = FastAPI()
|
||||
app.include_router(router)
|
||||
|
||||
DEFAULT_TENANT_ID = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
|
||||
CHECK_ID = "ffffffff-0001-0001-0001-000000000001"
|
||||
RESULT_ID = "eeeeeeee-0001-0001-0001-000000000001"
|
||||
MAPPING_ID = "dddddddd-0001-0001-0001-000000000001"
|
||||
EVIDENCE_ID = "cccccccc-0001-0001-0001-000000000001"
|
||||
NOW = datetime(2026, 3, 14, 12, 0, 0)
|
||||
|
||||
|
||||
def override_get_tenant_id():
|
||||
return DEFAULT_TENANT_ID
|
||||
|
||||
|
||||
app.dependency_overrides[get_tenant_id] = override_get_tenant_id
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_check_row(overrides=None):
|
||||
"""Create a mock DB row for a check."""
|
||||
data = {
|
||||
"id": CHECK_ID,
|
||||
"tenant_id": DEFAULT_TENANT_ID,
|
||||
"project_id": None,
|
||||
"check_code": "TLS-SCAN-001",
|
||||
"title": "TLS-Scan Hauptwebseite",
|
||||
"description": "Prueft TLS",
|
||||
"check_type": "tls_scan",
|
||||
"target_url": "https://example.com",
|
||||
"target_config": {},
|
||||
"linked_control_ids": [],
|
||||
"frequency": "monthly",
|
||||
"last_run_at": None,
|
||||
"next_run_at": None,
|
||||
"is_active": True,
|
||||
"created_at": NOW,
|
||||
"updated_at": NOW,
|
||||
}
|
||||
if overrides:
|
||||
data.update(overrides)
|
||||
row = MagicMock()
|
||||
row._mapping = data
|
||||
row.__getitem__ = lambda self, i: list(data.values())[i]
|
||||
return row
|
||||
|
||||
|
||||
def _make_result_row(overrides=None):
|
||||
"""Create a mock DB row for a result."""
|
||||
data = {
|
||||
"id": RESULT_ID,
|
||||
"check_id": CHECK_ID,
|
||||
"tenant_id": DEFAULT_TENANT_ID,
|
||||
"run_status": "passed",
|
||||
"result_data": {"tls_version": "TLSv1.3"},
|
||||
"summary": "TLS TLSv1.3",
|
||||
"findings_count": 0,
|
||||
"critical_findings": 0,
|
||||
"evidence_id": None,
|
||||
"duration_ms": 150,
|
||||
"run_at": NOW,
|
||||
}
|
||||
if overrides:
|
||||
data.update(overrides)
|
||||
row = MagicMock()
|
||||
row._mapping = data
|
||||
row.__getitem__ = lambda self, i: list(data.values())[i]
|
||||
return row
|
||||
|
||||
|
||||
def _make_mapping_row(overrides=None):
|
||||
data = {
|
||||
"id": MAPPING_ID,
|
||||
"tenant_id": DEFAULT_TENANT_ID,
|
||||
"evidence_id": EVIDENCE_ID,
|
||||
"control_code": "TOM-001",
|
||||
"mapping_type": "supports",
|
||||
"verified_at": None,
|
||||
"verified_by": None,
|
||||
"notes": "Test mapping",
|
||||
"created_at": NOW,
|
||||
}
|
||||
if overrides:
|
||||
data.update(overrides)
|
||||
row = MagicMock()
|
||||
row._mapping = data
|
||||
row.__getitem__ = lambda self, i: list(data.values())[i]
|
||||
return row
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestListChecks:
|
||||
def test_list_checks(self):
|
||||
mock_db = MagicMock()
|
||||
# COUNT query
|
||||
count_row = MagicMock()
|
||||
count_row.__getitem__ = lambda self, i: 2
|
||||
# Data rows
|
||||
rows = [_make_check_row(), _make_check_row({"check_code": "TLS-SCAN-002"})]
|
||||
mock_db.execute.side_effect = [
|
||||
MagicMock(fetchone=MagicMock(return_value=count_row)),
|
||||
MagicMock(fetchall=MagicMock(return_value=rows)),
|
||||
]
|
||||
|
||||
app.dependency_overrides[get_db] = lambda: (yield mock_db).__next__() or mock_db
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
resp = client.get("/evidence-checks")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "checks" in data
|
||||
assert len(data["checks"]) == 2
|
||||
|
||||
|
||||
class TestCreateCheck:
|
||||
def test_create_check(self):
|
||||
mock_db = MagicMock()
|
||||
mock_db.execute.return_value.fetchone.return_value = _make_check_row()
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
resp = client.post("/evidence-checks", json={
|
||||
"check_code": "TLS-SCAN-001",
|
||||
"title": "TLS-Scan Hauptwebseite",
|
||||
"check_type": "tls_scan",
|
||||
"frequency": "monthly",
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["check_code"] == "TLS-SCAN-001"
|
||||
|
||||
def test_create_check_invalid_type(self):
|
||||
mock_db = MagicMock()
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
resp = client.post("/evidence-checks", json={
|
||||
"check_code": "INVALID-001",
|
||||
"title": "Invalid Check",
|
||||
"check_type": "invalid_type",
|
||||
})
|
||||
assert resp.status_code == 400
|
||||
assert "Ungueltiger check_type" in resp.json()["detail"]
|
||||
|
||||
|
||||
class TestGetSingleCheck:
|
||||
def test_get_single_check(self):
|
||||
mock_db = MagicMock()
|
||||
check_row = _make_check_row()
|
||||
result_rows = [_make_result_row()]
|
||||
mock_db.execute.side_effect = [
|
||||
MagicMock(fetchone=MagicMock(return_value=check_row)),
|
||||
MagicMock(fetchall=MagicMock(return_value=result_rows)),
|
||||
]
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
resp = client.get(f"/evidence-checks/{CHECK_ID}")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["check_code"] == "TLS-SCAN-001"
|
||||
assert "recent_results" in data
|
||||
assert len(data["recent_results"]) == 1
|
||||
|
||||
|
||||
class TestUpdateCheck:
|
||||
def test_update_check(self):
|
||||
mock_db = MagicMock()
|
||||
updated_row = _make_check_row({"title": "Updated Title"})
|
||||
mock_db.execute.return_value.fetchone.return_value = updated_row
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
resp = client.put(f"/evidence-checks/{CHECK_ID}", json={
|
||||
"title": "Updated Title",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["title"] == "Updated Title"
|
||||
|
||||
|
||||
class TestDeleteCheck:
|
||||
def test_delete_check(self):
|
||||
mock_db = MagicMock()
|
||||
mock_db.execute.return_value.rowcount = 1
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
resp = client.delete(f"/evidence-checks/{CHECK_ID}")
|
||||
assert resp.status_code == 204
|
||||
|
||||
|
||||
class TestRunCheckTLS:
|
||||
def test_run_check_tls(self):
|
||||
mock_db = MagicMock()
|
||||
check_row = _make_check_row()
|
||||
result_insert_row = _make_result_row({"run_status": "running"})
|
||||
result_update_row = _make_result_row({"run_status": "passed"})
|
||||
|
||||
mock_db.execute.side_effect = [
|
||||
# Load check
|
||||
MagicMock(fetchone=MagicMock(return_value=check_row)),
|
||||
# Insert running result
|
||||
MagicMock(fetchone=MagicMock(return_value=result_insert_row)),
|
||||
# Update result
|
||||
MagicMock(fetchone=MagicMock(return_value=result_update_row)),
|
||||
# Update check timestamps
|
||||
MagicMock(),
|
||||
]
|
||||
mock_db.commit = MagicMock()
|
||||
|
||||
tls_result = {
|
||||
"run_status": "passed",
|
||||
"result_data": {"tls_version": "TLSv1.3", "findings": []},
|
||||
"summary": "TLS TLSv1.3, Zertifikat gueltig",
|
||||
"findings_count": 0,
|
||||
"critical_findings": 0,
|
||||
"duration_ms": 100,
|
||||
}
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
with patch("compliance.api.evidence_check_routes._run_tls_scan", new_callable=AsyncMock, return_value=tls_result):
|
||||
resp = client.post(f"/evidence-checks/{CHECK_ID}/run")
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["run_status"] == "passed"
|
||||
|
||||
|
||||
class TestRunCheckHeader:
|
||||
def test_run_check_header(self):
|
||||
mock_db = MagicMock()
|
||||
check_row = _make_check_row({"check_type": "header_check"})
|
||||
result_insert_row = _make_result_row({"run_status": "running"})
|
||||
result_update_row = _make_result_row({"run_status": "warning"})
|
||||
|
||||
mock_db.execute.side_effect = [
|
||||
MagicMock(fetchone=MagicMock(return_value=check_row)),
|
||||
MagicMock(fetchone=MagicMock(return_value=result_insert_row)),
|
||||
MagicMock(fetchone=MagicMock(return_value=result_update_row)),
|
||||
MagicMock(),
|
||||
]
|
||||
mock_db.commit = MagicMock()
|
||||
|
||||
header_result = {
|
||||
"run_status": "warning",
|
||||
"result_data": {"missing_headers": ["Permissions-Policy"], "findings": []},
|
||||
"summary": "5/6 Security-Header vorhanden",
|
||||
"findings_count": 1,
|
||||
"critical_findings": 0,
|
||||
"duration_ms": 200,
|
||||
}
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
with patch("compliance.api.evidence_check_routes._run_header_check", new_callable=AsyncMock, return_value=header_result):
|
||||
resp = client.post(f"/evidence-checks/{CHECK_ID}/run")
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["run_status"] == "warning"
|
||||
|
||||
|
||||
class TestSeedChecks:
|
||||
def test_seed_checks(self):
|
||||
mock_db = MagicMock()
|
||||
# Each seed INSERT returns rowcount=1
|
||||
mock_result = MagicMock()
|
||||
mock_result.rowcount = 1
|
||||
mock_db.execute.return_value = mock_result
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
resp = client.post("/evidence-checks/seed")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total_definitions"] == 15
|
||||
assert data["seeded"] == 15
|
||||
|
||||
|
||||
class TestMappingsCRUD:
|
||||
def test_mappings_crud(self):
|
||||
mock_db = MagicMock()
|
||||
|
||||
def override_db():
|
||||
yield mock_db
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
client = TestClient(app)
|
||||
|
||||
# Create mapping
|
||||
mapping_row = _make_mapping_row()
|
||||
mock_db.execute.return_value.fetchone.return_value = mapping_row
|
||||
|
||||
resp = client.post("/evidence-checks/mappings", json={
|
||||
"evidence_id": EVIDENCE_ID,
|
||||
"control_code": "TOM-001",
|
||||
"mapping_type": "supports",
|
||||
"notes": "Test mapping",
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["control_code"] == "TOM-001"
|
||||
|
||||
# List mappings
|
||||
mock_db.execute.return_value.fetchall.return_value = [mapping_row]
|
||||
resp = client.get("/evidence-checks/mappings")
|
||||
assert resp.status_code == 200
|
||||
assert "mappings" in resp.json()
|
||||
|
||||
# Delete mapping
|
||||
mock_db.execute.return_value.rowcount = 1
|
||||
resp = client.delete(f"/evidence-checks/mappings/{MAPPING_ID}")
|
||||
assert resp.status_code == 204
|
||||
Reference in New Issue
Block a user