busya
4205618ee6
Warnings fixed: - datetime.utcnow() → datetime.now(UTC) across 48+ files (src/ + tests/) - datetime.utcnow (callback ref) → lambda: datetime.now(UTC) in model fields (18 files) - Pydantic class Config → model_config = ConfigDict(...) (16 files) - Pydantic .dict() → .model_dump() (8 files) - ConfigDict(allow_population_by_field_name=True) → validate_by_name=True - SQLAlchemy declarative_base() import path updated - FastAPI on_event → lifespan context manager (app.py) - Import sorting (ruff I001) auto-fixed across all files - Fixed broken re-export chains that ruff F401 cleanup broke: _validate_bcp47: service.py now imports from dictionary_validation directly job_to_response: _job_routes.py and test imports from service_utils directly fetch_datasource_metadata: restored re-export in service.py - Added missing TranslateJobService import in _job_routes.py (was deleted by F401) - Added ConfigDict(protected_namespaces=()) for DashboardDatasetItem schema field - pytest.ini: replaced deprecated importmode with asyncio_mode All 440 tests pass with zero deprecation warnings.
301 lines
9.9 KiB
Python
301 lines
9.9 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
|
|
# #endregion test_candidate_manifest_services
|