8742cb7f5a
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 35s
CI / test-python-backend-compliance (push) Successful in 32s
CI / test-python-document-crawler (push) Successful in 41s
CI / test-python-dsms-gateway (push) Successful in 19s
- Qdrant: lokaler Container → qdrant-dev.breakpilot.ai (gehostet, API-Key) - MinIO: bp-core-minio → Hetzner Object Storage (nbg1.your-objectstorage.com) - CLAUDE.md, MkDocs, ARCHITECTURE.md, training.md, ci-cd-pipeline.md Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
725 lines
25 KiB
Python
725 lines
25 KiB
Python
"""Tests for Vendor Compliance routes (vendor_compliance_routes.py).
|
||
|
||
Includes:
|
||
- Vendors: CRUD (5) + Stats (1) + Status-Patch (1) + Filter (2)
|
||
- Contracts: CRUD (5) + Filter (1)
|
||
- Findings: CRUD (5) + Filter (2)
|
||
- Control Instances: CRUD (5) + Filter (1)
|
||
- Controls Library: List + Create + Delete (3)
|
||
- Export Stubs: 3 × 501
|
||
- Response-Format: success/data/timestamp wrapper (2)
|
||
- camelCase/snake_case round-trip (2)
|
||
"""
|
||
|
||
import pytest
|
||
import uuid
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import create_engine, text, event
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||
|
||
from classroom_engine.database import get_db
|
||
from compliance.api.vendor_compliance_routes import router as vendor_compliance_router
|
||
|
||
# =============================================================================
|
||
# Test App + SQLite Setup
|
||
# =============================================================================
|
||
|
||
SQLALCHEMY_DATABASE_URL = "sqlite:///./test_vendor_compliance.db"
|
||
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
|
||
_RawSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||
|
||
TENANT_ID = "default"
|
||
|
||
|
||
@event.listens_for(engine, "connect")
|
||
def _register_sqlite_functions(dbapi_conn, connection_record):
|
||
dbapi_conn.create_function("NOW", 0, lambda: datetime.utcnow().isoformat())
|
||
|
||
|
||
class _DictRow(dict):
|
||
pass
|
||
|
||
|
||
class _DictSession:
|
||
def __init__(self, session):
|
||
self._session = session
|
||
|
||
def execute(self, stmt, params=None):
|
||
import re
|
||
if hasattr(stmt, 'text'):
|
||
rewritten = re.sub(r'CAST\((:[\w]+)\s+AS\s+jsonb\)', r'\1', stmt.text)
|
||
# Remove FILTER (WHERE ...) for SQLite — replace with CASE/SUM
|
||
# Simple approach: rewrite COUNT(*) FILTER (WHERE cond) → SUM(CASE WHEN cond THEN 1 ELSE 0 END)
|
||
filter_re = r'COUNT\(\*\)\s+FILTER\s*\(\s*WHERE\s+([^)]+)\)'
|
||
rewritten = re.sub(filter_re, r'SUM(CASE WHEN \1 THEN 1 ELSE 0 END)', rewritten)
|
||
# ILIKE → LIKE for SQLite
|
||
rewritten = rewritten.replace(' ILIKE ', ' LIKE ')
|
||
if rewritten != stmt.text:
|
||
stmt = text(rewritten)
|
||
result = self._session.execute(stmt, params)
|
||
return _DictResult(result)
|
||
|
||
def flush(self):
|
||
self._session.flush()
|
||
|
||
def commit(self):
|
||
self._session.commit()
|
||
|
||
def rollback(self):
|
||
self._session.rollback()
|
||
|
||
def close(self):
|
||
self._session.close()
|
||
|
||
|
||
class _DictResult:
|
||
def __init__(self, result):
|
||
self._result = result
|
||
try:
|
||
self._keys = list(result.keys())
|
||
self._returns_rows = True
|
||
except Exception:
|
||
self._keys = []
|
||
self._returns_rows = False
|
||
|
||
def fetchone(self):
|
||
if not self._returns_rows:
|
||
return None
|
||
row = self._result.fetchone()
|
||
if row is None:
|
||
return None
|
||
return _DictRow(zip(self._keys, row))
|
||
|
||
def fetchall(self):
|
||
if not self._returns_rows:
|
||
return []
|
||
rows = self._result.fetchall()
|
||
return [_DictRow(zip(self._keys, r)) for r in rows]
|
||
|
||
@property
|
||
def rowcount(self):
|
||
return self._result.rowcount
|
||
|
||
|
||
app = FastAPI()
|
||
app.include_router(vendor_compliance_router, prefix="/api/compliance")
|
||
|
||
|
||
def override_get_db():
|
||
session = _RawSessionLocal()
|
||
db = _DictSession(session)
|
||
try:
|
||
yield db
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
app.dependency_overrides[get_db] = override_get_db
|
||
client = TestClient(app)
|
||
|
||
|
||
# =============================================================================
|
||
# SQLite Table Creation
|
||
# =============================================================================
|
||
|
||
CREATE_VENDORS = """
|
||
CREATE TABLE IF NOT EXISTS vendor_vendors (
|
||
id TEXT PRIMARY KEY,
|
||
tenant_id TEXT NOT NULL DEFAULT 'default',
|
||
name TEXT NOT NULL DEFAULT '',
|
||
legal_form TEXT DEFAULT '',
|
||
country TEXT DEFAULT '',
|
||
address TEXT DEFAULT '',
|
||
website TEXT DEFAULT '',
|
||
role TEXT DEFAULT 'PROCESSOR',
|
||
service_description TEXT DEFAULT '',
|
||
service_category TEXT DEFAULT 'OTHER',
|
||
data_access_level TEXT DEFAULT 'NONE',
|
||
processing_locations TEXT DEFAULT '[]',
|
||
transfer_mechanisms TEXT DEFAULT '[]',
|
||
certifications TEXT DEFAULT '[]',
|
||
primary_contact TEXT DEFAULT '{}',
|
||
dpo_contact TEXT DEFAULT '{}',
|
||
security_contact TEXT DEFAULT '{}',
|
||
contract_types TEXT DEFAULT '[]',
|
||
inherent_risk_score INTEGER DEFAULT 50,
|
||
residual_risk_score INTEGER DEFAULT 50,
|
||
manual_risk_adjustment INTEGER,
|
||
risk_justification TEXT DEFAULT '',
|
||
review_frequency TEXT DEFAULT 'ANNUAL',
|
||
last_review_date TIMESTAMP,
|
||
next_review_date TIMESTAMP,
|
||
status TEXT DEFAULT 'ACTIVE',
|
||
processing_activity_ids TEXT DEFAULT '[]',
|
||
notes TEXT DEFAULT '',
|
||
contact_name TEXT DEFAULT '',
|
||
contact_email TEXT DEFAULT '',
|
||
contact_phone TEXT DEFAULT '',
|
||
contact_department TEXT DEFAULT '',
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
created_by TEXT DEFAULT 'system'
|
||
)
|
||
"""
|
||
|
||
CREATE_CONTRACTS = """
|
||
CREATE TABLE IF NOT EXISTS vendor_contracts (
|
||
id TEXT PRIMARY KEY,
|
||
tenant_id TEXT NOT NULL DEFAULT 'default',
|
||
vendor_id TEXT NOT NULL DEFAULT '',
|
||
file_name TEXT DEFAULT '',
|
||
original_name TEXT DEFAULT '',
|
||
mime_type TEXT DEFAULT '',
|
||
file_size INTEGER DEFAULT 0,
|
||
storage_path TEXT DEFAULT '',
|
||
document_type TEXT DEFAULT 'AVV',
|
||
version INTEGER DEFAULT 1,
|
||
previous_version_id TEXT,
|
||
parties TEXT DEFAULT '[]',
|
||
effective_date TIMESTAMP,
|
||
expiration_date TIMESTAMP,
|
||
auto_renewal INTEGER DEFAULT 0,
|
||
renewal_notice_period TEXT DEFAULT '',
|
||
termination_notice_period TEXT DEFAULT '',
|
||
review_status TEXT DEFAULT 'PENDING',
|
||
review_completed_at TIMESTAMP,
|
||
compliance_score INTEGER,
|
||
status TEXT DEFAULT 'DRAFT',
|
||
extracted_text TEXT DEFAULT '',
|
||
page_count INTEGER DEFAULT 0,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
created_by TEXT DEFAULT 'system'
|
||
)
|
||
"""
|
||
|
||
CREATE_FINDINGS = """
|
||
CREATE TABLE IF NOT EXISTS vendor_findings (
|
||
id TEXT PRIMARY KEY,
|
||
tenant_id TEXT NOT NULL DEFAULT 'default',
|
||
vendor_id TEXT NOT NULL DEFAULT '',
|
||
contract_id TEXT,
|
||
finding_type TEXT DEFAULT 'UNKNOWN',
|
||
category TEXT DEFAULT '',
|
||
severity TEXT DEFAULT 'MEDIUM',
|
||
title TEXT DEFAULT '',
|
||
description TEXT DEFAULT '',
|
||
recommendation TEXT DEFAULT '',
|
||
citations TEXT DEFAULT '[]',
|
||
status TEXT DEFAULT 'OPEN',
|
||
assignee TEXT DEFAULT '',
|
||
due_date TIMESTAMP,
|
||
resolution TEXT DEFAULT '',
|
||
resolved_at TIMESTAMP,
|
||
resolved_by TEXT DEFAULT '',
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
created_by TEXT DEFAULT 'system'
|
||
)
|
||
"""
|
||
|
||
CREATE_CONTROL_INSTANCES = """
|
||
CREATE TABLE IF NOT EXISTS vendor_control_instances (
|
||
id TEXT PRIMARY KEY,
|
||
tenant_id TEXT NOT NULL DEFAULT 'default',
|
||
vendor_id TEXT NOT NULL DEFAULT '',
|
||
control_id TEXT DEFAULT '',
|
||
control_domain TEXT DEFAULT '',
|
||
status TEXT DEFAULT 'PLANNED',
|
||
evidence_ids TEXT DEFAULT '[]',
|
||
notes TEXT DEFAULT '',
|
||
last_assessed_at TIMESTAMP,
|
||
last_assessed_by TEXT DEFAULT '',
|
||
next_assessment_date TIMESTAMP,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
created_by TEXT DEFAULT 'system'
|
||
)
|
||
"""
|
||
|
||
CREATE_CONTROLS = """
|
||
CREATE TABLE IF NOT EXISTS vendor_compliance_controls (
|
||
id TEXT PRIMARY KEY,
|
||
tenant_id TEXT NOT NULL DEFAULT 'default',
|
||
domain TEXT DEFAULT '',
|
||
control_code TEXT DEFAULT '',
|
||
title TEXT DEFAULT '',
|
||
description TEXT DEFAULT '',
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
"""
|
||
|
||
|
||
def _setup_tables():
|
||
with engine.connect() as conn:
|
||
for sql in [CREATE_VENDORS, CREATE_CONTRACTS, CREATE_FINDINGS,
|
||
CREATE_CONTROL_INSTANCES, CREATE_CONTROLS]:
|
||
conn.execute(text(sql))
|
||
conn.commit()
|
||
|
||
|
||
def _teardown_tables():
|
||
with engine.connect() as conn:
|
||
for t in ["vendor_vendors", "vendor_contracts", "vendor_findings",
|
||
"vendor_control_instances", "vendor_compliance_controls"]:
|
||
conn.execute(text(f"DELETE FROM {t}"))
|
||
conn.commit()
|
||
|
||
|
||
_setup_tables()
|
||
|
||
|
||
# =============================================================================
|
||
# Fixtures
|
||
# =============================================================================
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def clean_tables():
|
||
_teardown_tables()
|
||
yield
|
||
_teardown_tables()
|
||
|
||
|
||
def _create_vendor(**kwargs):
|
||
payload = {
|
||
"name": kwargs.get("name", "Test Vendor GmbH"),
|
||
"country": "DE",
|
||
"role": "PROCESSOR",
|
||
"serviceCategory": "HOSTING",
|
||
"status": kwargs.get("status", "ACTIVE"),
|
||
"inherentRiskScore": kwargs.get("inherentRiskScore", 50),
|
||
}
|
||
payload.update(kwargs)
|
||
resp = client.post("/api/compliance/vendor-compliance/vendors", json=payload)
|
||
assert resp.status_code == 201
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _create_contract(vendor_id, **kwargs):
|
||
payload = {
|
||
"vendorId": vendor_id,
|
||
"documentType": "AVV",
|
||
"fileName": "avv-test.pdf",
|
||
"status": "DRAFT",
|
||
}
|
||
payload.update(kwargs)
|
||
resp = client.post("/api/compliance/vendor-compliance/contracts", json=payload)
|
||
assert resp.status_code == 201
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _create_finding(vendor_id, **kwargs):
|
||
payload = {
|
||
"vendorId": vendor_id,
|
||
"findingType": "GAP",
|
||
"severity": "HIGH",
|
||
"title": "Missing TOM Annex",
|
||
"status": "OPEN",
|
||
}
|
||
payload.update(kwargs)
|
||
resp = client.post("/api/compliance/vendor-compliance/findings", json=payload)
|
||
assert resp.status_code == 201
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _create_control_instance(vendor_id, **kwargs):
|
||
payload = {
|
||
"vendorId": vendor_id,
|
||
"controlId": "C-001",
|
||
"controlDomain": "priv",
|
||
"status": "PASS",
|
||
}
|
||
payload.update(kwargs)
|
||
resp = client.post("/api/compliance/vendor-compliance/control-instances", json=payload)
|
||
assert resp.status_code == 201
|
||
return resp.json()["data"]
|
||
|
||
|
||
# =============================================================================
|
||
# Response Format Tests
|
||
# =============================================================================
|
||
|
||
class TestResponseFormat:
|
||
def test_list_vendors_has_success_data_timestamp(self):
|
||
resp = client.get("/api/compliance/vendor-compliance/vendors")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["success"] is True
|
||
assert "data" in body
|
||
assert "timestamp" in body
|
||
|
||
def test_create_vendor_has_success_data_timestamp(self):
|
||
resp = client.post("/api/compliance/vendor-compliance/vendors", json={"name": "Test"})
|
||
assert resp.status_code == 201
|
||
body = resp.json()
|
||
assert body["success"] is True
|
||
assert "data" in body
|
||
assert body["data"]["name"] == "Test"
|
||
assert "timestamp" in body
|
||
|
||
|
||
# =============================================================================
|
||
# camelCase / snake_case Round-Trip Tests
|
||
# =============================================================================
|
||
|
||
class TestCamelSnakeConversion:
|
||
def test_create_with_camel_returns_camel(self):
|
||
vendor = _create_vendor(
|
||
name="CamelTest",
|
||
legalForm="GmbH",
|
||
serviceDescription="Cloud hosting",
|
||
dataAccessLevel="CONTENT",
|
||
inherentRiskScore=80,
|
||
)
|
||
assert vendor["legalForm"] == "GmbH"
|
||
assert vendor["serviceDescription"] == "Cloud hosting"
|
||
assert vendor["dataAccessLevel"] == "CONTENT"
|
||
assert vendor["inherentRiskScore"] == 80
|
||
|
||
def test_round_trip_preserves_values(self):
|
||
vendor = _create_vendor(
|
||
name="RoundTrip",
|
||
processingLocations=["DE", "US"],
|
||
primaryContact={"name": "Max", "email": "max@test.de"},
|
||
)
|
||
vid = vendor["id"]
|
||
resp = client.get(f"/api/compliance/vendor-compliance/vendors/{vid}")
|
||
assert resp.status_code == 200
|
||
fetched = resp.json()["data"]
|
||
assert fetched["processingLocations"] == ["DE", "US"]
|
||
assert fetched["primaryContact"]["name"] == "Max"
|
||
|
||
|
||
# =============================================================================
|
||
# Vendor Tests
|
||
# =============================================================================
|
||
|
||
class TestVendorsCRUD:
|
||
def test_list_empty(self):
|
||
resp = client.get("/api/compliance/vendor-compliance/vendors")
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert data["items"] == []
|
||
assert data["total"] == 0
|
||
|
||
def test_create_vendor(self):
|
||
vendor = _create_vendor(name="Hetzner GmbH")
|
||
assert vendor["name"] == "Hetzner GmbH"
|
||
assert "id" in vendor
|
||
|
||
def test_get_vendor(self):
|
||
vendor = _create_vendor()
|
||
resp = client.get(f"/api/compliance/vendor-compliance/vendors/{vendor['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["id"] == vendor["id"]
|
||
|
||
def test_update_vendor(self):
|
||
vendor = _create_vendor()
|
||
resp = client.put(
|
||
f"/api/compliance/vendor-compliance/vendors/{vendor['id']}",
|
||
json={"name": "Updated Name", "country": "AT"}
|
||
)
|
||
assert resp.status_code == 200
|
||
updated = resp.json()["data"]
|
||
assert updated["name"] == "Updated Name"
|
||
assert updated["country"] == "AT"
|
||
|
||
def test_delete_vendor(self):
|
||
vendor = _create_vendor()
|
||
resp = client.delete(f"/api/compliance/vendor-compliance/vendors/{vendor['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["deleted"] is True
|
||
resp2 = client.get(f"/api/compliance/vendor-compliance/vendors/{vendor['id']}")
|
||
assert resp2.status_code == 404
|
||
|
||
def test_get_nonexistent_vendor_404(self):
|
||
resp = client.get(f"/api/compliance/vendor-compliance/vendors/{uuid.uuid4()}")
|
||
assert resp.status_code == 404
|
||
|
||
def test_delete_nonexistent_vendor_404(self):
|
||
resp = client.delete(f"/api/compliance/vendor-compliance/vendors/{uuid.uuid4()}")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
class TestVendorStats:
|
||
def test_stats_empty(self):
|
||
resp = client.get("/api/compliance/vendor-compliance/vendors/stats")
|
||
assert resp.status_code == 200
|
||
stats = resp.json()["data"]
|
||
assert stats["total"] == 0
|
||
|
||
def test_stats_with_vendors(self):
|
||
_create_vendor(name="V1", status="ACTIVE", inherentRiskScore=80)
|
||
_create_vendor(name="V2", status="INACTIVE", inherentRiskScore=30)
|
||
_create_vendor(name="V3", status="PENDING_REVIEW", inherentRiskScore=90)
|
||
resp = client.get("/api/compliance/vendor-compliance/vendors/stats")
|
||
stats = resp.json()["data"]
|
||
assert stats["total"] == 3
|
||
assert stats["active"] == 1
|
||
assert stats["inactive"] == 1
|
||
assert stats["pendingReview"] == 1
|
||
assert stats["highRiskCount"] == 2 # 80 and 90
|
||
|
||
|
||
class TestVendorStatusPatch:
|
||
def test_patch_status(self):
|
||
vendor = _create_vendor(status="ACTIVE")
|
||
resp = client.patch(
|
||
f"/api/compliance/vendor-compliance/vendors/{vendor['id']}/status",
|
||
json={"status": "TERMINATED"}
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["status"] == "TERMINATED"
|
||
|
||
def test_patch_invalid_status_400(self):
|
||
vendor = _create_vendor()
|
||
resp = client.patch(
|
||
f"/api/compliance/vendor-compliance/vendors/{vendor['id']}/status",
|
||
json={"status": "INVALID"}
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
|
||
class TestVendorFilter:
|
||
def test_filter_by_status(self):
|
||
_create_vendor(name="Active1", status="ACTIVE")
|
||
_create_vendor(name="Inactive1", status="INACTIVE")
|
||
resp = client.get("/api/compliance/vendor-compliance/vendors?status=ACTIVE")
|
||
items = resp.json()["data"]["items"]
|
||
assert len(items) == 1
|
||
assert items[0]["name"] == "Active1"
|
||
|
||
def test_filter_by_search(self):
|
||
_create_vendor(name="Hetzner Online GmbH")
|
||
_create_vendor(name="AWS Deutschland")
|
||
resp = client.get("/api/compliance/vendor-compliance/vendors?search=Hetzner")
|
||
items = resp.json()["data"]["items"]
|
||
assert len(items) == 1
|
||
assert "Hetzner" in items[0]["name"]
|
||
|
||
|
||
# =============================================================================
|
||
# Contract Tests
|
||
# =============================================================================
|
||
|
||
class TestContractsCRUD:
|
||
def test_list_contracts_empty(self):
|
||
resp = client.get("/api/compliance/vendor-compliance/contracts")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"] == []
|
||
|
||
def test_create_contract(self):
|
||
vendor = _create_vendor()
|
||
contract = _create_contract(vendor["id"])
|
||
assert contract["vendorId"] == vendor["id"]
|
||
assert contract["documentType"] == "AVV"
|
||
|
||
def test_get_contract(self):
|
||
vendor = _create_vendor()
|
||
contract = _create_contract(vendor["id"])
|
||
resp = client.get(f"/api/compliance/vendor-compliance/contracts/{contract['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["id"] == contract["id"]
|
||
|
||
def test_update_contract(self):
|
||
vendor = _create_vendor()
|
||
contract = _create_contract(vendor["id"])
|
||
resp = client.put(
|
||
f"/api/compliance/vendor-compliance/contracts/{contract['id']}",
|
||
json={"status": "ACTIVE", "complianceScore": 85}
|
||
)
|
||
assert resp.status_code == 200
|
||
updated = resp.json()["data"]
|
||
assert updated["status"] == "ACTIVE"
|
||
assert updated["complianceScore"] == 85
|
||
|
||
def test_delete_contract(self):
|
||
vendor = _create_vendor()
|
||
contract = _create_contract(vendor["id"])
|
||
resp = client.delete(f"/api/compliance/vendor-compliance/contracts/{contract['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["deleted"] is True
|
||
|
||
|
||
class TestContractFilter:
|
||
def test_filter_by_vendor_id(self):
|
||
v1 = _create_vendor(name="V1")
|
||
v2 = _create_vendor(name="V2")
|
||
_create_contract(v1["id"])
|
||
_create_contract(v1["id"])
|
||
_create_contract(v2["id"])
|
||
resp = client.get(f"/api/compliance/vendor-compliance/contracts?vendor_id={v1['id']}")
|
||
assert len(resp.json()["data"]) == 2
|
||
|
||
|
||
# =============================================================================
|
||
# Finding Tests
|
||
# =============================================================================
|
||
|
||
class TestFindingsCRUD:
|
||
def test_list_findings_empty(self):
|
||
resp = client.get("/api/compliance/vendor-compliance/findings")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"] == []
|
||
|
||
def test_create_finding(self):
|
||
vendor = _create_vendor()
|
||
finding = _create_finding(vendor["id"])
|
||
assert finding["vendorId"] == vendor["id"]
|
||
assert finding["severity"] == "HIGH"
|
||
|
||
def test_get_finding(self):
|
||
vendor = _create_vendor()
|
||
finding = _create_finding(vendor["id"])
|
||
resp = client.get(f"/api/compliance/vendor-compliance/findings/{finding['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["title"] == "Missing TOM Annex"
|
||
|
||
def test_update_finding(self):
|
||
vendor = _create_vendor()
|
||
finding = _create_finding(vendor["id"])
|
||
resp = client.put(
|
||
f"/api/compliance/vendor-compliance/findings/{finding['id']}",
|
||
json={"status": "RESOLVED", "resolution": "TOM annex added"}
|
||
)
|
||
assert resp.status_code == 200
|
||
updated = resp.json()["data"]
|
||
assert updated["status"] == "RESOLVED"
|
||
assert updated["resolution"] == "TOM annex added"
|
||
|
||
def test_delete_finding(self):
|
||
vendor = _create_vendor()
|
||
finding = _create_finding(vendor["id"])
|
||
resp = client.delete(f"/api/compliance/vendor-compliance/findings/{finding['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["deleted"] is True
|
||
|
||
|
||
class TestFindingFilter:
|
||
def test_filter_by_severity(self):
|
||
vendor = _create_vendor()
|
||
_create_finding(vendor["id"], severity="HIGH")
|
||
_create_finding(vendor["id"], severity="LOW")
|
||
resp = client.get("/api/compliance/vendor-compliance/findings?severity=HIGH")
|
||
assert len(resp.json()["data"]) == 1
|
||
|
||
def test_filter_by_vendor_id(self):
|
||
v1 = _create_vendor(name="V1")
|
||
v2 = _create_vendor(name="V2")
|
||
_create_finding(v1["id"])
|
||
_create_finding(v2["id"])
|
||
resp = client.get(f"/api/compliance/vendor-compliance/findings?vendor_id={v1['id']}")
|
||
assert len(resp.json()["data"]) == 1
|
||
|
||
|
||
# =============================================================================
|
||
# Control Instance Tests
|
||
# =============================================================================
|
||
|
||
class TestControlInstancesCRUD:
|
||
def test_list_control_instances_empty(self):
|
||
resp = client.get("/api/compliance/vendor-compliance/control-instances")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"] == []
|
||
|
||
def test_create_control_instance(self):
|
||
vendor = _create_vendor()
|
||
ci = _create_control_instance(vendor["id"])
|
||
assert ci["vendorId"] == vendor["id"]
|
||
assert ci["controlId"] == "C-001"
|
||
assert ci["status"] == "PASS"
|
||
|
||
def test_get_control_instance(self):
|
||
vendor = _create_vendor()
|
||
ci = _create_control_instance(vendor["id"])
|
||
resp = client.get(f"/api/compliance/vendor-compliance/control-instances/{ci['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["controlDomain"] == "priv"
|
||
|
||
def test_update_control_instance(self):
|
||
vendor = _create_vendor()
|
||
ci = _create_control_instance(vendor["id"])
|
||
resp = client.put(
|
||
f"/api/compliance/vendor-compliance/control-instances/{ci['id']}",
|
||
json={"status": "FAIL", "notes": "Needs remediation"}
|
||
)
|
||
assert resp.status_code == 200
|
||
updated = resp.json()["data"]
|
||
assert updated["status"] == "FAIL"
|
||
assert updated["notes"] == "Needs remediation"
|
||
|
||
def test_delete_control_instance(self):
|
||
vendor = _create_vendor()
|
||
ci = _create_control_instance(vendor["id"])
|
||
resp = client.delete(f"/api/compliance/vendor-compliance/control-instances/{ci['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["deleted"] is True
|
||
|
||
|
||
class TestControlInstanceFilter:
|
||
def test_filter_by_vendor_id(self):
|
||
v1 = _create_vendor(name="V1")
|
||
v2 = _create_vendor(name="V2")
|
||
_create_control_instance(v1["id"])
|
||
_create_control_instance(v2["id"])
|
||
resp = client.get(f"/api/compliance/vendor-compliance/control-instances?vendor_id={v1['id']}")
|
||
assert len(resp.json()["data"]) == 1
|
||
|
||
|
||
# =============================================================================
|
||
# Controls Library Tests
|
||
# =============================================================================
|
||
|
||
class TestControlsLibrary:
|
||
def test_list_controls_empty(self):
|
||
resp = client.get("/api/compliance/vendor-compliance/controls")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"] == []
|
||
|
||
def test_create_control(self):
|
||
resp = client.post("/api/compliance/vendor-compliance/controls", json={
|
||
"domain": "priv",
|
||
"controlCode": "PRIV-001",
|
||
"title": "Datenschutz-Folgenabschaetzung",
|
||
"description": "Art. 35 DSGVO Compliance"
|
||
})
|
||
assert resp.status_code == 201
|
||
ctrl = resp.json()["data"]
|
||
assert ctrl["domain"] == "priv"
|
||
assert ctrl["controlCode"] == "PRIV-001"
|
||
|
||
def test_delete_control(self):
|
||
resp = client.post("/api/compliance/vendor-compliance/controls", json={
|
||
"domain": "iam", "controlCode": "IAM-001", "title": "Access Control"
|
||
})
|
||
ctrl_id = resp.json()["data"]["id"]
|
||
resp2 = client.delete(f"/api/compliance/vendor-compliance/controls/{ctrl_id}")
|
||
assert resp2.status_code == 200
|
||
assert resp2.json()["data"]["deleted"] is True
|
||
|
||
|
||
# =============================================================================
|
||
# Export Stub Tests
|
||
# =============================================================================
|
||
|
||
class TestExportStubs:
|
||
def test_post_export_501(self):
|
||
resp = client.post("/api/compliance/vendor-compliance/export", json={})
|
||
assert resp.status_code == 501
|
||
assert resp.json()["success"] is False
|
||
|
||
def test_get_export_501(self):
|
||
resp = client.get(f"/api/compliance/vendor-compliance/export/{uuid.uuid4()}")
|
||
assert resp.status_code == 501
|
||
|
||
def test_download_export_501(self):
|
||
resp = client.get(f"/api/compliance/vendor-compliance/export/{uuid.uuid4()}/download")
|
||
assert resp.status_code == 501
|