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

# [DEF:test_candidate_manifest_services:Module]
# @RELATION: BELONGS_TO -> [SrcRoot:Module]
# @COMPLEXITY: 3
# @PURPOSE: Test lifecycle and manifest versioning for release candidates.
# @LAYER: Tests

import pytest
from datetime import datetime, timezone
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from src.core.database import Base
from src.models.clean_release import (
    ReleaseCandidate,
    DistributionManifest,
    CandidateArtifact,
)
from src.services.clean_release.enums import CandidateStatus
from src.services.clean_release.candidate_service import register_candidate
from src.services.clean_release.manifest_service import build_manifest_snapshot
from src.services.clean_release.repository import CleanReleaseRepository


@pytest.fixture
def db_session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()


# [DEF:test_candidate_lifecycle_transitions:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Verify release candidate allows legal status transitions and rejects forbidden back-transitions.
def test_candidate_lifecycle_transitions(db_session):
    """
    @PURPOSE: Verify legal state transitions for ReleaseCandidate.
    """
    candidate = ReleaseCandidate(
        id="test-candidate-1",
        name="Test Candidate",
        version="1.0.0",
        source_snapshot_ref="ref-1",
        created_by="operator",
        status=CandidateStatus.DRAFT,
    )
    db_session.add(candidate)
    db_session.commit()

    # Valid transition: DRAFT -> PREPARED
    candidate.transition_to(CandidateStatus.PREPARED)
    assert candidate.status == CandidateStatus.PREPARED

    # Invalid transition: PREPARED -> DRAFT (should raise IllegalTransitionError)
    from src.services.clean_release.exceptions import IllegalTransitionError

    with pytest.raises(IllegalTransitionError, match="Forbidden transition"):
        candidate.transition_to(CandidateStatus.DRAFT)


# [/DEF:test_candidate_lifecycle_transitions:Function]


# [DEF:test_manifest_versioning_and_immutability:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Verify manifest versions increment monotonically and older snapshots remain queryable.
def test_manifest_versioning_and_immutability(db_session):
    """
    @PURPOSE: Verify manifest versioning and immutability invariants.
    """
    candidate_id = "test-candidate-2"

    # Create version 1
    m1 = DistributionManifest(
        id="manifest-v1",
        candidate_id=candidate_id,
        manifest_version=1,
        manifest_digest="hash1",
        artifacts_digest="hash1",
        source_snapshot_ref="ref1",
        content_json={},
        created_at=datetime.now(timezone.utc),
        created_by="operator",
    )
    db_session.add(m1)

    # Create version 2
    m2 = DistributionManifest(
        id="manifest-v2",
        candidate_id=candidate_id,
        manifest_version=2,
        manifest_digest="hash2",
        artifacts_digest="hash2",
        source_snapshot_ref="ref1",
        content_json={},
        created_at=datetime.now(timezone.utc),
        created_by="operator",
    )
    db_session.add(m2)
    db_session.commit()

    latest = (
        db_session.query(DistributionManifest)
        .filter_by(candidate_id=candidate_id)
        .order_by(DistributionManifest.manifest_version.desc())
        .first()
    )
    assert latest.manifest_version == 2
    assert latest.id == "manifest-v2"

    all_manifests = (
        db_session.query(DistributionManifest)
        .filter_by(candidate_id=candidate_id)
        .all()
    )
    assert len(all_manifests) == 2


# [/DEF:test_manifest_versioning_and_immutability:Function]


# [DEF:_valid_artifacts:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Provide canonical valid artifact payload used by candidate registration tests.
def _valid_artifacts():
    return [
        {
            "id": "art-1",
            "path": "bin/app",
            "sha256": "abc123",
            "size": 42,
        }
    ]


# [/DEF:_valid_artifacts:Function]


# [DEF:test_register_candidate_rejects_duplicate_candidate_id:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Verify duplicate candidate_id registration is rejected by service invariants.
def test_register_candidate_rejects_duplicate_candidate_id():
    repository = CleanReleaseRepository()
    register_candidate(
        repository=repository,
        candidate_id="dup-1",
        version="1.0.0",
        source_snapshot_ref="git:sha1",
        created_by="operator",
        artifacts=_valid_artifacts(),
    )

    with pytest.raises(ValueError, match="already exists"):
        register_candidate(
            repository=repository,
            candidate_id="dup-1",
            version="1.0.0",
            source_snapshot_ref="git:sha1",
            created_by="operator",
            artifacts=_valid_artifacts(),
        )


# [/DEF:test_register_candidate_rejects_duplicate_candidate_id:Function]


# [DEF:test_register_candidate_rejects_malformed_artifact_input:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Verify candidate registration rejects artifact payloads missing required fields.
def test_register_candidate_rejects_malformed_artifact_input():
    repository = CleanReleaseRepository()
    bad_artifacts = [{"id": "art-1", "path": "bin/app", "size": 42}]  # missing sha256

    with pytest.raises(ValueError, match="missing required field 'sha256'"):
        register_candidate(
            repository=repository,
            candidate_id="bad-art-1",
            version="1.0.0",
            source_snapshot_ref="git:sha2",
            created_by="operator",
            artifacts=bad_artifacts,
        )


# [/DEF:test_register_candidate_rejects_malformed_artifact_input:Function]


# [DEF:test_register_candidate_rejects_empty_artifact_set:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Verify candidate registration rejects empty artifact collections.
def test_register_candidate_rejects_empty_artifact_set():
    repository = CleanReleaseRepository()

    with pytest.raises(ValueError, match="artifacts must not be empty"):
        register_candidate(
            repository=repository,
            candidate_id="empty-art-1",
            version="1.0.0",
            source_snapshot_ref="git:sha3",
            created_by="operator",
            artifacts=[],
        )


# [/DEF:test_register_candidate_rejects_empty_artifact_set:Function]


# [DEF:test_manifest_service_rebuild_creates_new_version:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Verify repeated manifest build creates a new incremented immutable version.
def test_manifest_service_rebuild_creates_new_version():
    repository = CleanReleaseRepository()
    register_candidate(
        repository=repository,
        candidate_id="manifest-version-1",
        version="1.0.0",
        source_snapshot_ref="git:sha10",
        created_by="operator",
        artifacts=_valid_artifacts(),
    )

    first = build_manifest_snapshot(
        repository=repository, candidate_id="manifest-version-1", created_by="operator"
    )
    second = build_manifest_snapshot(
        repository=repository, candidate_id="manifest-version-1", created_by="operator"
    )

    assert first.manifest_version == 1
    assert second.manifest_version == 2
    assert first.id != second.id


# [/DEF:test_manifest_service_rebuild_creates_new_version:Function]


# [DEF:test_manifest_service_existing_manifest_cannot_be_mutated:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Verify existing manifest snapshot remains immutable when rebuilding newer manifest version.
def test_manifest_service_existing_manifest_cannot_be_mutated():
    repository = CleanReleaseRepository()
    register_candidate(
        repository=repository,
        candidate_id="manifest-immutable-1",
        version="1.0.0",
        source_snapshot_ref="git:sha11",
        created_by="operator",
        artifacts=_valid_artifacts(),
    )

    created = build_manifest_snapshot(
        repository=repository,
        candidate_id="manifest-immutable-1",
        created_by="operator",
    )
    original_digest = created.manifest_digest

    rebuilt = build_manifest_snapshot(
        repository=repository,
        candidate_id="manifest-immutable-1",
        created_by="operator",
    )
    old_manifest = repository.get_manifest(created.id)

    assert old_manifest is not None
    assert old_manifest.manifest_digest == original_digest
    assert old_manifest.id == created.id
    assert rebuilt.id != created.id


# [/DEF:test_manifest_service_existing_manifest_cannot_be_mutated:Function]


# [DEF:test_manifest_service_rejects_missing_candidate:Function]
# @RELATION: BINDS_TO -> [test_candidate_manifest_services:Module]
# @PURPOSE: Verify manifest build fails with missing candidate identifier.
def test_manifest_service_rejects_missing_candidate():
    repository = CleanReleaseRepository()

    with pytest.raises(ValueError, match="not found"):
        build_manifest_snapshot(
            repository=repository,
            candidate_id="missing-candidate",
            created_by="operator",
        )


# [/DEF:test_manifest_service_rejects_missing_candidate:Function]
# [/DEF:test_candidate_manifest_services:Module]