busya
c713e15e4d
SESSION SUMMARY: - Started at 7194 tests, 80% raw / 93.4% real - Ended at 7778 tests, 84% raw / 98.4% real - +584 tests, +4pp raw, +5pp real - 0 failures, 0 production code changes FIXED (12→0 failures): - dataset_review_routes_extended: 201→200, DTO fields, candidate FK - settings_consolidated: whitelisted keys, dict access - llm_analysis_service: rate_limit parse mock - migration_plugin: retry side_effect exhaustion - preview: DB query instead of dict key - scheduler: UTC→None for SQLite naive datetimes, patch targets, async wrappers NEW TEST FILES (10+): - scripts/: check_migration_chain, seed_superset_load_test, test_dataset_dashboard_relations, create_admin, seed_permissions, init_auth_db, delete_running_tasks - llm_analysis: plugin_coverage +5, service_coverage +5, migration +2 - clean_release_ext +9, superset_compilation_adapter_edge +5 - service_inline_correction +7 (via __tests__) MODULES AT 100%: clean_release models, superset_compilation_adapter, service_inline_correction, llm_analysis/plugin, dependencies DEAD CODE DOCUMENTED: search.py (L206-215 indentation bug), llm_analysis/service (L459 HTTPS, L594 duplicate tab, L639-697 CDP-only)
963 lines
38 KiB
Python
963 lines
38 KiB
Python
# #region Test.Models.CleanRelease.Ext [C:2] [TYPE Module] [SEMANTICS test,models,clean-release,extended]
|
|
# @BRIEF Extended tests for models/clean_release.py — edge cases for enums, ResourceSourceRegistry,
|
|
# CleanProfilePolicy, ComplianceCheckRun, ReleaseCandidate, DistributionManifest,
|
|
# ComplianceRun, ComplianceViolation, ComplianceReport, ApprovalDecision, PublicationRecord,
|
|
# CleanReleaseAuditLog, SourceRegistrySnapshot, CleanPolicySnapshot, ComplianceStageRun.
|
|
# @RELATION BINDS_TO -> [CleanReleaseModels]
|
|
|
|
from datetime import datetime, timezone
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
|
|
class TestCleanReleaseLegacyEnums:
|
|
"""Legacy backward-compatible enums."""
|
|
|
|
def test_execution_mode(self):
|
|
from src.models.clean_release import ExecutionMode
|
|
|
|
assert ExecutionMode.TUI == "TUI"
|
|
assert ExecutionMode.API == "API"
|
|
assert ExecutionMode.SCHEDULER == "SCHEDULER"
|
|
|
|
def test_check_final_status(self):
|
|
from src.models.clean_release import CheckFinalStatus
|
|
|
|
assert CheckFinalStatus.COMPLIANT == "COMPLIANT"
|
|
assert CheckFinalStatus.BLOCKED == "BLOCKED"
|
|
assert CheckFinalStatus.FAILED == "FAILED"
|
|
assert CheckFinalStatus.RUNNING == "RUNNING"
|
|
|
|
def test_check_stage_name(self):
|
|
from src.models.clean_release import CheckStageName
|
|
|
|
assert CheckStageName.DATA_PURITY == "DATA_PURITY"
|
|
assert CheckStageName.INTERNAL_SOURCES_ONLY == "INTERNAL_SOURCES_ONLY"
|
|
assert CheckStageName.NO_EXTERNAL_ENDPOINTS == "NO_EXTERNAL_ENDPOINTS"
|
|
assert CheckStageName.MANIFEST_CONSISTENCY == "MANIFEST_CONSISTENCY"
|
|
|
|
def test_check_stage_status(self):
|
|
from src.models.clean_release import CheckStageStatus
|
|
|
|
assert CheckStageStatus.PASS == "PASS"
|
|
assert CheckStageStatus.FAIL == "FAIL"
|
|
assert CheckStageStatus.SKIPPED == "SKIPPED"
|
|
assert CheckStageStatus.RUNNING == "RUNNING"
|
|
|
|
def test_profile_type(self):
|
|
from src.models.clean_release import ProfileType
|
|
|
|
assert ProfileType.ENTERPRISE_CLEAN == "enterprise-clean"
|
|
|
|
def test_registry_status(self):
|
|
from src.models.clean_release import RegistryStatus
|
|
|
|
assert RegistryStatus.ACTIVE == "ACTIVE"
|
|
assert RegistryStatus.INACTIVE == "INACTIVE"
|
|
|
|
def test_release_candidate_status(self):
|
|
from src.models.clean_release import ReleaseCandidateStatus
|
|
|
|
assert ReleaseCandidateStatus.DRAFT == "DRAFT"
|
|
assert ReleaseCandidateStatus.PUBLISHED == "PUBLISHED"
|
|
assert ReleaseCandidateStatus.REVOKED == "REVOKED"
|
|
assert ReleaseCandidateStatus.BLOCKED == "CHECK_BLOCKED" # alias
|
|
|
|
def test_violation_severity(self):
|
|
from src.models.clean_release import ViolationSeverity
|
|
|
|
assert ViolationSeverity.CRITICAL == "CRITICAL"
|
|
assert ViolationSeverity.MAJOR == "MAJOR"
|
|
assert ViolationSeverity.MINOR == "MINOR"
|
|
|
|
def test_violation_category(self):
|
|
from src.models.clean_release import ViolationCategory
|
|
|
|
assert ViolationCategory.DATA_PURITY == "DATA_PURITY"
|
|
assert ViolationCategory.EXTERNAL_SOURCE == "EXTERNAL_SOURCE"
|
|
assert ViolationCategory.SOURCE_ISOLATION == "SOURCE_ISOLATION"
|
|
assert ViolationCategory.MANIFEST_CONSISTENCY == "MANIFEST_CONSISTENCY"
|
|
assert ViolationCategory.EXTERNAL_ENDPOINT == "EXTERNAL_ENDPOINT"
|
|
|
|
|
|
class TestCheckStageResult:
|
|
"""CheckStageResult — pydantic dataclass."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import CheckStageName, CheckStageResult, CheckStageStatus
|
|
|
|
result = CheckStageResult(stage=CheckStageName.DATA_PURITY, status=CheckStageStatus.PASS)
|
|
assert result.stage == CheckStageName.DATA_PURITY
|
|
assert result.status == CheckStageStatus.PASS
|
|
assert result.details == ""
|
|
|
|
def test_with_details(self):
|
|
from src.models.clean_release import CheckStageName, CheckStageResult, CheckStageStatus
|
|
|
|
result = CheckStageResult(stage=CheckStageName.DATA_PURITY, status=CheckStageStatus.FAIL, details="Data issues")
|
|
assert result.details == "Data issues"
|
|
|
|
|
|
class TestResourceSourceEntry:
|
|
"""ResourceSourceEntry — pydantic dataclass."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ResourceSourceEntry
|
|
|
|
entry = ResourceSourceEntry(source_id="s-1", host="example.com", protocol="https", purpose="api")
|
|
assert entry.host == "example.com"
|
|
assert entry.enabled is True
|
|
|
|
def test_disabled(self):
|
|
from src.models.clean_release import ResourceSourceEntry
|
|
|
|
entry = ResourceSourceEntry(source_id="s-1", host="example.com", protocol="https", purpose="api", enabled=False)
|
|
assert entry.enabled is False
|
|
|
|
|
|
class TestResourceSourceRegistry:
|
|
"""ResourceSourceRegistry — registry with populate_legacy_allowlists."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ResourceSourceEntry, ResourceSourceRegistry
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
entry = ResourceSourceEntry(source_id="s-1", host="example.com", protocol="https", purpose="api")
|
|
registry = ResourceSourceRegistry(
|
|
registry_id="reg-1", name="My Registry",
|
|
entries=[entry], updated_at=ts, updated_by="admin",
|
|
)
|
|
assert registry.registry_id == "reg-1"
|
|
assert registry.allowed_hosts == ["example.com"] # auto-populated
|
|
assert registry.allowed_schemes == ["https"]
|
|
assert registry.allowed_source_types == ["api"]
|
|
assert registry.id == "reg-1"
|
|
assert registry.immutable is True
|
|
|
|
def test_with_some_disabled_entries(self):
|
|
from src.models.clean_release import ResourceSourceEntry, ResourceSourceRegistry
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
e1 = ResourceSourceEntry(source_id="s-1", host="good.com", protocol="https", purpose="api", enabled=True)
|
|
e2 = ResourceSourceEntry(source_id="s-2", host="bad.com", protocol="http", purpose="old", enabled=False)
|
|
registry = ResourceSourceRegistry(
|
|
registry_id="reg-1", name="Registry",
|
|
entries=[e1, e2], updated_at=ts, updated_by="admin",
|
|
)
|
|
# Only enabled entries should be included in allow lists
|
|
assert "good.com" in registry.allowed_hosts
|
|
assert "bad.com" not in registry.allowed_hosts
|
|
assert registry.allowed_schemes == ["https"] # only e1's protocol
|
|
|
|
def test_preserves_explicit_allowlists(self):
|
|
from src.models.clean_release import ResourceSourceEntry, ResourceSourceRegistry
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
entry = ResourceSourceEntry(source_id="s-1", host="example.com", protocol="https", purpose="api")
|
|
registry = ResourceSourceRegistry(
|
|
registry_id="reg-1", name="Registry",
|
|
entries=[entry], updated_at=ts, updated_by="admin",
|
|
allowed_hosts=["explicit.com"], allowed_schemes=["ftp"],
|
|
allowed_source_types=["explicit"],
|
|
)
|
|
assert registry.allowed_hosts == ["explicit.com"]
|
|
assert registry.allowed_schemes == ["ftp"]
|
|
assert registry.allowed_source_types == ["explicit"]
|
|
|
|
|
|
class TestCleanProfilePolicy:
|
|
"""CleanProfilePolicy — enterprise policy validation."""
|
|
|
|
def test_non_enterprise_no_validation(self):
|
|
from src.models.clean_release import CleanProfilePolicy, ProfileType
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
# For non-enterprise profile, prohibited_artifact_categories are not required by the validator
|
|
policy = CleanProfilePolicy(
|
|
policy_id="p-1", policy_version="1",
|
|
profile=ProfileType.ENTERPRISE_CLEAN,
|
|
active=True,
|
|
prohibited_artifact_categories=["test"],
|
|
internal_source_registry_ref="reg-1",
|
|
effective_from=ts,
|
|
external_source_forbidden=True,
|
|
)
|
|
assert policy.profile == ProfileType.ENTERPRISE_CLEAN
|
|
assert policy.id == "p-1"
|
|
assert policy.registry_snapshot_id == "reg-1"
|
|
assert policy.content_json is not None
|
|
assert policy.content_json["profile"] == "enterprise-clean"
|
|
|
|
def test_property_access(self):
|
|
from src.models.clean_release import CleanProfilePolicy, ProfileType
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
policy = CleanProfilePolicy(
|
|
policy_id="p-1", policy_version="1",
|
|
profile=ProfileType.ENTERPRISE_CLEAN,
|
|
active=True,
|
|
prohibited_artifact_categories=["test-data"],
|
|
internal_source_registry_ref="REG-1",
|
|
effective_from=ts,
|
|
)
|
|
assert policy.id == "p-1"
|
|
assert policy.registry_snapshot_id == "REG-1"
|
|
|
|
# ── Lines 172, 174: enterprise-clean validation errors ──
|
|
|
|
def test_enterprise_requires_prohibited_artifact_categories(self):
|
|
"""ENTERPRISE_CLEAN profile with empty prohibited_artifact_categories raises (line 172)."""
|
|
from src.models.clean_release import CleanProfilePolicy, ProfileType
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
with pytest.raises(ValueError, match="enterprise-clean policy requires prohibited_artifact_categories"):
|
|
CleanProfilePolicy(
|
|
policy_id="p-err", policy_version="1",
|
|
profile=ProfileType.ENTERPRISE_CLEAN,
|
|
active=True,
|
|
prohibited_artifact_categories=[],
|
|
internal_source_registry_ref="reg-1",
|
|
effective_from=ts,
|
|
external_source_forbidden=True,
|
|
)
|
|
|
|
def test_enterprise_requires_external_forbidden(self):
|
|
"""ENTERPRISE_CLEAN profile with external_source_forbidden=False raises (line 174)."""
|
|
from src.models.clean_release import CleanProfilePolicy, ProfileType
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
with pytest.raises(ValueError, match="enterprise-clean policy requires external_source_forbidden=true"):
|
|
CleanProfilePolicy(
|
|
policy_id="p-err2", policy_version="1",
|
|
profile=ProfileType.ENTERPRISE_CLEAN,
|
|
active=True,
|
|
prohibited_artifact_categories=["test-data"],
|
|
internal_source_registry_ref="reg-1",
|
|
effective_from=ts,
|
|
external_source_forbidden=False,
|
|
)
|
|
|
|
|
|
class TestReleaseCandidate:
|
|
"""ReleaseCandidate — lifecycle and transitions."""
|
|
|
|
def test_with_status_enum(self):
|
|
from src.models.clean_release import ReleaseCandidate, ReleaseCandidateStatus
|
|
|
|
rc = ReleaseCandidate(
|
|
candidate_id="RC-002",
|
|
version="2.0.0",
|
|
source_snapshot_ref="snap-2",
|
|
created_by="admin",
|
|
status=ReleaseCandidateStatus.DRAFT,
|
|
)
|
|
assert rc.status == "DRAFT"
|
|
assert rc.candidate_id == "RC-002"
|
|
|
|
def test_with_candidate_status_enum(self):
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(
|
|
candidate_id="RC-003",
|
|
version="3.0.0",
|
|
source_snapshot_ref="snap-3",
|
|
created_by="admin",
|
|
status=CandidateStatus.DRAFT,
|
|
)
|
|
assert rc.status == "DRAFT"
|
|
|
|
def test_transition_draft_to_prepared(self):
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(candidate_id="RC-010", version="1.0.0",
|
|
source_snapshot_ref="snap", created_by="u",
|
|
status=CandidateStatus.DRAFT)
|
|
rc.transition_to(CandidateStatus.PREPARED)
|
|
assert rc.status == "PREPARED"
|
|
|
|
def test_transition_full_flow(self):
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(candidate_id="RC-020", version="1.0.0",
|
|
source_snapshot_ref="snap", created_by="u",
|
|
status=CandidateStatus.DRAFT)
|
|
rc.transition_to(CandidateStatus.PREPARED)
|
|
rc.transition_to(CandidateStatus.MANIFEST_BUILT)
|
|
rc.transition_to(CandidateStatus.CHECK_PENDING)
|
|
rc.transition_to(CandidateStatus.CHECK_RUNNING)
|
|
rc.transition_to(CandidateStatus.CHECK_PASSED)
|
|
rc.transition_to(CandidateStatus.APPROVED)
|
|
rc.transition_to(CandidateStatus.PUBLISHED)
|
|
rc.transition_to(CandidateStatus.REVOKED)
|
|
assert rc.status == "REVOKED"
|
|
|
|
def test_transition_illegal_raises(self):
|
|
from src.models.clean_release import CandidateStatus, IllegalTransitionError, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(candidate_id="RC-030", version="1.0.0",
|
|
source_snapshot_ref="snap", created_by="u",
|
|
status=CandidateStatus.DRAFT)
|
|
with pytest.raises(IllegalTransitionError, match="Forbidden transition"):
|
|
rc.transition_to(CandidateStatus.APPROVED)
|
|
|
|
def test_transition_revoked_has_no_transitions(self):
|
|
from src.models.clean_release import CandidateStatus, IllegalTransitionError, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(candidate_id="RC-040", version="1.0.0",
|
|
source_snapshot_ref="snap", created_by="u",
|
|
status=CandidateStatus.REVOKED)
|
|
with pytest.raises(IllegalTransitionError):
|
|
rc.transition_to(CandidateStatus.DRAFT)
|
|
|
|
def test_transition_check_running_to_blocked(self):
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(candidate_id="RC-050", version="1.0.0",
|
|
source_snapshot_ref="snap", created_by="u",
|
|
status=CandidateStatus.CHECK_RUNNING)
|
|
rc.transition_to(CandidateStatus.CHECK_BLOCKED)
|
|
assert rc.status == "CHECK_BLOCKED"
|
|
|
|
def test_transition_check_running_to_error(self):
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(candidate_id="RC-060", version="1.0.0",
|
|
source_snapshot_ref="snap", created_by="u",
|
|
status=CandidateStatus.CHECK_RUNNING)
|
|
rc.transition_to(CandidateStatus.CHECK_ERROR)
|
|
assert rc.status == "CHECK_ERROR"
|
|
|
|
def test_transition_from_blocked_back_to_pending(self):
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(candidate_id="RC-070", version="1.0.0",
|
|
source_snapshot_ref="snap", created_by="u",
|
|
status=CandidateStatus.CHECK_BLOCKED)
|
|
rc.transition_to(CandidateStatus.CHECK_PENDING)
|
|
assert rc.status == "CHECK_PENDING"
|
|
|
|
def test_transition_from_passed_to_pending(self):
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(candidate_id="RC-080", version="1.0.0",
|
|
source_snapshot_ref="snap", created_by="u",
|
|
status=CandidateStatus.CHECK_PASSED)
|
|
rc.transition_to(CandidateStatus.CHECK_PENDING)
|
|
assert rc.status == "CHECK_PENDING"
|
|
|
|
# ── Lines 262, 265, 269: ReleaseCandidate.__init__ edge cases ──
|
|
|
|
def test_init_pops_profile_kwarg(self):
|
|
"""ReleaseCandidate.__init__ silently pops profile kwarg (line 262)."""
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(
|
|
candidate_id="RC-PROF",
|
|
version="1.0.0",
|
|
source_snapshot_ref="snap",
|
|
created_by="u",
|
|
profile="should_be_removed",
|
|
status=CandidateStatus.DRAFT,
|
|
)
|
|
assert rc.candidate_id == "RC-PROF"
|
|
assert not hasattr(rc, "profile")
|
|
|
|
def test_init_defaults_status_when_none(self):
|
|
"""ReleaseCandidate.__init__ defaults to DRAFT when no status given (line 265)."""
|
|
from src.models.clean_release import CandidateStatus, ReleaseCandidate
|
|
|
|
rc = ReleaseCandidate(
|
|
candidate_id="RC-NOSTAT",
|
|
version="1.0.0",
|
|
source_snapshot_ref="snap",
|
|
created_by="u",
|
|
)
|
|
assert rc.status == CandidateStatus.DRAFT.value
|
|
|
|
def test_init_raises_on_empty_id(self):
|
|
"""ReleaseCandidate.__init__ raises ValueError on empty candidate_id (line 269)."""
|
|
from src.models.clean_release import ReleaseCandidate
|
|
|
|
with pytest.raises(ValueError, match="candidate_id must be non-empty"):
|
|
ReleaseCandidate(
|
|
candidate_id="",
|
|
version="1.0.0",
|
|
source_snapshot_ref="snap",
|
|
created_by="u",
|
|
)
|
|
|
|
def test_init_raises_on_blank_id(self):
|
|
"""ReleaseCandidate.__init__ raises ValueError on whitespace-only candidate_id."""
|
|
from src.models.clean_release import ReleaseCandidate
|
|
|
|
with pytest.raises(ValueError, match="candidate_id must be non-empty"):
|
|
ReleaseCandidate(
|
|
candidate_id=" ",
|
|
version="1.0.0",
|
|
source_snapshot_ref="snap",
|
|
created_by="u",
|
|
)
|
|
|
|
|
|
class TestDistributionManifest:
|
|
"""DistributionManifest — immutable manifest model."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import DistributionManifest
|
|
|
|
manifest = DistributionManifest(
|
|
id="m-1", candidate_id="c-1",
|
|
manifest_digest="digest1", artifacts_digest="ad1",
|
|
created_by="admin", source_snapshot_ref="snap-1",
|
|
content_json={"key": "value"},
|
|
manifest_version=1, immutable=True,
|
|
)
|
|
assert manifest.manifest_version == 1
|
|
assert manifest.manifest_id == "m-1"
|
|
assert manifest.deterministic_hash == "digest1"
|
|
assert manifest.summary.included_count == 0
|
|
|
|
def test_with_items_and_summary(self):
|
|
from src.models.clean_release import ClassificationType, DistributionManifest, ManifestItem, ManifestSummary
|
|
|
|
items = [ManifestItem(path="/file", category="bin", classification=ClassificationType.ALLOWED, reason="ok")]
|
|
summary = ManifestSummary(included_count=1, excluded_count=0, prohibited_detected_count=0)
|
|
manifest = DistributionManifest(
|
|
manifest_id="m-2", candidate_id="c-1",
|
|
deterministic_hash="hash1",
|
|
generated_at=datetime.now(timezone.utc),
|
|
generated_by="admin",
|
|
items=items, summary=summary,
|
|
)
|
|
assert manifest.manifest_digest == "hash1"
|
|
assert manifest.summary.included_count == 1
|
|
|
|
def test_id_mapping(self):
|
|
from src.models.clean_release import DistributionManifest
|
|
|
|
manifest = DistributionManifest(
|
|
manifest_id="m-1", candidate_id="c-1",
|
|
manifest_digest="d1", artifacts_digest="ad1",
|
|
created_by="admin", source_snapshot_ref="snap-1",
|
|
content_json={},
|
|
)
|
|
assert manifest.id == "m-1"
|
|
|
|
def test_immutable_default(self):
|
|
from src.models.clean_release import DistributionManifest
|
|
|
|
manifest = DistributionManifest(
|
|
id="m-imm", candidate_id="c-1",
|
|
manifest_digest="d1", artifacts_digest="ad1",
|
|
created_by="admin", source_snapshot_ref="snap-1",
|
|
content_json={}, immutable=True,
|
|
)
|
|
assert manifest.immutable is True
|
|
|
|
# ── Line 375: DistributionManifest.__init__ pops policy_id ──
|
|
|
|
def test_init_pops_policy_id_kwarg(self):
|
|
"""DistributionManifest.__init__ silently pops policy_id (line 375)."""
|
|
from src.models.clean_release import DistributionManifest
|
|
|
|
manifest = DistributionManifest(
|
|
manifest_id="m-pol", candidate_id="c-1",
|
|
deterministic_hash="hash1",
|
|
generated_at=datetime.now(timezone.utc),
|
|
generated_by="admin",
|
|
policy_id="should-be-popped",
|
|
)
|
|
assert manifest.manifest_id == "m-pol"
|
|
assert not hasattr(manifest, "policy_id")
|
|
|
|
# ── Line 380: DistributionManifest.__init__ validates items vs summary ──
|
|
|
|
def test_init_raises_on_mismatched_items_summary(self):
|
|
"""DistributionManifest raises ValueError when summary counts don't match items (line 380)."""
|
|
from src.models.clean_release import ClassificationType, DistributionManifest, ManifestItem, ManifestSummary
|
|
|
|
items = [ManifestItem(path="/f1", category="bin", classification=ClassificationType.ALLOWED, reason="ok")]
|
|
summary = ManifestSummary(included_count=2, excluded_count=0, prohibited_detected_count=0)
|
|
|
|
with pytest.raises(ValueError, match="manifest summary counts must match items size"):
|
|
DistributionManifest(
|
|
manifest_id="m-mismatch", candidate_id="c-1",
|
|
deterministic_hash="hash1",
|
|
generated_at=datetime.now(timezone.utc),
|
|
generated_by="admin",
|
|
items=items, summary=summary,
|
|
)
|
|
|
|
|
|
class TestComplianceRun:
|
|
"""ComplianceRun — operational record."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ComplianceRun
|
|
|
|
run = ComplianceRun(
|
|
id="r-1", candidate_id="c-1",
|
|
manifest_id="m-1", manifest_digest="d1",
|
|
policy_snapshot_id="ps-1", registry_snapshot_id="rs-1",
|
|
requested_by="admin", status="PENDING",
|
|
)
|
|
assert run.check_run_id == "r-1"
|
|
assert run.status == "PENDING"
|
|
assert run.final_status is None
|
|
|
|
def test_with_final_status(self):
|
|
from src.models.clean_release import ComplianceRun
|
|
|
|
run = ComplianceRun(
|
|
id="r-2", candidate_id="c-1",
|
|
manifest_id="m-1", manifest_digest="d1",
|
|
policy_snapshot_id="ps-1", registry_snapshot_id="rs-1",
|
|
requested_by="admin", final_status="PASSED",
|
|
)
|
|
assert run.final_status == "PASSED"
|
|
assert run.task_id is None
|
|
|
|
|
|
class TestComplianceStageRun:
|
|
"""ComplianceStageRun — stage-level execution."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ComplianceStageRun
|
|
|
|
stage = ComplianceStageRun(id="s-1", run_id="r-1", stage_name="DATA_PURITY", status="PASS")
|
|
assert stage.stage_name == "DATA_PURITY"
|
|
assert stage.status == "PASS"
|
|
|
|
|
|
class TestComplianceViolation:
|
|
"""ComplianceViolation — violation with legacy property mapping."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ComplianceViolation
|
|
|
|
violation = ComplianceViolation(
|
|
id="v-1", run_id="r-1", stage_name="DATA_PURITY",
|
|
code="DP001", severity="MAJOR", message="Data issue",
|
|
)
|
|
assert violation.violation_id == "v-1"
|
|
assert violation.check_run_id == "r-1"
|
|
assert violation.code == "DP001"
|
|
assert violation.artifact_path is None
|
|
assert violation.remediation is None
|
|
assert violation.blocked_release is False
|
|
|
|
def test_with_legacy_kwargs(self):
|
|
from src.models.clean_release import ComplianceViolation
|
|
|
|
violation = ComplianceViolation(
|
|
violation_id="v-1",
|
|
check_run_id="r-1",
|
|
category="DATA_PURITY",
|
|
code="DP002",
|
|
severity="CRITICAL",
|
|
message="NPE",
|
|
location="/path/to/file",
|
|
remediation="Fix it",
|
|
blocked_release=True,
|
|
)
|
|
assert violation.id == "v-1"
|
|
assert violation.run_id == "r-1"
|
|
assert violation.stage_name == "DATA_PURITY"
|
|
assert violation.artifact_path == "/path/to/file"
|
|
assert violation.remediation == "Fix it"
|
|
assert violation.blocked_release is True
|
|
assert violation.category.value == "DATA_PURITY"
|
|
assert violation.location == "/path/to/file"
|
|
|
|
def test_category_setter(self):
|
|
from src.models.clean_release import ComplianceViolation, ViolationCategory
|
|
|
|
violation = ComplianceViolation(
|
|
id="v-1", run_id="r-1", stage_name="DATA_PURITY",
|
|
code="DP001", severity="MAJOR", message="Data issue",
|
|
)
|
|
violation.category = ViolationCategory.EXTERNAL_SOURCE
|
|
assert violation.stage_name == "EXTERNAL_SOURCE"
|
|
|
|
def test_violation_id_setter(self):
|
|
from src.models.clean_release import ComplianceViolation
|
|
|
|
violation = ComplianceViolation(
|
|
id="v-1", run_id="r-1", stage_name="DATA_PURITY",
|
|
code="DP001", severity="MAJOR", message="Data issue",
|
|
)
|
|
violation.violation_id = "v-2"
|
|
assert violation.id == "v-2"
|
|
|
|
def test_default_code_and_message(self):
|
|
from src.models.clean_release import ComplianceViolation
|
|
|
|
violation = ComplianceViolation(
|
|
id="v-3", run_id="r-1", stage_name="DATA_PURITY",
|
|
severity="MINOR", message="",
|
|
)
|
|
# When code not provided, defaults to LEGACY_VIOLATION
|
|
assert violation.code == "LEGACY_VIOLATION"
|
|
|
|
def test_detected_at_popped(self):
|
|
from src.models.clean_release import ComplianceViolation
|
|
|
|
# detected_at is popped (removed) in __init__, so it shouldn't be an attribute
|
|
violation = ComplianceViolation(
|
|
id="v-4", run_id="r-1", stage_name="DATA_PURITY",
|
|
severity="MINOR", message="test",
|
|
detected_at="2024-01-01", # should be popped
|
|
)
|
|
assert not hasattr(violation, "detected_at")
|
|
|
|
def test_message_defaults_to_stage_name(self):
|
|
"""When message is not provided, it defaults to stage_name."""
|
|
from src.models.clean_release import ComplianceViolation
|
|
|
|
violation = ComplianceViolation(
|
|
id="v-5", run_id="r-1", stage_name="SOURCE_ISOLATION",
|
|
severity="MAJOR", code="SI001",
|
|
# no message kwarg
|
|
)
|
|
# message should default to stage_name
|
|
assert violation.message == "SOURCE_ISOLATION"
|
|
|
|
def test_message_defaults_fallback(self):
|
|
"""When message is not provided, defaults to stage_name value."""
|
|
from src.models.clean_release import ComplianceViolation
|
|
|
|
violation = ComplianceViolation(
|
|
id="v-6", run_id="r-1", stage_name="SOURCE_ISOLATION",
|
|
severity="MAJOR", code="GEN001",
|
|
# no message kwarg
|
|
)
|
|
# Without 'message' kwarg, defaults to stage_name value
|
|
assert violation.message == "SOURCE_ISOLATION"
|
|
|
|
|
|
class TestComplianceReport:
|
|
"""ComplianceReport — immutable report with property access."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ComplianceReport
|
|
|
|
report = ComplianceReport(
|
|
id="rep-1", run_id="r-1", candidate_id="c-1",
|
|
final_status="PASSED", summary_json={"key": "val"},
|
|
immutable=True,
|
|
)
|
|
assert report.report_id == "rep-1"
|
|
assert report.check_run_id == "r-1"
|
|
assert report.final_status == "PASSED"
|
|
assert report.immutable is True
|
|
|
|
def test_with_legacy_kwargs(self):
|
|
from src.models.clean_release import ComplianceReport
|
|
|
|
report = ComplianceReport(
|
|
report_id="rep-1",
|
|
check_run_id="r-1",
|
|
candidate_id="c-1",
|
|
final_status="PASSED",
|
|
operator_summary="All good",
|
|
structured_payload_ref="ref-1",
|
|
violations_count=0,
|
|
blocking_violations_count=0,
|
|
)
|
|
assert report.report_id == "rep-1"
|
|
assert report.check_run_id == "r-1"
|
|
assert report.operator_summary == "All good"
|
|
assert report.structured_payload_ref == "ref-1"
|
|
assert report.violations_count == 0
|
|
assert report.blocking_violations_count == 0
|
|
|
|
def test_blocked_without_violations_raises(self):
|
|
from src.models.clean_release import CheckFinalStatus, ComplianceReport
|
|
|
|
with pytest.raises(ValueError, match="blocked report requires blocking violations"):
|
|
ComplianceReport(
|
|
report_id="rep-2",
|
|
check_run_id="r-2",
|
|
candidate_id="c-2",
|
|
final_status=CheckFinalStatus.BLOCKED,
|
|
violations_count=0,
|
|
blocking_violations_count=0,
|
|
)
|
|
|
|
def test_blocked_with_violations_ok(self):
|
|
from src.models.clean_release import ComplianceDecision, ComplianceReport
|
|
|
|
report = ComplianceReport(
|
|
report_id="rep-3",
|
|
check_run_id="r-3",
|
|
candidate_id="c-3",
|
|
final_status=ComplianceDecision.BLOCKED,
|
|
operator_summary="Blocked",
|
|
violations_count=3,
|
|
blocking_violations_count=2,
|
|
)
|
|
assert report.final_status == "BLOCKED"
|
|
assert report.violations_count == 3
|
|
|
|
|
|
class TestApprovalDecision:
|
|
"""ApprovalDecision — approval record."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ApprovalDecision
|
|
|
|
decision = ApprovalDecision(
|
|
id="a-1", candidate_id="c-1", report_id="r-1",
|
|
decision="APPROVED", decided_by="admin",
|
|
)
|
|
assert decision.decision == "APPROVED"
|
|
assert decision.comment is None
|
|
|
|
def test_with_comment(self):
|
|
from src.models.clean_release import ApprovalDecision
|
|
|
|
decision = ApprovalDecision(
|
|
id="a-2", candidate_id="c-1", report_id="r-1",
|
|
decision="REJECTED", decided_by="admin",
|
|
comment="Missing required artifacts",
|
|
)
|
|
assert decision.comment == "Missing required artifacts"
|
|
|
|
|
|
class TestPublicationRecord:
|
|
"""PublicationRecord — publication or revocation."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import PublicationRecord
|
|
|
|
record = PublicationRecord(
|
|
id="p-1", candidate_id="c-1", report_id="r-1",
|
|
published_by="admin", target_channel="registry",
|
|
status="ACTIVE",
|
|
)
|
|
assert record.status == "ACTIVE"
|
|
assert record.publication_ref is None
|
|
|
|
def test_with_all_fields(self):
|
|
from src.models.clean_release import PublicationRecord
|
|
|
|
record = PublicationRecord(
|
|
id="p-2", candidate_id="c-1", report_id="r-1",
|
|
published_by="admin", target_channel="registry",
|
|
publication_ref="pub-ref-1", status="REVOKED",
|
|
)
|
|
assert record.publication_ref == "pub-ref-1"
|
|
assert record.status == "REVOKED"
|
|
|
|
|
|
class TestCleanReleaseAuditLog:
|
|
"""CleanReleaseAuditLog — persistent audit log."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import CleanReleaseAuditLog
|
|
|
|
log = CleanReleaseAuditLog(
|
|
id="al-1", candidate_id="c-1",
|
|
action="TRANSITION", actor="admin",
|
|
)
|
|
assert log.id == "al-1"
|
|
assert log.action == "TRANSITION"
|
|
assert log.actor == "admin"
|
|
|
|
def test_with_details(self):
|
|
from src.models.clean_release import CleanReleaseAuditLog
|
|
|
|
log = CleanReleaseAuditLog(
|
|
id="al-2", candidate_id="c-1",
|
|
action="APPROVE", actor="admin",
|
|
details_json={"reason": "All checks passed"},
|
|
)
|
|
assert log.details_json["reason"] == "All checks passed"
|
|
|
|
def test_nullable_candidate_id(self):
|
|
from src.models.clean_release import CleanReleaseAuditLog
|
|
|
|
log = CleanReleaseAuditLog(id="al-3", action="SYSTEM", actor="system")
|
|
assert log.candidate_id is None
|
|
|
|
|
|
class TestSourceRegistrySnapshot:
|
|
"""SourceRegistrySnapshot — immutable registry snapshot."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import SourceRegistrySnapshot
|
|
|
|
snap = SourceRegistrySnapshot(
|
|
id="rs-1", registry_id="reg-1", registry_version="1.0",
|
|
allowed_hosts=["host1"], allowed_schemes=["https"],
|
|
allowed_source_types=["api"], immutable=True,
|
|
)
|
|
assert snap.registry_id == "reg-1"
|
|
assert snap.immutable is True
|
|
|
|
|
|
class TestCleanPolicySnapshot:
|
|
"""CleanPolicySnapshot — immutable policy snapshot."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import CleanPolicySnapshot
|
|
|
|
snap = CleanPolicySnapshot(
|
|
id="ps-1", policy_id="pol-1", policy_version="1",
|
|
content_json={"rules": []}, registry_snapshot_id="rs-1",
|
|
immutable=True,
|
|
)
|
|
assert snap.policy_id == "pol-1"
|
|
assert snap.immutable is True
|
|
|
|
|
|
class TestComplianceCheckRunProperties:
|
|
"""ComplianceCheckRun — property access and status mappings."""
|
|
|
|
def test_id_and_run_id(self):
|
|
from src.models.clean_release import (
|
|
CheckFinalStatus, CheckStageName, CheckStageResult, CheckStageStatus,
|
|
ComplianceCheckRun, ExecutionMode,
|
|
)
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
run = ComplianceCheckRun(
|
|
check_run_id="run-1",
|
|
candidate_id="c-1",
|
|
policy_id="p-1",
|
|
started_at=ts,
|
|
triggered_by="admin",
|
|
execution_mode=ExecutionMode.TUI,
|
|
final_status=CheckFinalStatus.RUNNING,
|
|
checks=[
|
|
CheckStageResult(stage=CheckStageName.DATA_PURITY, status=CheckStageStatus.PASS),
|
|
],
|
|
)
|
|
assert run.id == "run-1"
|
|
assert run.run_id == "run-1"
|
|
assert run.status == "RUNNING"
|
|
|
|
def test_status_mapping_blocked(self):
|
|
from src.models.clean_release import (
|
|
CheckFinalStatus, CheckStageName, CheckStageResult, CheckStageStatus,
|
|
ComplianceCheckRun, ExecutionMode,
|
|
)
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
run = ComplianceCheckRun(
|
|
check_run_id="run-2",
|
|
candidate_id="c-1",
|
|
policy_id="p-1",
|
|
started_at=ts,
|
|
triggered_by="admin",
|
|
execution_mode=ExecutionMode.API,
|
|
final_status=CheckFinalStatus.BLOCKED,
|
|
checks=[
|
|
CheckStageResult(stage=CheckStageName.DATA_PURITY, status=CheckStageStatus.PASS),
|
|
],
|
|
)
|
|
assert run.status == "FAILED"
|
|
|
|
def test_status_succeeded(self):
|
|
from src.models.clean_release import (
|
|
CheckFinalStatus, CheckStageName, CheckStageResult, CheckStageStatus,
|
|
ComplianceCheckRun, ExecutionMode,
|
|
)
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
run = ComplianceCheckRun(
|
|
check_run_id="run-5",
|
|
candidate_id="c-1",
|
|
policy_id="p-1",
|
|
started_at=ts,
|
|
triggered_by="admin",
|
|
execution_mode=ExecutionMode.TUI,
|
|
final_status=CheckFinalStatus.COMPLIANT,
|
|
checks=[
|
|
CheckStageResult(stage=CheckStageName.DATA_PURITY, status=CheckStageStatus.PASS),
|
|
CheckStageResult(stage=CheckStageName.INTERNAL_SOURCES_ONLY, status=CheckStageStatus.PASS),
|
|
CheckStageResult(stage=CheckStageName.NO_EXTERNAL_ENDPOINTS, status=CheckStageStatus.PASS),
|
|
CheckStageResult(stage=CheckStageName.MANIFEST_CONSISTENCY, status=CheckStageStatus.PASS),
|
|
],
|
|
)
|
|
assert run.status == "SUCCEEDED"
|
|
|
|
# ── Lines 219, 221: ComplianceCheckRun COMPLIANT validation errors ──
|
|
|
|
def test_compliant_requires_all_stages(self):
|
|
"""COMPLIANT status with missing stages raises ValueError (line 219)."""
|
|
from src.models.clean_release import (
|
|
CheckFinalStatus, CheckStageName, CheckStageResult, CheckStageStatus,
|
|
ComplianceCheckRun, ExecutionMode,
|
|
)
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
with pytest.raises(ValueError, match="compliant run requires all mandatory stages"):
|
|
ComplianceCheckRun(
|
|
check_run_id="run-err1",
|
|
candidate_id="c-1",
|
|
policy_id="p-1",
|
|
started_at=ts,
|
|
triggered_by="admin",
|
|
execution_mode=ExecutionMode.TUI,
|
|
final_status=CheckFinalStatus.COMPLIANT,
|
|
checks=[
|
|
CheckStageResult(stage=CheckStageName.DATA_PURITY, status=CheckStageStatus.PASS),
|
|
# Missing: INTERNAL_SOURCES_ONLY, NO_EXTERNAL_ENDPOINTS, MANIFEST_CONSISTENCY
|
|
],
|
|
)
|
|
|
|
def test_compliant_requires_all_pass(self):
|
|
"""COMPLIANT status with FAIL checks raises ValueError (line 221)."""
|
|
from src.models.clean_release import (
|
|
CheckFinalStatus, CheckStageName, CheckStageResult, CheckStageStatus,
|
|
ComplianceCheckRun, ExecutionMode,
|
|
)
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
with pytest.raises(ValueError, match="compliant run requires PASS on all mandatory stages"):
|
|
ComplianceCheckRun(
|
|
check_run_id="run-err2",
|
|
candidate_id="c-1",
|
|
policy_id="p-1",
|
|
started_at=ts,
|
|
triggered_by="admin",
|
|
execution_mode=ExecutionMode.TUI,
|
|
final_status=CheckFinalStatus.COMPLIANT,
|
|
checks=[
|
|
CheckStageResult(stage=CheckStageName.DATA_PURITY, status=CheckStageStatus.PASS),
|
|
CheckStageResult(stage=CheckStageName.INTERNAL_SOURCES_ONLY, status=CheckStageStatus.PASS),
|
|
CheckStageResult(stage=CheckStageName.NO_EXTERNAL_ENDPOINTS, status=CheckStageStatus.PASS),
|
|
CheckStageResult(stage=CheckStageName.MANIFEST_CONSISTENCY, status=CheckStageStatus.FAIL),
|
|
],
|
|
)
|
|
|
|
|
|
class TestManifestItem:
|
|
"""ManifestItem — pydantic dataclass."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ClassificationType, ManifestItem
|
|
|
|
item = ManifestItem(path="/file.txt", category="data", classification=ClassificationType.ALLOWED, reason="ok")
|
|
assert item.path == "/file.txt"
|
|
assert item.checksum is None
|
|
|
|
def test_with_checksum(self):
|
|
from src.models.clean_release import ClassificationType, ManifestItem
|
|
|
|
item = ManifestItem(path="/file.txt", category="data", classification=ClassificationType.REQUIRED_SYSTEM, reason="core", checksum="sha256:abc")
|
|
assert item.checksum == "sha256:abc"
|
|
|
|
|
|
class TestManifestSummary:
|
|
"""ManifestSummary — pydantic dataclass."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.clean_release import ManifestSummary
|
|
|
|
summary = ManifestSummary(included_count=1, excluded_count=2, prohibited_detected_count=0)
|
|
assert summary.included_count == 1
|
|
assert summary.excluded_count == 2
|
|
# #endregion Test.Models.CleanRelease.Ext
|