ss-tools/backend/tests/services/clean_release/test_preparation_service.py

# #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