99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
|
from unittest.mock import MagicMock, patch
|
||
|
|
from datetime import datetime
|
||
|
|
from app.domain.compliance.ports import (
|
||
|
|
AnalysisRecord,
|
||
|
|
FindingRecord,
|
||
|
|
ComplianceRepository,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _mock_pool():
|
||
|
|
"""Return a mock psycopg2 ThreadedConnectionPool."""
|
||
|
|
conn = MagicMock()
|
||
|
|
cursor = MagicMock()
|
||
|
|
cursor.__enter__ = MagicMock(return_value=cursor)
|
||
|
|
cursor.__exit__ = MagicMock(return_value=False)
|
||
|
|
conn.cursor.return_value = cursor
|
||
|
|
pool = MagicMock()
|
||
|
|
pool.getconn.return_value = conn
|
||
|
|
return pool, conn, cursor
|
||
|
|
|
||
|
|
|
||
|
|
@patch("app.infrastructure.compliance.repository.psycopg2.pool.ThreadedConnectionPool")
|
||
|
|
def test_save_analysis_returns_uuid(mock_pool_cls):
|
||
|
|
from app.infrastructure.compliance.repository import PostgresComplianceRepository
|
||
|
|
pool, conn, cursor = _mock_pool()
|
||
|
|
mock_pool_cls.return_value = pool
|
||
|
|
cursor.fetchone.return_value = {"id": "abc-123"}
|
||
|
|
|
||
|
|
repo = PostgresComplianceRepository(
|
||
|
|
host="localhost", port=5432, user="u", password="p", dbname="db"
|
||
|
|
)
|
||
|
|
record = AnalysisRecord(
|
||
|
|
id="", created_at=datetime.utcnow(), created_by="user1",
|
||
|
|
doc_name="doc.pdf", standard_name="EU AI Act",
|
||
|
|
risk_score=50, conclusion="OK", actions=[], para_text="p",
|
||
|
|
highlight_terms=[], findings=[],
|
||
|
|
)
|
||
|
|
result = repo.save_analysis(record)
|
||
|
|
assert result == "abc-123"
|
||
|
|
|
||
|
|
|
||
|
|
def test_analysis_record_construction():
|
||
|
|
record = AnalysisRecord(
|
||
|
|
id="",
|
||
|
|
created_at=datetime.utcnow(),
|
||
|
|
created_by="user1",
|
||
|
|
doc_name="test.pdf",
|
||
|
|
standard_name="EU AI Act",
|
||
|
|
risk_score=72,
|
||
|
|
conclusion="Several gaps found.",
|
||
|
|
actions=[{"label": "Fix", "value": "Update docs"}],
|
||
|
|
para_text="The system shall...",
|
||
|
|
highlight_terms=["CSMS", "ISO 21434"],
|
||
|
|
findings=[
|
||
|
|
FindingRecord(
|
||
|
|
id="",
|
||
|
|
analysis_id="",
|
||
|
|
seq=0,
|
||
|
|
title="Missing CSMS",
|
||
|
|
description="No CSMS certification found.",
|
||
|
|
status="risk",
|
||
|
|
clause_ref="Art.9.1",
|
||
|
|
)
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert record.doc_name == "test.pdf"
|
||
|
|
assert len(record.findings) == 1
|
||
|
|
assert record.findings[0].status == "risk"
|
||
|
|
|
||
|
|
|
||
|
|
def test_compliance_repository_is_abstract():
|
||
|
|
import inspect
|
||
|
|
assert inspect.isabstract(ComplianceRepository)
|
||
|
|
|
||
|
|
|
||
|
|
def test_generate_docx_returns_bytes():
|
||
|
|
from app.infrastructure.compliance.docx_export import generate_docx
|
||
|
|
record = AnalysisRecord(
|
||
|
|
id="test-id", created_at=datetime(2026, 6, 8), created_by="user1",
|
||
|
|
doc_name="test.pdf", standard_name="EU AI Act",
|
||
|
|
risk_score=72, conclusion="Several gaps found.",
|
||
|
|
actions=[{"label": "Fix", "value": "Update CSMS docs"}],
|
||
|
|
para_text="The system shall implement CSMS.",
|
||
|
|
highlight_terms=["CSMS"],
|
||
|
|
findings=[
|
||
|
|
FindingRecord(
|
||
|
|
id="f1", analysis_id="test-id", seq=0,
|
||
|
|
title="Missing CSMS", description="No CSMS cert.",
|
||
|
|
status="risk", clause_ref="Art.9.1",
|
||
|
|
)
|
||
|
|
],
|
||
|
|
)
|
||
|
|
data = generate_docx(record)
|
||
|
|
assert isinstance(data, bytes)
|
||
|
|
assert len(data) > 1000 # DOCX is at minimum a ZIP with ~1 KB overhead
|
||
|
|
# Verify it's a valid ZIP (DOCX = ZIP container)
|
||
|
|
import zipfile, io
|
||
|
|
assert zipfile.is_zipfile(io.BytesIO(data))
|