bd9796725a
All checks were successful
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / test-go-ai-compliance (push) Successful in 36s
CI / test-python-backend-compliance (push) Successful in 35s
CI / test-python-document-crawler (push) Successful in 23s
CI / test-python-dsms-gateway (push) Successful in 21s
- 74 neue Tests (test_risk_routes, test_evidence_routes, test_requirement_routes, test_control_routes) Enum-Mocking (.value), ControlStatusEnum-Validierung, db.query() direkte Mocks - MkDocs: docs-src/services/sdk-modules/compliance-kern.md Endpunkt-Tabellen, Schema-Erklärungen, CI/CD-Beispiele, Risikomatrix - controls/page.tsx: "KI-Controls aus RAG vorschlagen" Button POST /api/sdk/v1/compliance/ai/suggest-controls, Suggestion-Panel, Requirement-ID-Eingabe + Dropdown, Konfidenz-Anzeige, Hinzufügen-Aktion Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
323 lines
13 KiB
Python
323 lines
13 KiB
Python
"""Tests for Requirements routes (routes.py → /compliance/requirements)."""
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import MagicMock, patch
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from compliance.api.routes import router as compliance_router
|
|
from classroom_engine.database import get_db
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# App setup with mocked DB dependency
|
|
# ---------------------------------------------------------------------------
|
|
|
|
app = FastAPI()
|
|
app.include_router(compliance_router)
|
|
|
|
mock_db = MagicMock()
|
|
|
|
|
|
def override_get_db():
|
|
yield mock_db
|
|
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
client = TestClient(app)
|
|
|
|
REQ_UUID = "rrrrrrrr-1111-2222-3333-rrrrrrrrrrrr"
|
|
REG_UUID = "gggggggg-1111-2222-3333-gggggggggggg"
|
|
NOW = datetime(2024, 3, 1, 12, 0, 0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def make_requirement(overrides=None):
|
|
r = MagicMock()
|
|
r.id = REQ_UUID
|
|
r.regulation_id = REG_UUID
|
|
r.regulation_code = "GDPR"
|
|
# regulation is accessed as r.regulation.code in list route
|
|
r.regulation = MagicMock()
|
|
r.regulation.code = "GDPR"
|
|
r.article = "Art. 32"
|
|
r.paragraph = "Abs. 1"
|
|
r.title = "Sicherheit der Verarbeitung"
|
|
r.description = "Angemessene technische Massnahmen"
|
|
r.requirement_text = "Implementierung geeigneter TOMs"
|
|
r.breakpilot_interpretation = "Risikobasierter Ansatz"
|
|
r.is_applicable = True
|
|
r.applicability_reason = "Kernanforderung"
|
|
r.priority = 1
|
|
r.implementation_status = "implemented"
|
|
r.implementation_details = None
|
|
r.code_references = None
|
|
r.documentation_links = None
|
|
r.evidence_description = None
|
|
r.evidence_artifacts = None
|
|
r.auditor_notes = None
|
|
r.audit_status = "pending"
|
|
r.last_audit_date = None
|
|
r.last_auditor = None
|
|
r.source_page = None
|
|
r.source_section = None
|
|
r.created_at = NOW
|
|
r.updated_at = NOW
|
|
if overrides:
|
|
for k, v in overrides.items():
|
|
setattr(r, k, v)
|
|
return r
|
|
|
|
|
|
def make_regulation(overrides=None):
|
|
reg = MagicMock()
|
|
reg.id = REG_UUID
|
|
reg.code = "GDPR"
|
|
reg.name = "DSGVO"
|
|
if overrides:
|
|
for k, v in overrides.items():
|
|
setattr(reg, k, v)
|
|
return reg
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: GET /compliance/requirements (paginated)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListRequirements:
|
|
"""Tests for GET /compliance/requirements."""
|
|
|
|
def test_list_empty(self):
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo:
|
|
MockRepo.return_value.get_paginated.return_value = ([], 0)
|
|
response = client.get("/compliance/requirements")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["data"] == []
|
|
assert data["pagination"]["total"] == 0
|
|
|
|
def test_list_with_requirement(self):
|
|
req = make_requirement()
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo:
|
|
MockRepo.return_value.get_paginated.return_value = ([req], 1)
|
|
response = client.get("/compliance/requirements")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["pagination"]["total"] == 1
|
|
r = data["data"][0]
|
|
assert r["article"] == "Art. 32"
|
|
assert r["title"] == "Sicherheit der Verarbeitung"
|
|
assert r["is_applicable"] is True
|
|
assert r["regulation_code"] == "GDPR"
|
|
|
|
def test_list_pagination_params(self):
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo:
|
|
MockRepo.return_value.get_paginated.return_value = ([], 0)
|
|
response = client.get("/compliance/requirements", params={"page": 2, "page_size": 25})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["pagination"]["page"] == 2
|
|
|
|
def test_list_filter_is_applicable(self):
|
|
req = make_requirement({"is_applicable": True})
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo:
|
|
MockRepo.return_value.get_paginated.return_value = ([req], 1)
|
|
response = client.get("/compliance/requirements", params={"is_applicable": "true"})
|
|
assert response.status_code == 200
|
|
assert response.json()["pagination"]["total"] == 1
|
|
|
|
def test_list_filter_search(self):
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo:
|
|
MockRepo.return_value.get_paginated.return_value = ([], 0)
|
|
response = client.get("/compliance/requirements", params={"search": "TOM"})
|
|
assert response.status_code == 200
|
|
|
|
def test_list_multiple_requirements(self):
|
|
r2 = make_requirement()
|
|
r2.id = "rrrrrrrr-2222-2222-2222-rrrrrrrrrrrr"
|
|
r2.article = "Art. 5"
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo:
|
|
MockRepo.return_value.get_paginated.return_value = ([make_requirement(), r2], 2)
|
|
response = client.get("/compliance/requirements")
|
|
assert response.status_code == 200
|
|
assert response.json()["pagination"]["total"] == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: GET /compliance/requirements/{id}
|
|
# Note: This route uses db.query() directly, not RequirementRepository.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetRequirementById:
|
|
"""Tests for GET /compliance/requirements/{id}."""
|
|
|
|
def test_get_existing(self):
|
|
req = make_requirement()
|
|
regulation = make_regulation()
|
|
# db.query(RequirementDB).filter(...).first() → req
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = [req, regulation]
|
|
response = client.get(f"/compliance/requirements/{REQ_UUID}")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == REQ_UUID
|
|
assert data["article"] == "Art. 32"
|
|
|
|
def test_get_not_found(self):
|
|
# Reset side_effect then set return_value to None
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = None
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
response = client.get("/compliance/requirements/nonexistent-id")
|
|
assert response.status_code == 404
|
|
|
|
def test_get_returns_dict_structure(self):
|
|
req = make_requirement()
|
|
regulation = make_regulation()
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = [req, regulation]
|
|
response = client.get(f"/compliance/requirements/{REQ_UUID}")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# dict response has these keys
|
|
assert "id" in data
|
|
assert "article" in data
|
|
assert "implementation_status" in data
|
|
assert "audit_status" in data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: GET /compliance/regulations/{code}/requirements
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetRequirementsByRegulation:
|
|
"""Tests for GET /compliance/regulations/{code}/requirements."""
|
|
|
|
def test_get_by_regulation_code(self):
|
|
req = make_requirement()
|
|
regulation = make_regulation()
|
|
with patch("compliance.api.routes.RegulationRepository") as MockRegRepo, \
|
|
patch("compliance.api.routes.RequirementRepository") as MockReqRepo:
|
|
MockRegRepo.return_value.get_by_code.return_value = regulation
|
|
MockReqRepo.return_value.get_by_regulation.return_value = [req]
|
|
response = client.get("/compliance/regulations/GDPR/requirements")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 1
|
|
assert data["requirements"][0]["article"] == "Art. 32"
|
|
|
|
def test_get_by_regulation_not_found(self):
|
|
with patch("compliance.api.routes.RegulationRepository") as MockRegRepo:
|
|
MockRegRepo.return_value.get_by_code.return_value = None
|
|
response = client.get("/compliance/regulations/UNKNOWN/requirements")
|
|
assert response.status_code == 404
|
|
|
|
def test_get_by_regulation_empty_requirements(self):
|
|
regulation = make_regulation()
|
|
with patch("compliance.api.routes.RegulationRepository") as MockRegRepo, \
|
|
patch("compliance.api.routes.RequirementRepository") as MockReqRepo:
|
|
MockRegRepo.return_value.get_by_code.return_value = regulation
|
|
MockReqRepo.return_value.get_by_regulation.return_value = []
|
|
response = client.get("/compliance/regulations/GDPR/requirements")
|
|
assert response.status_code == 200
|
|
assert response.json()["total"] == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: POST /compliance/requirements
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCreateRequirement:
|
|
"""Tests for POST /compliance/requirements."""
|
|
|
|
def test_create_success(self):
|
|
req = make_requirement()
|
|
regulation = make_regulation()
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo, \
|
|
patch("compliance.api.routes.RegulationRepository") as MockRegRepo:
|
|
MockRegRepo.return_value.get_by_id.return_value = regulation
|
|
MockRepo.return_value.create.return_value = req
|
|
response = client.post("/compliance/requirements", json={
|
|
"regulation_id": REG_UUID,
|
|
"article": "Art. 32",
|
|
"title": "Sicherheit der Verarbeitung",
|
|
"is_applicable": True,
|
|
"priority": 1,
|
|
})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["article"] == "Art. 32"
|
|
assert data["regulation_id"] == REG_UUID
|
|
|
|
def test_create_regulation_not_found(self):
|
|
with patch("compliance.api.routes.RegulationRepository") as MockRegRepo:
|
|
MockRegRepo.return_value.get_by_id.return_value = None
|
|
response = client.post("/compliance/requirements", json={
|
|
"regulation_id": "nonexistent-reg",
|
|
"article": "Art. 99",
|
|
"title": "Test",
|
|
"is_applicable": True,
|
|
"priority": 2,
|
|
})
|
|
assert response.status_code == 404
|
|
|
|
def test_create_missing_required_fields(self):
|
|
"""Missing title → 422."""
|
|
response = client.post("/compliance/requirements", json={
|
|
"regulation_id": REG_UUID,
|
|
"article": "Art. 32",
|
|
})
|
|
assert response.status_code == 422
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: DELETE /compliance/requirements/{id}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDeleteRequirement:
|
|
"""Tests for DELETE /compliance/requirements/{id}."""
|
|
|
|
def test_delete_success(self):
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo:
|
|
MockRepo.return_value.delete.return_value = True
|
|
response = client.delete(f"/compliance/requirements/{REQ_UUID}")
|
|
assert response.status_code == 200
|
|
assert response.json()["success"] is True
|
|
|
|
def test_delete_not_found(self):
|
|
with patch("compliance.api.routes.RequirementRepository") as MockRepo:
|
|
MockRepo.return_value.delete.return_value = False
|
|
response = client.delete("/compliance/requirements/nonexistent")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: PUT /compliance/requirements/{id}
|
|
# Note: This route uses db.query() directly.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestUpdateRequirement:
|
|
"""Tests for PUT /compliance/requirements/{id}."""
|
|
|
|
def test_update_success(self):
|
|
req = make_requirement()
|
|
req.implementation_status = "implemented"
|
|
# Reset any previous side_effect before setting return_value
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = None
|
|
mock_db.query.return_value.filter.return_value.first.return_value = req
|
|
response = client.put(
|
|
f"/compliance/requirements/{REQ_UUID}",
|
|
json={"implementation_status": "implemented"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
|
|
def test_update_not_found(self):
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = None
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
response = client.put(
|
|
"/compliance/requirements/nonexistent",
|
|
json={"implementation_status": "implemented"},
|
|
)
|
|
assert response.status_code == 404
|