240 lines
10 KiB
Python
240 lines
10 KiB
Python
# #region Test.CleanRelease.PreparationService [C:3] [TYPE Module] [SEMANTICS test,clean-release,preparation,policy,manifest]
|
|
# @BRIEF Tests for prepare_candidate — policy evaluation, manifest creation, status transitions, violation handling.
|
|
# @RELATION BINDS_TO -> [PreparationService]
|
|
# @TEST_EDGE: missing_candidate -> raises ValueError
|
|
# @TEST_EDGE: missing_policy -> raises ValueError
|
|
# @TEST_EDGE: missing_registry -> raises ValueError
|
|
# @TEST_EDGE: invalid_policy -> raises ValueError
|
|
# @TEST_EDGE: clean_preparation -> PREPARED status, manifest created
|
|
# @TEST_EDGE: preparation_with_violations -> BLOCKED status
|
|
# @TEST_EDGE: preparation_with_back_transition -> still PREPARED
|
|
# @TEST_EDGE: legacy_wrapper
|
|
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
|
|
from src.models.clean_release import (
|
|
CleanPolicySnapshot,
|
|
ReleaseCandidate,
|
|
SourceRegistrySnapshot,
|
|
)
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
from src.services.clean_release.preparation_service import prepare_candidate, prepare_candidate_legacy
|
|
|
|
|
|
def _make_repo_with_policy_registry(
|
|
policy_registry_ref: str = "registry-prep-1",
|
|
policy_content: dict | None = None,
|
|
) -> CleanReleaseRepository:
|
|
"""Create a repository with candidate, policy, and registry."""
|
|
repo = CleanReleaseRepository()
|
|
repo.save_candidate(ReleaseCandidate(
|
|
id="cand-prep-1", version="1.0.0", source_snapshot_ref="git:sha1",
|
|
created_by="tester", created_at=datetime.now(UTC),
|
|
status=CandidateStatus.DRAFT.value,
|
|
))
|
|
repo.save_candidate(ReleaseCandidate(
|
|
id="cand-prep-transitioned", version="1.0.0", source_snapshot_ref="git:sha2",
|
|
created_by="tester", created_at=datetime.now(UTC),
|
|
status=CandidateStatus.DRAFT.value,
|
|
))
|
|
repo.save_policy(CleanPolicySnapshot(
|
|
id="policy-prep-1", policy_id="policy-prep-1", policy_version="1",
|
|
content_json=policy_content or {
|
|
"profile": "enterprise-clean",
|
|
"prohibited_artifact_categories": ["malware"],
|
|
"required_system_categories": [],
|
|
"external_source_forbidden": True,
|
|
},
|
|
registry_snapshot_id=policy_registry_ref,
|
|
immutable=True,
|
|
))
|
|
repo.save_registry(SourceRegistrySnapshot(
|
|
id=policy_registry_ref, registry_id=policy_registry_ref,
|
|
registry_version="1", allowed_hosts=["internal.local"],
|
|
immutable=True,
|
|
))
|
|
return repo
|
|
|
|
|
|
class TestPrepareCandidate:
|
|
"""prepare_candidate — full preparation flow."""
|
|
|
|
# #region test_missing_candidate_raises [C:2] [TYPE Function]
|
|
def test_missing_candidate_raises(self):
|
|
"""Missing candidate → ValueError."""
|
|
repo = CleanReleaseRepository()
|
|
with pytest.raises(ValueError, match="not found"):
|
|
prepare_candidate(
|
|
repository=repo, candidate_id="nonexistent",
|
|
artifacts=[], sources=[], operator_id="tester",
|
|
)
|
|
# #endregion test_missing_candidate_raises
|
|
|
|
# #region test_missing_policy_raises [C:2] [TYPE Function]
|
|
def test_missing_policy_raises(self):
|
|
"""Missing active policy → ValueError."""
|
|
repo = CleanReleaseRepository()
|
|
repo.save_candidate(ReleaseCandidate(
|
|
id="cand-nopolicy", version="1.0.0", source_snapshot_ref="ref",
|
|
created_by="tester", created_at=datetime.now(UTC),
|
|
status=CandidateStatus.DRAFT.value,
|
|
))
|
|
with pytest.raises(ValueError, match="policy"):
|
|
prepare_candidate(
|
|
repository=repo, candidate_id="cand-nopolicy",
|
|
artifacts=[], sources=[], operator_id="tester",
|
|
)
|
|
# #endregion test_missing_policy_raises
|
|
|
|
# #region test_clean_preparation [C:2] [TYPE Function]
|
|
def test_clean_preparation(self):
|
|
"""Clean artifacts → PREPARED status, manifest created."""
|
|
repo = _make_repo_with_policy_registry()
|
|
artifacts = [
|
|
{"path": "bin/app", "category": "generic", "reason": "test", "checksum": "abc"},
|
|
]
|
|
result = prepare_candidate(
|
|
repository=repo, candidate_id="cand-prep-1",
|
|
artifacts=artifacts, sources=["internal.local"],
|
|
operator_id="tester",
|
|
)
|
|
assert result["status"] == "PREPARED"
|
|
assert result["candidate_id"] == "cand-prep-1"
|
|
assert result["manifest_id"] is not None
|
|
assert result["violations"] == []
|
|
assert "prepared_at" in result
|
|
|
|
candidate = repo.get_candidate("cand-prep-1")
|
|
assert candidate.status == CandidateStatus.PREPARED.value
|
|
assert len(repo.manifests) == 1
|
|
# #endregion test_clean_preparation
|
|
|
|
# #region test_preparation_with_violations [C:2] [TYPE Function]
|
|
def test_preparation_with_violations(self):
|
|
"""Artifacts with prohibited category → BLOCKED status."""
|
|
repo = _make_repo_with_policy_registry()
|
|
artifacts = [
|
|
{"path": "lib/bad.exe", "category": "malware", "reason": "test", "checksum": "bad"},
|
|
]
|
|
result = prepare_candidate(
|
|
repository=repo, candidate_id="cand-prep-1",
|
|
artifacts=artifacts, sources=["internal.local"],
|
|
operator_id="tester",
|
|
)
|
|
assert result["status"] == "CHECK_BLOCKED"
|
|
assert len(result["violations"]) == 1
|
|
assert result["violations"][0]["category"] == "data-purity"
|
|
# #endregion test_preparation_with_violations
|
|
|
|
# #region test_preparation_with_external_source_violation [C:2] [TYPE Function]
|
|
def test_preparation_with_external_source_violation(self):
|
|
"""External source → BLOCKED status."""
|
|
repo = _make_repo_with_policy_registry()
|
|
artifacts = [
|
|
{"path": "bin/app", "category": "generic", "reason": "test", "checksum": "abc"},
|
|
]
|
|
result = prepare_candidate(
|
|
repository=repo, candidate_id="cand-prep-1",
|
|
artifacts=artifacts, sources=["external.bad.com"],
|
|
operator_id="tester",
|
|
)
|
|
assert result["status"] == "CHECK_BLOCKED"
|
|
assert len(result["violations"]) == 1
|
|
assert result["violations"][0]["category"] == "external-source"
|
|
# #endregion test_preparation_with_external_source_violation
|
|
|
|
# #region test_preparation_legacy_wrapper [C:2] [TYPE Function]
|
|
def test_preparation_legacy_wrapper(self):
|
|
"""prepare_candidate_legacy delegates to prepare_candidate."""
|
|
repo = _make_repo_with_policy_registry()
|
|
artifacts = [
|
|
{"path": "bin/app", "category": "generic", "reason": "test", "checksum": "abc"},
|
|
]
|
|
result_legacy = prepare_candidate_legacy(
|
|
repository=repo, candidate_id="cand-prep-transitioned",
|
|
artifacts=artifacts, sources=["internal.local"],
|
|
operator_id="tester",
|
|
)
|
|
result_canonical = prepare_candidate(
|
|
repository=repo, candidate_id="cand-prep-transitioned",
|
|
artifacts=artifacts, sources=["internal.local"],
|
|
operator_id="tester",
|
|
)
|
|
assert result_legacy["status"] == result_canonical["status"]
|
|
# #endregion test_preparation_legacy_wrapper
|
|
|
|
# #region test_registry_not_found_raises [C:2] [TYPE Function]
|
|
def test_registry_not_found_raises(self):
|
|
"""Registry lookup returns None → ValueError (line 44)."""
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
from src.models.clean_release import ReleaseCandidate, CleanPolicySnapshot
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
repo = CleanReleaseRepository()
|
|
repo.save_candidate(ReleaseCandidate(
|
|
id="cand-noregistry", version="1.0.0", source_snapshot_ref="git:sha1",
|
|
created_by="tester", created_at=datetime.now(UTC),
|
|
status=CandidateStatus.DRAFT.value,
|
|
))
|
|
# Policy references a registry_id that does not exist in repository
|
|
repo.save_policy(CleanPolicySnapshot(
|
|
id="policy-noreg", policy_id="policy-noreg", policy_version="1",
|
|
content_json={
|
|
"profile": "enterprise-clean",
|
|
"prohibited_artifact_categories": ["malware"],
|
|
"required_system_categories": [],
|
|
"external_source_forbidden": True,
|
|
},
|
|
registry_snapshot_id="registry-nonexistent",
|
|
immutable=True,
|
|
))
|
|
|
|
with pytest.raises(ValueError, match="Registry not found"):
|
|
prepare_candidate(
|
|
repository=repo, candidate_id="cand-noregistry",
|
|
artifacts=[{"path": "bin/app", "category": "generic", "reason": "test", "checksum": "abc"}],
|
|
sources=[], operator_id="tester",
|
|
)
|
|
# #endregion test_registry_not_found_raises
|
|
|
|
# #region test_invalid_policy_raises [C:2] [TYPE Function]
|
|
def test_invalid_policy_raises(self):
|
|
"""Policy validation fails → ValueError (line 49)."""
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
from src.models.clean_release import ReleaseCandidate, CleanPolicySnapshot, SourceRegistrySnapshot
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
repo = CleanReleaseRepository()
|
|
repo.save_candidate(ReleaseCandidate(
|
|
id="cand-badpolicy", version="1.0.0", source_snapshot_ref="git:sha1",
|
|
created_by="tester", created_at=datetime.now(UTC),
|
|
status=CandidateStatus.DRAFT.value,
|
|
))
|
|
# Create policy with missing profile settings to make validate_policy fail
|
|
repo.save_policy(CleanPolicySnapshot(
|
|
id="policy-invalid", policy_id="policy-invalid", policy_version="1",
|
|
content_json={
|
|
"profile": "enterprise-clean",
|
|
# Missing prohibited_artifact_categories — should fail validation
|
|
"external_source_forbidden": True,
|
|
},
|
|
registry_snapshot_id="registry-invalid",
|
|
immutable=True,
|
|
))
|
|
repo.save_registry(SourceRegistrySnapshot(
|
|
id="registry-invalid", registry_id="registry-invalid",
|
|
registry_version="1", allowed_hosts=["internal.local"],
|
|
immutable=True,
|
|
))
|
|
|
|
with pytest.raises(ValueError, match="Invalid policy"):
|
|
prepare_candidate(
|
|
repository=repo, candidate_id="cand-badpolicy",
|
|
artifacts=[{"path": "bin/app", "category": "generic", "reason": "test", "checksum": "abc"}],
|
|
sources=[], operator_id="tester",
|
|
)
|
|
# #endregion test_invalid_policy_raises
|