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

# [DEF:backend.tests.services.clean_release.test_publication_service:Module]
# @COMPLEXITY: 5
# @SEMANTICS: tests, clean-release, publication, revoke, gate
# @PURPOSE: Define publication gate contracts over approved candidates and immutable publication records.
# @LAYER: Tests
# @RELATION: TESTS -> src.services.clean_release.publication_service
# @RELATION: TESTS -> src.services.clean_release.approval_service
# @RELATION: TESTS -> src.services.clean_release.repository
# @INVARIANT: Publish requires approval; revoke requires existing publication; republish after revoke is allowed as a new record.

from __future__ import annotations

from datetime import datetime, timezone

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


# [DEF:_seed_candidate_with_passed_report:Function]
# @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(timezone.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(timezone.utc),
            immutable=True,
        )
    )
    return repository, candidate_id, report_id
# [/DEF:_seed_candidate_with_passed_report:Function]


# [DEF:test_publish_without_approval_rejected:Function]
# @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",
        )
# [/DEF:test_publish_without_approval_rejected:Function]


# [DEF:test_revoke_unknown_publication_rejected:Function]
# @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",
        )
# [/DEF:test_revoke_unknown_publication_rejected:Function]


# [DEF:test_republish_after_revoke_creates_new_active_record:Function]
# @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
# [/DEF:test_republish_after_revoke_creates_new_active_record:Function]

# [/DEF:backend.tests.services.clean_release.test_publication_service:Module]