434 lines
17 KiB
Python
434 lines
17 KiB
Python
# #region Test.CleanRelease.DtoEnumsExceptions [C:2] [TYPE Module] [SEMANTICS test,clean-release,dto,enums,exceptions]
|
|
# @BRIEF Tests for clean_release/dto.py, enums.py, exceptions.py.
|
|
# @RELATION BINDS_TO -> [CleanReleaseDto.Enums.Exceptions]
|
|
# @TEST_EDGE: missing_field -> optional fields default correctly
|
|
# @TEST_EDGE: serialization -> round-trip via model_dump/model_validate
|
|
# @TEST_EDGE: exception_hierarchy -> custom exceptions form proper chain
|
|
|
|
from datetime import datetime, timezone
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
|
|
# ── DTOs ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestCandidateDTO:
|
|
"""CandidateDTO — DTO for ReleaseCandidate."""
|
|
|
|
def test_minimal(self):
|
|
from src.services.clean_release.dto import CandidateDTO
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = CandidateDTO(
|
|
id="c-1", version="1.0.0",
|
|
source_snapshot_ref="snap-1",
|
|
created_at=ts, created_by="admin",
|
|
status=CandidateStatus.DRAFT,
|
|
)
|
|
assert dto.id == "c-1"
|
|
assert dto.build_id is None
|
|
|
|
def test_with_build_id(self):
|
|
from src.services.clean_release.dto import CandidateDTO
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = CandidateDTO(
|
|
id="c-1", version="1.0.0",
|
|
source_snapshot_ref="snap-1",
|
|
build_id="build-1",
|
|
created_at=ts, created_by="admin",
|
|
status=CandidateStatus.PREPARED,
|
|
)
|
|
assert dto.build_id == "build-1"
|
|
assert dto.status == "PREPARED"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.services.clean_release.dto import CandidateDTO
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = CandidateDTO(
|
|
id="c-1", version="1.0.0",
|
|
source_snapshot_ref="snap-1",
|
|
created_at=ts, created_by="admin",
|
|
status=CandidateStatus.DRAFT,
|
|
)
|
|
data = dto.model_dump()
|
|
restored = CandidateDTO.model_validate(data)
|
|
assert restored.id == "c-1"
|
|
assert restored.version == "1.0.0"
|
|
|
|
def test_missing_required_raises(self):
|
|
from src.services.clean_release.dto import CandidateDTO
|
|
|
|
with pytest.raises(ValidationError):
|
|
CandidateDTO()
|
|
|
|
|
|
class TestArtifactDTO:
|
|
"""ArtifactDTO — DTO for CandidateArtifact."""
|
|
|
|
def test_minimal(self):
|
|
from src.services.clean_release.dto import ArtifactDTO
|
|
|
|
dto = ArtifactDTO(id="a-1", candidate_id="c-1", path="/file.txt", sha256="abc123", size=1024)
|
|
assert dto.id == "a-1"
|
|
assert dto.path == "/file.txt"
|
|
assert dto.size == 1024
|
|
assert dto.metadata == {}
|
|
|
|
def test_with_all_fields(self):
|
|
from src.services.clean_release.dto import ArtifactDTO
|
|
|
|
dto = ArtifactDTO(
|
|
id="a-1", candidate_id="c-1", path="/file.txt",
|
|
sha256="abc123", size=1024,
|
|
detected_category="binary", declared_category="application",
|
|
source_uri="http://example.com/file.txt",
|
|
source_host="example.com",
|
|
metadata={"key": "value"},
|
|
)
|
|
assert dto.detected_category == "binary"
|
|
assert dto.source_uri == "http://example.com/file.txt"
|
|
assert dto.metadata == {"key": "value"}
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.services.clean_release.dto import ArtifactDTO
|
|
|
|
dto = ArtifactDTO(id="a-1", candidate_id="c-1", path="/f", sha256="h", size=1)
|
|
data = dto.model_dump()
|
|
restored = ArtifactDTO.model_validate(data)
|
|
assert restored.id == "a-1"
|
|
assert restored.size == 1
|
|
|
|
|
|
class TestManifestDTO:
|
|
"""ManifestDTO — DTO for DistributionManifest."""
|
|
|
|
def test_minimal(self):
|
|
from src.services.clean_release.dto import ManifestDTO
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ManifestDTO(
|
|
id="m-1", candidate_id="c-1",
|
|
manifest_version=1, manifest_digest="digest1",
|
|
artifacts_digest="adigest1",
|
|
created_at=ts, created_by="admin",
|
|
source_snapshot_ref="snap-1",
|
|
content_json={"key": "value"},
|
|
)
|
|
assert dto.manifest_version == 1
|
|
assert dto.manifest_digest == "digest1"
|
|
assert dto.content_json == {"key": "value"}
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.services.clean_release.dto import ManifestDTO
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ManifestDTO(
|
|
id="m-1", candidate_id="c-1",
|
|
manifest_version=1, manifest_digest="d1",
|
|
artifacts_digest="ad1",
|
|
created_at=ts, created_by="admin",
|
|
source_snapshot_ref="snap-1", content_json={},
|
|
)
|
|
data = dto.model_dump()
|
|
restored = ManifestDTO.model_validate(data)
|
|
assert restored.manifest_digest == "d1"
|
|
|
|
|
|
class TestComplianceRunDTO:
|
|
"""ComplianceRunDTO — DTO for run status tracking."""
|
|
|
|
def test_minimal(self):
|
|
from src.services.clean_release.dto import ComplianceRunDTO
|
|
from src.services.clean_release.enums import RunStatus
|
|
|
|
dto = ComplianceRunDTO(run_id="r-1", candidate_id="c-1", status=RunStatus.PENDING)
|
|
assert dto.run_id == "r-1"
|
|
assert dto.status == "PENDING"
|
|
assert dto.final_status is None
|
|
assert dto.report_id is None
|
|
|
|
def test_with_all_fields(self):
|
|
from src.services.clean_release.dto import ComplianceRunDTO
|
|
from src.services.clean_release.enums import ComplianceDecision, RunStatus
|
|
|
|
dto = ComplianceRunDTO(
|
|
run_id="r-1", candidate_id="c-1",
|
|
status=RunStatus.SUCCEEDED,
|
|
final_status=ComplianceDecision.PASSED,
|
|
report_id="rep-1", task_id="task-1",
|
|
)
|
|
assert dto.final_status == "PASSED"
|
|
assert dto.report_id == "rep-1"
|
|
assert dto.task_id == "task-1"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.services.clean_release.dto import ComplianceRunDTO
|
|
from src.services.clean_release.enums import RunStatus
|
|
|
|
dto = ComplianceRunDTO(run_id="r-1", candidate_id="c-1", status=RunStatus.PENDING)
|
|
data = dto.model_dump()
|
|
restored = ComplianceRunDTO.model_validate(data)
|
|
assert restored.run_id == "r-1"
|
|
|
|
|
|
class TestReportDTO:
|
|
"""ReportDTO — compact report view."""
|
|
|
|
def test_minimal(self):
|
|
from src.services.clean_release.dto import ReportDTO
|
|
from src.services.clean_release.enums import ComplianceDecision
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ReportDTO(
|
|
report_id="r-1", candidate_id="c-1",
|
|
final_status=ComplianceDecision.PASSED,
|
|
policy_version="1.0", manifest_digest="digest1",
|
|
violation_count=0, generated_at=ts,
|
|
)
|
|
assert dto.report_id == "r-1"
|
|
assert dto.final_status == "PASSED"
|
|
assert dto.violation_count == 0
|
|
|
|
def test_with_violations(self):
|
|
from src.services.clean_release.dto import ReportDTO
|
|
from src.services.clean_release.enums import ComplianceDecision
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ReportDTO(
|
|
report_id="r-1", candidate_id="c-1",
|
|
final_status=ComplianceDecision.BLOCKED,
|
|
policy_version="1.0", manifest_digest="digest1",
|
|
violation_count=5, generated_at=ts,
|
|
)
|
|
assert dto.violation_count == 5
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.services.clean_release.dto import ReportDTO
|
|
from src.services.clean_release.enums import ComplianceDecision
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ReportDTO(report_id="r-1", candidate_id="c-1",
|
|
final_status=ComplianceDecision.PASSED,
|
|
policy_version="1", manifest_digest="d1",
|
|
violation_count=0, generated_at=ts)
|
|
data = dto.model_dump()
|
|
restored = ReportDTO.model_validate(data)
|
|
assert restored.violation_count == 0
|
|
|
|
|
|
class TestCandidateOverviewDTO:
|
|
"""CandidateOverviewDTO — full candidate overview with all status refs."""
|
|
|
|
def test_minimal(self):
|
|
from src.services.clean_release.dto import CandidateOverviewDTO
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
dto = CandidateOverviewDTO(
|
|
candidate_id="c-1", version="1.0.0",
|
|
source_snapshot_ref="snap-1",
|
|
status=CandidateStatus.DRAFT,
|
|
)
|
|
assert dto.candidate_id == "c-1"
|
|
assert dto.latest_manifest_id is None
|
|
assert dto.latest_approval_decision is None
|
|
assert dto.latest_publication_id is None
|
|
|
|
def test_with_all_fields(self):
|
|
from src.services.clean_release.dto import CandidateOverviewDTO
|
|
from src.services.clean_release.enums import CandidateStatus, ComplianceDecision, RunStatus
|
|
|
|
dto = CandidateOverviewDTO(
|
|
candidate_id="c-1", version="1.0.0",
|
|
source_snapshot_ref="snap-1",
|
|
status=CandidateStatus.APPROVED,
|
|
latest_manifest_id="m-1",
|
|
latest_manifest_digest="digest1",
|
|
latest_run_id="run-1",
|
|
latest_run_status=RunStatus.SUCCEEDED,
|
|
latest_report_id="rep-1",
|
|
latest_report_final_status=ComplianceDecision.PASSED,
|
|
latest_policy_snapshot_id="ps-1",
|
|
latest_policy_version="1.0",
|
|
latest_registry_snapshot_id="rs-1",
|
|
latest_registry_version="1.0",
|
|
latest_approval_decision="APPROVED",
|
|
latest_publication_id="pub-1",
|
|
latest_publication_status="ACTIVE",
|
|
)
|
|
assert dto.latest_manifest_id == "m-1"
|
|
assert dto.latest_run_status == "SUCCEEDED"
|
|
assert dto.latest_approval_decision == "APPROVED"
|
|
assert dto.latest_publication_status == "ACTIVE"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.services.clean_release.dto import CandidateOverviewDTO
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
dto = CandidateOverviewDTO(
|
|
candidate_id="c-1", version="1.0.0",
|
|
source_snapshot_ref="snap-1",
|
|
status=CandidateStatus.DRAFT,
|
|
)
|
|
data = dto.model_dump()
|
|
restored = CandidateOverviewDTO.model_validate(data)
|
|
assert restored.candidate_id == "c-1"
|
|
|
|
|
|
# ── Enums ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestCleanReleaseEnums:
|
|
"""All enums in enums.py."""
|
|
|
|
def test_candidate_status(self):
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
assert CandidateStatus.DRAFT == "DRAFT"
|
|
assert CandidateStatus.PREPARED == "PREPARED"
|
|
assert CandidateStatus.MANIFEST_BUILT == "MANIFEST_BUILT"
|
|
assert CandidateStatus.CHECK_PENDING == "CHECK_PENDING"
|
|
assert CandidateStatus.CHECK_RUNNING == "CHECK_RUNNING"
|
|
assert CandidateStatus.CHECK_PASSED == "CHECK_PASSED"
|
|
assert CandidateStatus.CHECK_BLOCKED == "CHECK_BLOCKED"
|
|
assert CandidateStatus.CHECK_ERROR == "CHECK_ERROR"
|
|
assert CandidateStatus.APPROVED == "APPROVED"
|
|
assert CandidateStatus.PUBLISHED == "PUBLISHED"
|
|
assert CandidateStatus.REVOKED == "REVOKED"
|
|
|
|
def test_run_status(self):
|
|
from src.services.clean_release.enums import RunStatus
|
|
|
|
assert RunStatus.PENDING == "PENDING"
|
|
assert RunStatus.RUNNING == "RUNNING"
|
|
assert RunStatus.SUCCEEDED == "SUCCEEDED"
|
|
assert RunStatus.FAILED == "FAILED"
|
|
assert RunStatus.CANCELLED == "CANCELLED"
|
|
|
|
def test_compliance_decision(self):
|
|
from src.services.clean_release.enums import ComplianceDecision
|
|
|
|
assert ComplianceDecision.PASSED == "PASSED"
|
|
assert ComplianceDecision.BLOCKED == "BLOCKED"
|
|
assert ComplianceDecision.ERROR == "ERROR"
|
|
|
|
def test_approval_decision_type(self):
|
|
from src.services.clean_release.enums import ApprovalDecisionType
|
|
|
|
assert ApprovalDecisionType.APPROVED == "APPROVED"
|
|
assert ApprovalDecisionType.REJECTED == "REJECTED"
|
|
|
|
def test_publication_status(self):
|
|
from src.services.clean_release.enums import PublicationStatus
|
|
|
|
assert PublicationStatus.ACTIVE == "ACTIVE"
|
|
assert PublicationStatus.REVOKED == "REVOKED"
|
|
|
|
def test_compliance_stage_name(self):
|
|
from src.services.clean_release.enums import ComplianceStageName
|
|
|
|
assert ComplianceStageName.DATA_PURITY == "DATA_PURITY"
|
|
assert ComplianceStageName.INTERNAL_SOURCES_ONLY == "INTERNAL_SOURCES_ONLY"
|
|
assert ComplianceStageName.NO_EXTERNAL_ENDPOINTS == "NO_EXTERNAL_ENDPOINTS"
|
|
assert ComplianceStageName.MANIFEST_CONSISTENCY == "MANIFEST_CONSISTENCY"
|
|
|
|
def test_classification_type(self):
|
|
from src.services.clean_release.enums import ClassificationType
|
|
|
|
assert ClassificationType.REQUIRED_SYSTEM == "required-system"
|
|
assert ClassificationType.ALLOWED == "allowed"
|
|
assert ClassificationType.EXCLUDED_PROHIBITED == "excluded-prohibited"
|
|
|
|
def test_violation_severity(self):
|
|
from src.services.clean_release.enums import ViolationSeverity
|
|
|
|
assert ViolationSeverity.CRITICAL == "CRITICAL"
|
|
assert ViolationSeverity.MAJOR == "MAJOR"
|
|
assert ViolationSeverity.MINOR == "MINOR"
|
|
|
|
def test_violation_category(self):
|
|
from src.services.clean_release.enums import ViolationCategory
|
|
|
|
assert ViolationCategory.DATA_PURITY == "DATA_PURITY"
|
|
assert ViolationCategory.SOURCE_ISOLATION == "SOURCE_ISOLATION"
|
|
assert ViolationCategory.MANIFEST_CONSISTENCY == "MANIFEST_CONSISTENCY"
|
|
assert ViolationCategory.EXTERNAL_ENDPOINT == "EXTERNAL_ENDPOINT"
|
|
|
|
|
|
# ── Exceptions ──────────────────────────────────────────────
|
|
|
|
|
|
class TestCleanReleaseExceptions:
|
|
"""Exception hierarchy for clean release."""
|
|
|
|
def test_base_exception(self):
|
|
from src.services.clean_release.exceptions import CleanReleaseError
|
|
|
|
exc = CleanReleaseError("base error")
|
|
assert str(exc) == "base error"
|
|
assert isinstance(exc, Exception)
|
|
|
|
def test_candidate_not_found(self):
|
|
from src.services.clean_release.exceptions import CandidateNotFoundError, CleanReleaseError
|
|
|
|
exc = CandidateNotFoundError("candidate not found")
|
|
assert isinstance(exc, CleanReleaseError)
|
|
assert str(exc) == "candidate not found"
|
|
|
|
def test_illegal_transition(self):
|
|
from src.services.clean_release.exceptions import IllegalTransitionError, CleanReleaseError
|
|
|
|
exc = IllegalTransitionError("forbidden transition")
|
|
assert isinstance(exc, CleanReleaseError)
|
|
|
|
def test_manifest_immutable(self):
|
|
from src.services.clean_release.exceptions import ManifestImmutableError, CleanReleaseError
|
|
|
|
exc = ManifestImmutableError("manifest is immutable")
|
|
assert isinstance(exc, CleanReleaseError)
|
|
|
|
def test_policy_resolution(self):
|
|
from src.services.clean_release.exceptions import PolicyResolutionError, CleanReleaseError
|
|
|
|
exc = PolicyResolutionError("cannot resolve policy")
|
|
assert isinstance(exc, CleanReleaseError)
|
|
|
|
def test_compliance_run(self):
|
|
from src.services.clean_release.exceptions import ComplianceRunError, CleanReleaseError
|
|
|
|
exc = ComplianceRunError("compliance run failed")
|
|
assert isinstance(exc, CleanReleaseError)
|
|
|
|
def test_approval_gate(self):
|
|
from src.services.clean_release.exceptions import ApprovalGateError, CleanReleaseError
|
|
|
|
exc = ApprovalGateError("approval not met")
|
|
assert isinstance(exc, CleanReleaseError)
|
|
|
|
def test_publication_gate(self):
|
|
from src.services.clean_release.exceptions import PublicationGateError, CleanReleaseError
|
|
|
|
exc = PublicationGateError("publication not met")
|
|
assert isinstance(exc, CleanReleaseError)
|
|
|
|
def test_all_exceptions_distinct(self):
|
|
from src.services.clean_release.exceptions import (
|
|
ApprovalGateError, CandidateNotFoundError, CleanReleaseError,
|
|
ComplianceRunError, IllegalTransitionError, ManifestImmutableError,
|
|
PolicyResolutionError, PublicationGateError,
|
|
)
|
|
|
|
assert CandidateNotFoundError is not CleanReleaseError
|
|
assert IllegalTransitionError is not CleanReleaseError
|
|
assert ManifestImmutableError is not CleanReleaseError
|
|
assert PolicyResolutionError is not CleanReleaseError
|
|
assert ComplianceRunError is not CleanReleaseError
|
|
assert ApprovalGateError is not CleanReleaseError
|
|
assert PublicationGateError is not CleanReleaseError
|
|
# #endregion Test.CleanRelease.DtoEnumsExceptions
|