feat(freigabe): Import/Screening/Modules/RAG — API-Tests, Migration 031, Bug-Fix
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 40s
CI / test-python-backend-compliance (push) Successful in 34s
CI / test-python-document-crawler (push) Successful in 26s
CI / test-python-dsms-gateway (push) Successful in 21s

- import_routes: GET /gap-analysis/{document_id} implementiert
- import_routes: Bug-Fix — gap_analysis_result vor try-Block initialisiert
  (verhindert UnboundLocalError bei DB-Fehler)
- test_import_routes: 21 neue API-Endpoint-Tests (59 total, alle grün)
- test_screening_routes: 18 neue API-Endpoint-Tests (74 total, alle grün)
- 031_modules.sql: Migration für compliance_service_modules,
  compliance_module_regulations, compliance_module_risks
- test_module_routes: 20 neue Tests für Module-Registry-Routen (alle grün)
- freigabe-module.md: MkDocs-Seite für Import/Screening/Modules/RAG
- mkdocs.yml: Nav-Eintrag "Freigabe-Module (Paket 2)"

Gesamt: 146 neue Tests, alle bestanden

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-05 11:42:19 +01:00
parent 0503e72a80
commit 3913931d5b
7 changed files with 1246 additions and 14 deletions

View File

@@ -438,3 +438,254 @@ class TestExtractFixVersionExtended:
}
result = extract_fix_version(vuln, "pkg")
assert result == "2.0.1"
# =============================================================================
# API Endpoint Tests
# =============================================================================
from fastapi import FastAPI
from fastapi.testclient import TestClient
from unittest.mock import MagicMock, patch, AsyncMock
from compliance.api.screening_routes import router as screening_router
_app_scr = FastAPI()
_app_scr.include_router(screening_router)
_client_scr = TestClient(_app_scr)
TENANT_ID = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
HEADERS = {"X-Tenant-ID": TENANT_ID}
SCREENING_UUID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
def _make_screening_row():
"""Return a row-like list for a screening DB record."""
# id, status, sbom_format, sbom_version, total_components, total_issues,
# critical_issues, high_issues, medium_issues, low_issues,
# sbom_data, started_at, completed_at
return [
SCREENING_UUID, "completed", "CycloneDX", "1.5",
3, 0, 0, 0, 0, 0,
{"components": [], "metadata": {}}, "2024-01-15T10:00:00", "2024-01-15T10:01:00",
]
class TestScanEndpoint:
"""API tests for POST /v1/screening/scan."""
def test_scan_requirements_txt_success(self):
"""Valid requirements.txt returns completed screening."""
txt = b"fastapi==0.100.0\nhttpx==0.25.0\npydantic==2.0.0"
with patch("compliance.api.screening_routes.SessionLocal") as MockSL, \
patch("compliance.api.screening_routes.scan_vulnerabilities", new_callable=AsyncMock) as mock_scan:
mock_scan.return_value = []
mock_session = MagicMock()
MockSL.return_value = mock_session
response = _client_scr.post(
"/v1/screening/scan",
files={"file": ("requirements.txt", txt, "text/plain")},
data={"tenant_id": TENANT_ID},
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "completed"
assert data["total_components"] == 3
assert data["total_issues"] == 0
assert data["sbom_format"] == "CycloneDX"
def test_scan_package_lock_success(self):
"""Valid package-lock.json returns completed screening."""
import json as _json
pkg_lock = _json.dumps({
"packages": {
"node_modules/react": {"version": "18.3.0", "license": "MIT"},
"node_modules/lodash": {"version": "4.17.21", "license": "MIT"},
}
}).encode()
with patch("compliance.api.screening_routes.SessionLocal") as MockSL, \
patch("compliance.api.screening_routes.scan_vulnerabilities", new_callable=AsyncMock) as mock_scan:
mock_scan.return_value = []
mock_session = MagicMock()
MockSL.return_value = mock_session
response = _client_scr.post(
"/v1/screening/scan",
files={"file": ("package-lock.json", pkg_lock, "application/json")},
data={"tenant_id": TENANT_ID},
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "completed"
assert data["total_components"] == 2
def test_scan_missing_file_returns_422(self):
"""Request without file returns 422."""
response = _client_scr.post(
"/v1/screening/scan",
data={"tenant_id": TENANT_ID},
)
assert response.status_code == 422
def test_scan_unparseable_file_returns_400(self):
"""File that cannot be parsed returns 400."""
with patch("compliance.api.screening_routes.SessionLocal"):
response = _client_scr.post(
"/v1/screening/scan",
files={"file": ("readme.md", b"# Just a readme", "text/plain")},
data={"tenant_id": TENANT_ID},
)
assert response.status_code == 400
def test_scan_with_vulnerabilities(self):
"""When vulnerabilities are found, issues list is populated."""
txt = b"fastapi==0.1.0"
fake_issue = {
"id": "issue-uuid",
"severity": "HIGH",
"title": "Remote Code Execution",
"description": "RCE vulnerability in fastapi",
"cve": "CVE-2024-0001",
"cvss": 7.5,
"affected_component": "fastapi",
"affected_version": "0.1.0",
"fixed_in": "0.2.0",
"remediation": "Upgrade fastapi to 0.2.0",
"status": "OPEN",
}
with patch("compliance.api.screening_routes.SessionLocal") as MockSL, \
patch("compliance.api.screening_routes.scan_vulnerabilities", new_callable=AsyncMock) as mock_scan:
mock_scan.return_value = [fake_issue]
mock_session = MagicMock()
MockSL.return_value = mock_session
response = _client_scr.post(
"/v1/screening/scan",
files={"file": ("requirements.txt", txt, "text/plain")},
data={"tenant_id": TENANT_ID},
)
assert response.status_code == 200
data = response.json()
assert data["total_issues"] == 1
assert data["high_issues"] == 1
assert len(data["issues"]) == 1
assert data["issues"][0]["cve"] == "CVE-2024-0001"
class TestGetScreeningEndpoint:
"""API tests for GET /v1/screening/{screening_id}."""
def test_get_screening_success(self):
"""Returns ScreeningResponse for a known ID."""
with patch("compliance.api.screening_routes.SessionLocal") as MockSL:
mock_session = MagicMock()
MockSL.return_value = mock_session
mock_result = MagicMock()
mock_result.fetchone.return_value = _make_screening_row()
mock_issues = MagicMock()
mock_issues.fetchall.return_value = []
mock_session.execute.side_effect = [mock_result, mock_issues]
response = _client_scr.get(f"/v1/screening/{SCREENING_UUID}")
assert response.status_code == 200
data = response.json()
assert data["id"] == SCREENING_UUID
assert data["status"] == "completed"
assert data["sbom_format"] == "CycloneDX"
def test_get_screening_not_found(self):
"""Returns 404 for unknown screening ID."""
with patch("compliance.api.screening_routes.SessionLocal") as MockSL:
mock_session = MagicMock()
MockSL.return_value = mock_session
mock_result = MagicMock()
mock_result.fetchone.return_value = None
mock_session.execute.return_value = mock_result
response = _client_scr.get("/v1/screening/nonexistent-uuid")
assert response.status_code == 404
def test_get_screening_includes_issues(self):
"""Issues from DB are included in response."""
with patch("compliance.api.screening_routes.SessionLocal") as MockSL:
mock_session = MagicMock()
MockSL.return_value = mock_session
mock_result = MagicMock()
mock_result.fetchone.return_value = _make_screening_row()
mock_issues = MagicMock()
# Row: id, severity, title, description, cve, cvss,
# affected_component, affected_version, fixed_in, remediation, status
mock_issues.fetchall.return_value = [
["issue-1", "HIGH", "XSS Vuln", "desc", "CVE-2024-001",
7.5, "react", "18.0.0", "18.3.0", "Upgrade react", "OPEN"],
]
mock_session.execute.side_effect = [mock_result, mock_issues]
response = _client_scr.get(f"/v1/screening/{SCREENING_UUID}")
assert response.status_code == 200
data = response.json()
assert len(data["issues"]) == 1
assert data["issues"][0]["severity"] == "HIGH"
class TestListScreeningsEndpoint:
"""API tests for GET /v1/screening."""
def test_list_empty(self):
"""Returns empty list when no screenings exist."""
with patch("compliance.api.screening_routes.SessionLocal") as MockSL:
mock_session = MagicMock()
MockSL.return_value = mock_session
mock_result = MagicMock()
mock_result.fetchall.return_value = []
mock_session.execute.return_value = mock_result
response = _client_scr.get("/v1/screening", params={"tenant_id": TENANT_ID})
assert response.status_code == 200
data = response.json()
assert data["screenings"] == []
assert data["total"] == 0
def test_list_with_data(self):
"""Returns correct total count."""
with patch("compliance.api.screening_routes.SessionLocal") as MockSL:
mock_session = MagicMock()
MockSL.return_value = mock_session
mock_result = MagicMock()
# Row: id, status, total_components, total_issues,
# critical, high, medium, low, started_at, completed_at, created_at
mock_result.fetchall.return_value = [
["uuid-1", "completed", 10, 2, 0, 1, 1, 0,
"2024-01-15T10:00:00", "2024-01-15T10:01:00", "2024-01-15"],
["uuid-2", "completed", 5, 0, 0, 0, 0, 0,
"2024-01-16T09:00:00", "2024-01-16T09:00:30", "2024-01-16"],
]
mock_session.execute.return_value = mock_result
response = _client_scr.get("/v1/screening", params={"tenant_id": TENANT_ID})
assert response.status_code == 200
data = response.json()
assert data["total"] == 2
assert len(data["screenings"]) == 2
def test_list_tenant_filter(self):
"""Tenant ID is used to filter screenings."""
with patch("compliance.api.screening_routes.SessionLocal") as MockSL:
mock_session = MagicMock()
MockSL.return_value = mock_session
mock_result = MagicMock()
mock_result.fetchall.return_value = []
mock_session.execute.return_value = mock_result
_client_scr.get("/v1/screening", params={"tenant_id": "specific-tenant"})
call_args = mock_session.execute.call_args
assert "specific-tenant" in str(call_args)