422 lines
15 KiB
Python
422 lines
15 KiB
Python
# #region test_candidate_manifest_services [C:2] [TYPE Module]
|
|
# @RELATION BINDS_TO -> [SrcRoot]
|
|
# @PURPOSE: Test lifecycle and manifest versioning for release candidates.
|
|
# @LAYER Tests
|
|
|
|
from datetime import UTC, datetime
|
|
import pytest
|
|
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from src.core.database import Base
|
|
from src.models.clean_release import (
|
|
DistributionManifest,
|
|
ReleaseCandidate,
|
|
)
|
|
from src.services.clean_release.candidate_service import register_candidate
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
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:")
|
|
event.listen(engine, "connect", lambda c, _: c.execute("PRAGMA foreign_keys=ON"))
|
|
Base.metadata.create_all(engine)
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
# Create FK parent for test_manifest_versioning_and_immutability
|
|
existing = session.query(ReleaseCandidate).filter(ReleaseCandidate.id == 'test-candidate-2').first()
|
|
if not existing:
|
|
session.add(ReleaseCandidate(
|
|
id='test-candidate-2', version='1.0.0',
|
|
source_snapshot_ref='ref1', created_by='operator',
|
|
status=CandidateStatus.DRAFT
|
|
))
|
|
session.commit()
|
|
yield session
|
|
session.close()
|
|
|
|
|
|
# #region test_candidate_lifecycle_transitions [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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)
|
|
|
|
|
|
# #endregion test_candidate_lifecycle_transitions
|
|
|
|
|
|
# #region test_manifest_versioning_and_immutability [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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(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(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
|
|
|
|
|
|
# #endregion test_manifest_versioning_and_immutability
|
|
|
|
|
|
# #region _valid_artifacts [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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,
|
|
}
|
|
]
|
|
|
|
|
|
# #endregion _valid_artifacts
|
|
|
|
|
|
# #region test_register_candidate_rejects_duplicate_candidate_id [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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(),
|
|
)
|
|
|
|
|
|
# #endregion test_register_candidate_rejects_duplicate_candidate_id
|
|
|
|
|
|
# #region test_register_candidate_rejects_malformed_artifact_input [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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,
|
|
)
|
|
|
|
|
|
# #endregion test_register_candidate_rejects_malformed_artifact_input
|
|
|
|
|
|
# #region test_register_candidate_rejects_empty_artifact_set [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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=[],
|
|
)
|
|
|
|
|
|
# #endregion test_register_candidate_rejects_empty_artifact_set
|
|
|
|
|
|
# #region test_manifest_service_rebuild_creates_new_version [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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
|
|
|
|
|
|
# #endregion test_manifest_service_rebuild_creates_new_version
|
|
|
|
|
|
# #region test_manifest_service_existing_manifest_cannot_be_mutated [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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
|
|
|
|
|
|
# #endregion test_manifest_service_existing_manifest_cannot_be_mutated
|
|
|
|
|
|
# #region test_manifest_service_rejects_missing_candidate [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> [test_candidate_manifest_services]
|
|
# @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",
|
|
)
|
|
|
|
|
|
# #endregion test_manifest_service_rejects_missing_candidate
|
|
|
|
|
|
# #region test_manifest_service_rejects_empty_candidate_id [C:2] [TYPE Function]
|
|
def test_manifest_service_rejects_empty_candidate_id():
|
|
"""Empty candidate_id → ValueError (line 36)."""
|
|
from src.services.clean_release.manifest_service import build_manifest_snapshot
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
|
|
repository = CleanReleaseRepository()
|
|
with pytest.raises(ValueError, match="candidate_id must be non-empty"):
|
|
build_manifest_snapshot(
|
|
repository=repository,
|
|
candidate_id=" ",
|
|
created_by="operator",
|
|
)
|
|
# #endregion test_manifest_service_rejects_empty_candidate_id
|
|
|
|
|
|
# #region test_manifest_service_rejects_empty_created_by [C:2] [TYPE Function]
|
|
def test_manifest_service_rejects_empty_created_by():
|
|
"""Empty created_by → ValueError (line 38)."""
|
|
from src.services.clean_release.manifest_service import build_manifest_snapshot
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
|
|
repository = CleanReleaseRepository()
|
|
with pytest.raises(ValueError, match="created_by must be non-empty"):
|
|
build_manifest_snapshot(
|
|
repository=repository,
|
|
candidate_id="cand-1",
|
|
created_by=" ",
|
|
)
|
|
# #endregion test_manifest_service_rejects_empty_created_by
|
|
|
|
|
|
# #region test_manifest_service_rejects_wrong_status [C:2] [TYPE Function]
|
|
def test_manifest_service_rejects_wrong_status():
|
|
"""Candidate with wrong status → ValueError (line 48)."""
|
|
from src.services.clean_release.manifest_service import build_manifest_snapshot
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
from src.models.clean_release import ReleaseCandidate
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
repository = CleanReleaseRepository()
|
|
# Register in DRAFT (not PREPARED or MANIFEST_BUILT)
|
|
repository.save_candidate(ReleaseCandidate(
|
|
id="cand-wrong-status", version="1.0.0",
|
|
source_snapshot_ref="git:sha1", created_by="tester",
|
|
status=CandidateStatus.DRAFT.value,
|
|
))
|
|
|
|
with pytest.raises(ValueError, match="must be PREPARED or MANIFEST_BUILT"):
|
|
build_manifest_snapshot(
|
|
repository=repository,
|
|
candidate_id="cand-wrong-status",
|
|
created_by="operator",
|
|
)
|
|
# #endregion test_manifest_service_rejects_wrong_status
|
|
|
|
|
|
# #region test_manifest_service_rejects_no_artifacts [C:2] [TYPE Function]
|
|
def test_manifest_service_rejects_no_artifacts():
|
|
"""No artifacts → ValueError (line 54)."""
|
|
from src.services.clean_release.manifest_service import build_manifest_snapshot
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
from src.models.clean_release import ReleaseCandidate
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
|
|
repository = CleanReleaseRepository()
|
|
repository.save_candidate(ReleaseCandidate(
|
|
id="cand-no-artifacts", version="1.0.0",
|
|
source_snapshot_ref="git:sha1", created_by="tester",
|
|
status=CandidateStatus.PREPARED.value,
|
|
))
|
|
|
|
with pytest.raises(ValueError, match="artifacts are required"):
|
|
build_manifest_snapshot(
|
|
repository=repository,
|
|
candidate_id="cand-no-artifacts",
|
|
created_by="operator",
|
|
)
|
|
# #endregion test_manifest_service_rejects_no_artifacts
|
|
|
|
|
|
# #region test_manifest_service_rejects_existing_non_immutable [C:2] [TYPE Function]
|
|
def test_manifest_service_rejects_existing_non_immutable():
|
|
"""Existing manifest with immutable=False → ValueError (line 59)."""
|
|
from src.services.clean_release.manifest_service import build_manifest_snapshot
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
from src.models.clean_release import DistributionManifest, CandidateArtifact, ReleaseCandidate
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
from datetime import UTC, datetime
|
|
|
|
repository = CleanReleaseRepository()
|
|
repository.save_candidate(ReleaseCandidate(
|
|
id="cand-bad-manifest", version="1.0.0",
|
|
source_snapshot_ref="git:sha1", created_by="tester",
|
|
status=CandidateStatus.PREPARED.value,
|
|
))
|
|
# Add an artifact so it passes the artifacts check
|
|
repository.save_artifact(CandidateArtifact(
|
|
id="art-1", candidate_id="cand-bad-manifest",
|
|
path="bin/app", sha256="abc123", size=42,
|
|
))
|
|
# Add an existing manifest that is NOT immutable
|
|
repository.save_manifest(DistributionManifest(
|
|
id="manifest-existing", candidate_id="cand-bad-manifest",
|
|
manifest_version=1, manifest_digest="d1", artifacts_digest="d1",
|
|
source_snapshot_ref="ref1", content_json={},
|
|
created_at=datetime.now(UTC), created_by="tester",
|
|
immutable=False, # violation!
|
|
))
|
|
|
|
with pytest.raises(ValueError, match="immutability invariant violated"):
|
|
build_manifest_snapshot(
|
|
repository=repository,
|
|
candidate_id="cand-bad-manifest",
|
|
created_by="operator",
|
|
)
|
|
# #endregion test_manifest_service_rejects_existing_non_immutable
|
|
|
|
|
|
# #endregion test_candidate_manifest_services
|