477 lines
17 KiB
Python
477 lines
17 KiB
Python
# #region TestPublicationService [C:2] [TYPE Module]
|
|
# @RELATION BINDS_TO -> SrcRoot
|
|
# @SEMANTICS: tests, clean-release, publication, revoke, gate
|
|
# @PURPOSE: Define publication gate contracts over approved candidates and immutable publication records.
|
|
# @LAYER Tests
|
|
# @INVARIANT: Publish requires approval; revoke requires existing publication; republish after revoke is allowed as a new record.
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
import pytest
|
|
|
|
from src.models.clean_release import ComplianceReport, ReleaseCandidate
|
|
from src.services.clean_release.enums import CandidateStatus, ComplianceDecision, PublicationStatus
|
|
from src.services.clean_release.exceptions import PublicationGateError
|
|
from src.services.clean_release.repository import CleanReleaseRepository
|
|
|
|
|
|
# #region _seed_candidate_with_passed_report [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestPublicationService
|
|
# @PURPOSE: Seed candidate/report fixtures for publication gate scenarios.
|
|
# @PRE: candidate_id and report_id are non-empty.
|
|
# @POST: Repository contains candidate and PASSED report.
|
|
def _seed_candidate_with_passed_report(
|
|
*,
|
|
candidate_id: str = "cand-publish-1",
|
|
report_id: str = "CCR-publish-1",
|
|
candidate_status: CandidateStatus = CandidateStatus.CHECK_PASSED,
|
|
) -> tuple[CleanReleaseRepository, str, str]:
|
|
repository = CleanReleaseRepository()
|
|
repository.save_candidate(
|
|
ReleaseCandidate(
|
|
id=candidate_id,
|
|
version="1.0.0",
|
|
source_snapshot_ref="git:sha-publish-1",
|
|
created_by="tester",
|
|
created_at=datetime.now(UTC),
|
|
status=candidate_status.value,
|
|
)
|
|
)
|
|
repository.save_report(
|
|
ComplianceReport(
|
|
id=report_id,
|
|
run_id="run-publish-1",
|
|
candidate_id=candidate_id,
|
|
final_status=ComplianceDecision.PASSED.value,
|
|
summary_json={"operator_summary": "seed", "violations_count": 0, "blocking_violations_count": 0},
|
|
generated_at=datetime.now(UTC),
|
|
immutable=True,
|
|
)
|
|
)
|
|
return repository, candidate_id, report_id
|
|
# #endregion _seed_candidate_with_passed_report
|
|
|
|
|
|
# #region test_publish_without_approval_rejected [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestPublicationService
|
|
# @PURPOSE: Ensure publish action is blocked until candidate is approved.
|
|
# @PRE: Candidate has PASSED report but status is not APPROVED.
|
|
# @POST: publish_candidate raises PublicationGateError.
|
|
def test_publish_without_approval_rejected():
|
|
from src.services.clean_release.publication_service import publish_candidate
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
|
|
with pytest.raises(PublicationGateError, match="APPROVED"):
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
publication_ref="rel-1",
|
|
)
|
|
# #endregion test_publish_without_approval_rejected
|
|
|
|
|
|
# #region test_revoke_unknown_publication_rejected [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestPublicationService
|
|
# @PURPOSE: Ensure revocation is rejected for unknown publication id.
|
|
# @PRE: Repository has no matching publication record.
|
|
# @POST: revoke_publication raises PublicationGateError.
|
|
def test_revoke_unknown_publication_rejected():
|
|
from src.services.clean_release.publication_service import revoke_publication
|
|
|
|
repository, _, _ = _seed_candidate_with_passed_report()
|
|
|
|
with pytest.raises(PublicationGateError, match="not found"):
|
|
revoke_publication(
|
|
repository=repository,
|
|
publication_id="missing-publication",
|
|
revoked_by="publisher",
|
|
comment="unknown publication id",
|
|
)
|
|
# #endregion test_revoke_unknown_publication_rejected
|
|
|
|
|
|
# #region test_republish_after_revoke_creates_new_active_record [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestPublicationService
|
|
# @PURPOSE: Ensure republish after revoke is allowed and creates a new ACTIVE record.
|
|
# @PRE: Candidate is APPROVED and first publication has been revoked.
|
|
# @POST: New publish call returns distinct publication id with ACTIVE status.
|
|
def test_republish_after_revoke_creates_new_active_record():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate, revoke_publication
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
comment="approval before publication",
|
|
)
|
|
|
|
first = publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
publication_ref="release-1",
|
|
)
|
|
revoked = revoke_publication(
|
|
repository=repository,
|
|
publication_id=first.id,
|
|
revoked_by="publisher",
|
|
comment="rollback",
|
|
)
|
|
second = publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
publication_ref="release-2",
|
|
)
|
|
|
|
assert first.id != second.id
|
|
assert revoked.status == PublicationStatus.REVOKED.value
|
|
assert second.status == PublicationStatus.ACTIVE.value
|
|
# #endregion test_republish_after_revoke_creates_new_active_record
|
|
|
|
# #region test_publish_rejects_empty_published_by [C:2] [TYPE Function]
|
|
def test_publish_rejects_empty_published_by():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
comment="approved",
|
|
)
|
|
|
|
with pytest.raises(PublicationGateError, match="non-empty"):
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="",
|
|
target_channel="stable",
|
|
)
|
|
# #endregion test_publish_rejects_empty_published_by
|
|
|
|
|
|
# #region test_publish_rejects_empty_target_channel [C:2] [TYPE Function]
|
|
def test_publish_rejects_empty_target_channel():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
comment="approved",
|
|
)
|
|
|
|
with pytest.raises(PublicationGateError, match="non-empty"):
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="",
|
|
)
|
|
# #endregion test_publish_rejects_empty_target_channel
|
|
|
|
|
|
# #region test_publish_rejects_nonexistent_candidate [C:2] [TYPE Function]
|
|
def test_publish_rejects_nonexistent_candidate():
|
|
from src.services.clean_release.publication_service import publish_candidate
|
|
|
|
repository, _, report_id = _seed_candidate_with_passed_report()
|
|
|
|
with pytest.raises(PublicationGateError, match="not found"):
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id="nonexistent",
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
# #endregion test_publish_rejects_nonexistent_candidate
|
|
|
|
|
|
# #region test_publish_rejects_nonexistent_report [C:2] [TYPE Function]
|
|
def test_publish_rejects_nonexistent_report():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
comment="approved",
|
|
)
|
|
|
|
with pytest.raises(PublicationGateError, match="not found"):
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id="nonexistent",
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
# #endregion test_publish_rejects_nonexistent_report
|
|
|
|
|
|
# #region test_publish_rejects_foreign_report [C:2] [TYPE Function]
|
|
def test_publish_rejects_foreign_report():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate
|
|
from src.models.clean_release import ComplianceReport
|
|
|
|
repository, candidate_id, _ = _seed_candidate_with_passed_report(
|
|
candidate_id="cand-foreign-pub",
|
|
report_id="CCR-foreign-pub",
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id="CCR-foreign-pub",
|
|
decided_by="approver",
|
|
comment="approved",
|
|
)
|
|
|
|
# Save a foreign report directly into the same repository
|
|
foreign_report = ComplianceReport(
|
|
id="CCR-other-pub",
|
|
run_id="run-foreign",
|
|
candidate_id="cand-other",
|
|
final_status=ComplianceDecision.PASSED.value,
|
|
summary_json={"operator_summary": "foreign", "violations_count": 0, "blocking_violations_count": 0},
|
|
generated_at=datetime.now(UTC),
|
|
immutable=True,
|
|
)
|
|
repository.save_report(foreign_report)
|
|
|
|
with pytest.raises(PublicationGateError, match="belongs to another candidate"):
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id="CCR-other-pub",
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
# #endregion test_publish_rejects_foreign_report
|
|
|
|
|
|
# #region test_publish_rejects_duplicate_active_publication [C:2] [TYPE Function]
|
|
def test_publish_rejects_duplicate_active_publication():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_id="cand-dup-pub",
|
|
report_id="CCR-dup-pub",
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
comment="approved",
|
|
)
|
|
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
|
|
with pytest.raises(PublicationGateError, match="already has active publication"):
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
# #endregion test_publish_rejects_duplicate_active_publication
|
|
|
|
|
|
# #region test_revoke_rejects_empty_revoked_by [C:2] [TYPE Function]
|
|
def test_revoke_rejects_empty_revoked_by():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate, revoke_publication
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
)
|
|
pub = publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
|
|
with pytest.raises(PublicationGateError, match="non-empty"):
|
|
revoke_publication(
|
|
repository=repository,
|
|
publication_id=pub.id,
|
|
revoked_by="",
|
|
)
|
|
# #endregion test_revoke_rejects_empty_revoked_by
|
|
|
|
|
|
# #region test_revoke_rejects_empty_publication_id [C:2] [TYPE Function]
|
|
def test_revoke_rejects_empty_publication_id():
|
|
from src.services.clean_release.publication_service import revoke_publication
|
|
|
|
repository, _, _ = _seed_candidate_with_passed_report()
|
|
|
|
with pytest.raises(PublicationGateError, match="non-empty"):
|
|
revoke_publication(
|
|
repository=repository,
|
|
publication_id="",
|
|
revoked_by="publisher",
|
|
)
|
|
# #endregion test_revoke_rejects_empty_publication_id
|
|
|
|
|
|
# #region test_revoke_rejects_already_revoked [C:2] [TYPE Function]
|
|
def test_revoke_rejects_already_revoked():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate, revoke_publication
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
)
|
|
pub = publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
revoke_publication(
|
|
repository=repository,
|
|
publication_id=pub.id,
|
|
revoked_by="publisher",
|
|
)
|
|
|
|
with pytest.raises(PublicationGateError, match="already revoked"):
|
|
revoke_publication(
|
|
repository=repository,
|
|
publication_id=pub.id,
|
|
revoked_by="publisher",
|
|
)
|
|
# #endregion test_revoke_rejects_already_revoked
|
|
|
|
|
|
# #region test_publish_persists_audit_event [C:2] [TYPE Function]
|
|
def test_publish_persists_audit_event():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_id="cand-pub-audit",
|
|
report_id="CCR-pub-audit",
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
)
|
|
|
|
publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
|
|
assert len(repository.audit_events) >= 1
|
|
last_event = repository.audit_events[-1]
|
|
# audit_preparation sets action="PREPARATION", decision type goes in "status"
|
|
assert "PUBLISHED" in str(last_event.get("status", ""))
|
|
# #endregion test_publish_persists_audit_event
|
|
|
|
|
|
# #region test_revoke_persists_audit_event [C:2] [TYPE Function]
|
|
def test_revoke_persists_audit_event():
|
|
from src.services.clean_release.approval_service import approve_candidate
|
|
from src.services.clean_release.publication_service import publish_candidate, revoke_publication
|
|
|
|
repository, candidate_id, report_id = _seed_candidate_with_passed_report(
|
|
candidate_id="cand-rev-audit",
|
|
report_id="CCR-rev-audit",
|
|
candidate_status=CandidateStatus.CHECK_PASSED,
|
|
)
|
|
approve_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
decided_by="approver",
|
|
)
|
|
pub = publish_candidate(
|
|
repository=repository,
|
|
candidate_id=candidate_id,
|
|
report_id=report_id,
|
|
published_by="publisher",
|
|
target_channel="stable",
|
|
)
|
|
|
|
revoke_publication(
|
|
repository=repository,
|
|
publication_id=pub.id,
|
|
revoked_by="publisher",
|
|
)
|
|
|
|
assert len(repository.audit_events) >= 1
|
|
last_event = repository.audit_events[-1]
|
|
# audit_preparation sets action="PREPARATION", decision type goes in "status"
|
|
assert "REVOKED" in str(last_event.get("status", ""))
|
|
# #endregion test_revoke_persists_audit_event
|
|
|
|
|
|
# #endregion TestPublicationService
|