platform-ci/tests/test_release_completion.py

"""Central publication and completion orchestration, the generic completion receipt, idempotent release-record finalization, and the configuration that asks for them.

NOT YET RUN: written while test execution was suspended. The frozen Central primitives are reached through injected fakes of their interfaces (`publish_central`, `CentralPending`,
`CentralDuplicate`, `verified_purls`), so these cases prove OUR orchestration, binding and receipt handling, not Central, the primitives or any credential.
"""

import hashlib
import importlib
import io
import json
import tempfile
import unittest
import urllib.error
from collections.abc import Callable, Iterable
from dataclasses import dataclass, field, replace
from email.message import Message
from pathlib import Path
from types import SimpleNamespace
from typing import Any, ClassVar, Literal, NoReturn
from unittest.mock import patch

from platform_ci.configuration import ConfigurationError, parse_project_configuration
from platform_ci.delivery import completion_receipt, legacy
from platform_ci.delivery.completion import CentralOutcome, CentralServices, central_evidence, ensure_central_completion, reactor_for, verified_modules
from platform_ci.delivery.enrollment import CentralAuthorization, EnrollmentError, parse_enrollment
from platform_ci.delivery.errors import DeliveryError
from platform_ci.delivery.manifest import ManifestFile
from platform_ci.delivery.maven_reactor import build_reactor
from tests.test_maven_reactor import retained
from tests.test_tag_driven_release import configuration

BUILD = "c" * 40
GAV = ["com.example:lib:1.2.3", "com.example:parent:1.2.3"]
PURLS = ["pkg:maven/com.example/lib@1.2.3", "pkg:maven/com.example/parent@1.2.3"]


class PendingError(Exception):
   """Stands in for qrun_central.CentralPending."""


class DuplicateError(Exception):
   """Stands in for qrun_central.CentralDuplicate."""


def entry(path: str, digest: str) -> ManifestFile:
   """A retained manifest file entry."""
   return ManifestFile(kind="artifact", name="maven-reactor", path=path, size=1, sha256=digest)


def artifact() -> Any:
   """The parts of a verified release artifact the orchestration reads."""
   return SimpleNamespace(build_sha=BUILD, run=SimpleNamespace(run_id="41", created_at="2026-11-05T01:02:03Z"), manifest=SimpleNamespace(digest="a" * 64))


def modules(version: str = "1.2.3", jar: bytes = b"jar-bytes") -> list[dict[str, Any]]:
   """Two verified modules of the retained reactor: a parent POM and a jar module."""
   return [
      {"groupId": "com.example", "artifactId": "lib", "version": version, "packaging": "jar", "files": [{"classifier": None, "extension": "jar", "data": jar}, {"classifier": None, "extension": "pom", "data": b"<lib/>"}]},
      {"groupId": "com.example", "artifactId": "parent", "version": version, "packaging": "pom", "files": [{"classifier": None, "extension": "pom", "data": b"<parent/>"}]},
   ]


FILES = (entry("artifact/p/pom.xml", "1" * 64), entry("artifact/p/lib.jar", "2" * 64))


def verified_purls(evidence: object, purls: Iterable[str]) -> list[str]:
   """Stand in for `qrun_central.verified_purls`: the purls, sorted."""
   return sorted(purls)


def refuse_as_duplicate(*_arguments: object, **_options: object) -> NoReturn:
   """Stand in for a `publish_central` that finds the deployment already at Central."""
   raise DuplicateError


class FakeCentral:
   """The interface of `qrun_central`, recording every upload."""

   def __init__(self, *, pending: bool = False) -> None:
      self.pending = pending
      self.uploads: list[str | None] = []
      self.central = SimpleNamespace(CentralPending=PendingError, CentralDuplicate=DuplicateError, publish_central=self.publish_central, verified_purls=verified_purls)

   def publish_central(self, evidence: dict[str, Any], *transport: object, deployment_id: str | None, save_deployment: Callable[[str], None], wait: bool) -> dict[str, Any]:
      """Record the upload; `transport` is the primitive's (signer, client, snapshot), positional."""
      snapshot = transport[2]
      assert wait is False
      self.uploads.append(deployment_id)
      if deployment_id is None:
         save_deployment("dep-1")
      if self.pending:
         raise PendingError("Original deployment is PENDING")
      if snapshot is not None:
         return {"central_status": "UPLOADED", "central_deployment": None, "central_purls": [], "central_files": [{"path": "a.jar", "sha256": "3" * 64}]}
      return {"central_status": "PUBLISHED", "central_deployment": "dep-1", "central_files": [], "central_purls": PURLS}


class Registry:
   """An in-memory generic registry: GET (404 as the client raises it) and PUT."""

   def __init__(self) -> None:
      self.files: dict[str, bytes] = {}

   def reader(self, method: str, path: str) -> tuple[int, bytes]:
      if path not in self.files:
         raise urllib.error.HTTPError(path, 404, "nf", Message(), io.BytesIO(b""))
      return 200, self.files[path]

   def writer(self, method: str, path: str, data: bytes, content_type: str) -> tuple[int, bytes]:
      self.files[path] = data
      return 201, b""


class Client:
   """A Central client double for the re-check of an existing receipt."""

   def __init__(self, state: str = "PUBLISHED") -> None:
      self.state = state

   def status(self, deployment_id: str) -> dict[str, Any]:
      return {"deploymentState": self.state, "deploymentId": deployment_id, "purls": PURLS}

   def read_snapshot(self, path: str) -> bytes:
      return b"x"


class Forge:
   """The repository document read for Central's public-route rule."""

   def __init__(self, *, private: bool = False, identity: int = 7) -> None:
      self.document = {"id": identity, "private": private}

   def get(self, path: str) -> object:
      return self.document


@dataclass(frozen=True)
class ServiceFactory:
   """A services factory over the in-memory registry, recording that its signer was closed."""

   made: CentralServices
   closed: list[bool] = field(default_factory=list[bool])

   def __call__(self) -> CentralServices:
      """Return the services."""
      return self.made


def services(registry: Registry, client: Client | None = None) -> ServiceFactory:
   """A services factory over the in-memory registry, recording that its signer was closed."""
   closed: list[bool] = []
   made = CentralServices(client=client or Client(), signer=object(), completion_writer=registry.writer, completion_reader=registry.reader, close=lambda: closed.append(True))
   return ServiceFactory(made, closed)


@dataclass(frozen=True)
class Scenario:
   """What one orchestration run varies; the defaults are the valid published release (`"default"` means the fixture's own authorization or services)."""

   version: str = "1.2.3"
   repository: str = "QRun-IO/r"
   forge: Forge | None = None
   authorization: CentralAuthorization | Literal["default"] | None = "default"
   factory: Callable[[], CentralServices] | Literal["default"] | None = "default"
   jar: bytes = b"jar-bytes"


def ensure(directory: Path, central: FakeCentral, registry: Registry, scenario: Scenario | None = None) -> CentralOutcome:
   """Run the orchestration for the fixture bundle with the fakes."""
   chosen = scenario or Scenario()
   authorization = CentralAuthorization("QRun-IO") if isinstance(chosen.authorization, str) else chosen.authorization
   factory = services(registry) if isinstance(chosen.factory, str) else chosen.factory
   with patch("platform_ci.delivery.completion.maven_modules", return_value=modules(chosen.version, chosen.jar)):
      return ensure_central_completion(
         repository=chosen.repository,
         repository_id=7,
         authorization=authorization,
         forge=chosen.forge or Forge(),
         artifact=artifact(),
         files=FILES,
         version=chosen.version,
         request_name="pub",
         channel="release",
         receipts=directory,
         services=factory,
         primitives=SimpleNamespace(central=central.central),
      )


class ReactorBindingTest(unittest.TestCase):
   """The receipt's hashes are the hashes of the deterministic reactor archive and its manifest.json, rebuilt from the retained files."""

   def test_the_manifest_hash_is_the_hash_of_the_manifest_json_and_everything_is_deterministic(self) -> None:
      with patch("platform_ci.delivery.completion.maven_modules", return_value=modules()):
         first, second = reactor_for(artifact(), FILES, "1.2.3"), reactor_for(artifact(), FILES, "1.2.3")
      self.assertEqual((first.archive, first.members), (second.archive, second.members))
      self.assertEqual(first.manifest_sha256, hashlib.sha256(first.members["manifest.json"]).hexdigest())
      self.assertEqual(first.archive_sha256, hashlib.sha256(first.archive).hexdigest())
      with patch("platform_ci.delivery.completion.maven_modules", return_value=modules(jar=b"other")):
         changed = reactor_for(artifact(), FILES, "1.2.3")
      self.assertNotEqual((changed.archive_sha256, changed.manifest_sha256), (first.archive_sha256, first.manifest_sha256))

   def test_a_version_the_build_did_not_record_is_refused(self) -> None:
      with patch("platform_ci.delivery.completion.maven_modules", return_value=modules("1.2.3")), self.assertRaises(DeliveryError):
         reactor_for(artifact(), FILES, "1.2.4")


def captured_bundle() -> tuple[Any, tuple[ManifestFile, ...], list[dict[str, Any]], dict[str, bytes]]:
   """An artifact whose bundle retains a reactor capture under `artifact/p/`: (artifact, retained entries, the modules the adapter grouped, the members)."""
   members = retained()
   data = {f"artifact/p/{name}": content for name, content in members.items()}
   entries = tuple(ManifestFile(kind="artifact", name="maven-reactor", path=path, size=len(content), sha256=hashlib.sha256(content).hexdigest()) for path, content in sorted(data.items()))

   def read_bytes(item: ManifestFile) -> bytes:
      return data[item.path]

   bundle = SimpleNamespace(read_bytes=read_bytes)
   subject = SimpleNamespace(build_sha=BUILD, run=SimpleNamespace(run_id="41", created_at="2026-11-05T01:02:03Z"), manifest=SimpleNamespace(digest="a" * 64), bundle=bundle)
   grouped = [
      {
         "groupId": "com.example",
         "artifactId": "lib",
         "version": "1.2.3",
         "packaging": "jar",
         "files": [{"classifier": None, "extension": "jar", "data": members["files/lib/lib-1.2.3.jar"]}, {"classifier": None, "extension": "pom", "data": members["files/lib/lib-1.2.3.pom"]}, {"classifier": None, "extension": "module", "data": members["files/lib/lib-1.2.3.module"]}],
      },
      {"groupId": "com.example", "artifactId": "parent", "version": "1.2.3", "packaging": "pom", "files": [{"classifier": None, "extension": "pom", "data": members["files/parent/parent-1.2.3.pom"]}]},
   ]
   return subject, entries, grouped, members


class RetainedCaptureTest(unittest.TestCase):
   """A captured reactor's manifest bytes stay exact for `manifest_sha256`; the `.module` file reaches Central; contradictions are refused."""

   def test_the_retained_manifest_bytes_give_the_receipt_hash_and_the_module_file_reaches_the_evidence(self) -> None:
      subject, entries, grouped, members = captured_bundle()
      with patch("platform_ci.delivery.completion.maven_modules", return_value=grouped):
         reactor = reactor_for(subject, entries, "1.2.3")
         evidence = central_evidence(verified_modules(subject, entries, "1.2.3"), "1.2.3")
      self.assertEqual(reactor.members["manifest.json"], members["manifest.json"])
      self.assertEqual(reactor.manifest_sha256, hashlib.sha256(members["manifest.json"]).hexdigest())
      self.assertIn("files/lib/lib-1.2.3.module", reactor.members)
      self.assertIn("module", {item["extension"] for module in evidence["modules"] for item in module["files"]})
      self.assertEqual([module["artifactId"] for module in evidence["modules"]], ["parent", "lib"])

   def test_a_hash_the_bundle_records_for_other_manifest_bytes_or_unretained_module_data_is_refused(self) -> None:
      subject, entries, grouped, _members = captured_bundle()
      tampered = tuple(replace(item, sha256="0" * 64) if item.path.endswith("/manifest.json") else item for item in entries)
      with patch("platform_ci.delivery.completion.maven_modules", return_value=grouped), self.assertRaises(DeliveryError):
         reactor_for(subject, tampered, "1.2.3")
      other = [{**module, "files": [{**item, "data": b"not retained"} if item["extension"] == "jar" else item for item in module["files"]]} for module in grouped]
      with patch("platform_ci.delivery.completion.maven_modules", return_value=other), self.assertRaises(DeliveryError):
         reactor_for(subject, entries, "1.2.3")

   def test_incomplete_or_ambiguous_metadata_and_foreign_roots_are_refused(self) -> None:
      subject, entries, grouped, _members = captured_bundle()
      for broken in (tuple(item for item in entries if not item.path.endswith("/SOURCE_COMMIT")), (*entries, entry("elsewhere/manifest.json", "1" * 64))):
         with self.subTest(len(broken)), patch("platform_ci.delivery.completion.maven_modules", return_value=grouped), self.assertRaises(DeliveryError):
            reactor_for(subject, broken, "1.2.3")

   def test_a_bundle_without_captured_metadata_still_generates_the_deterministic_archive(self) -> None:
      with patch("platform_ci.delivery.completion.maven_modules", return_value=modules()):
         generated = reactor_for(artifact(), FILES, "1.2.3")
      self.assertEqual(generated.members["manifest.json"], build_reactor(BUILD, "1.2.3", modules()).members["manifest.json"])


class CentralCompletionTest(unittest.TestCase):
   """Complete only with a terminal bound receipt; pending resumes the saved deployment; a repeat uploads nothing."""

   def test_a_published_release_is_recorded_with_the_reactor_hashes_in_the_configured_namespace(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         central, registry = FakeCentral(), Registry()
         outcome = ensure(Path(directory), central, registry)
         with patch("platform_ci.delivery.completion.maven_modules", return_value=modules()):
            reactor = reactor_for(artifact(), FILES, "1.2.3")
         self.assertEqual(outcome.state, "complete")
         (path,) = registry.files
         self.assertEqual(path, f"/api/packages/QRun-IO/generic/qrun-publication-completion/repo-7-run-41/{BUILD}.json")
         stored = json.loads(registry.files[path])
         self.assertEqual((stored["manifest_sha256"], stored["archive_sha256"], stored["gav"], stored["target"], stored["state"], stored["schema"]), (reactor.manifest_sha256, reactor.archive_sha256, GAV, "central-publisher", "PUBLISHED", "qrun-publication-completion/1"))

   def test_a_repeat_uploads_nothing_and_a_changed_byte_is_refused(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         central, registry = FakeCentral(), Registry()
         ensure(Path(directory), central, registry)
         self.assertEqual(ensure(Path(directory), central, registry).state, "complete")
         self.assertEqual(central.uploads, [None], "the second call read the immutable receipt and uploaded nothing")
         with self.assertRaises(DeliveryError):
            ensure(Path(directory), central, registry, Scenario(jar=b"relabelled"))
         with self.assertRaises(DeliveryError):
            ensure(Path(directory), central, registry, Scenario(factory=services(registry, Client("FAILED"))))

   def test_a_pending_deployment_is_not_complete_and_is_resumed_not_repeated(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         central, registry = FakeCentral(pending=True), Registry()
         first, second = ensure(Path(directory), central, registry), ensure(Path(directory), central, registry)
         self.assertEqual((first.state, second.state, registry.files), ("pending", "pending", {}))
         self.assertEqual(central.uploads, [None, "dep-1"], "the saved original deployment id is reused, never a second upload")
         record = json.loads((Path(directory) / "QRun-IO/r" / f"{BUILD}.pub.release.central-upload.json").read_text(encoding="utf-8"))
         self.assertEqual((record["deployment_id"], record["version"], record["source_commit"]), ("dep-1", "1.2.3", BUILD))

   def test_a_saved_upload_for_different_bytes_is_refused(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         target = Path(directory) / "QRun-IO/r" / f"{BUILD}.pub.release.central-upload.json"
         target.parent.mkdir(parents=True)
         target.write_text(json.dumps({"source_commit": BUILD, "archive_sha256": "0" * 64, "version": "1.2.3", "deployment_id": "old"}), encoding="utf-8")
         with self.assertRaises(DeliveryError):
            ensure(Path(directory), FakeCentral(), Registry())

   def test_a_snapshot_is_uploaded_and_recorded_with_its_files(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         registry = Registry()
         outcome = ensure(Path(directory), FakeCentral(), registry, Scenario(version="1.2.3-SNAPSHOT"))
         self.assertEqual((outcome.state, outcome.completion and outcome.completion["target"]), ("complete", "central-snapshots"))

   def test_the_destination_is_authorization_data_for_any_owner_and_nothing_is_switched_on_a_name(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         registry = Registry()
         outcome = ensure(Path(directory), FakeCentral(), registry, Scenario(repository="Acme/r", authorization=CentralAuthorization("Acme")))
         self.assertEqual(outcome.state, "complete")
         self.assertTrue(next(iter(registry.files)).startswith("/api/packages/Acme/generic/qrun-publication-completion/"))
         with self.assertRaises(DeliveryError):  # a repository outside the configured owner is refused by the receipt's own validation
            ensure(Path(directory), FakeCentral(), Registry(), Scenario(repository="Other/r", authorization=CentralAuthorization("Acme")))

   def test_missing_authorization_private_repositories_identity_credentials_and_duplicates_fail_closed(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         base = Path(directory)
         with self.assertRaises(DeliveryError):
            ensure(base, FakeCentral(), Registry(), Scenario(forge=Forge(private=True)))
         self.assertEqual(ensure(base, FakeCentral(), Registry(), Scenario(forge=Forge(private=True), authorization=CentralAuthorization("QRun-IO", allow_private=True))).state, "complete")
         for name, scenario in (("authorization", Scenario(authorization=None)), ("forge", Scenario(forge=Forge(identity=8))), ("factory", Scenario(factory=None))):
            with self.subTest(name), self.assertRaises(DeliveryError):
               ensure(base, FakeCentral(), Registry(), scenario)
         central = FakeCentral()
         central.central.publish_central = refuse_as_duplicate
         with self.assertRaises(DeliveryError):
            ensure(base / "dup", central, Registry())

   def test_the_signer_is_always_closed_and_an_unvendored_primitive_is_an_explicit_error(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         registry = Registry()
         factory = services(registry)
         ensure(Path(directory), FakeCentral(), registry, Scenario(factory=factory))
         self.assertEqual(factory.closed, [True])
      with patch("platform_ci.delivery.legacy.importlib.import_module", side_effect=ModuleNotFoundError("qrun_central")), self.assertRaisesRegex(DeliveryError, "not vendored"):
         legacy.central_primitives()


class CompletionReceiptTest(unittest.TestCase):
   """The generic receipt module keeps the frozen primitive's rules and takes the owner as data."""

   def receipt(self, owner: str = "QRun-IO") -> dict[str, Any]:
      return {
         "schema": "qrun-publication-completion/1",
         "repository": f"{owner}/r",
         "repository_id": 7,
         "run_id": 41,
         "source_commit": BUILD,
         "version": "1.2.3",
         "archive_sha256": "e" * 64,
         "manifest_sha256": "f" * 64,
         "gav": GAV,
         "target": "central-publisher",
         "state": "PUBLISHED",
         "deployment_id": "dep-1",
         "files": [],
         "purls": PURLS,
      }

   def test_the_frozen_rules_still_refuse_every_incomplete_or_contradictory_receipt(self) -> None:
      completion_receipt.validate("QRun-IO", self.receipt())
      for changes in ({"state": "PENDING"}, {"deployment_id": None}, {"files": [{"path": "a", "sha256": "1" * 64}]}, {"purls": PURLS[:1]}, {"gav": GAV[::-1]}, {"version": "1.2.4"}, {"manifest_sha256": "x"}, {"schema": "other/1"}, {"target": "other"}, {"repository": "Other/r"}):
         with self.subTest(changes), self.assertRaises(ValueError):
            completion_receipt.validate("QRun-IO", self.receipt() | changes)

   def test_publish_is_immutable_and_read_back_independently(self) -> None:
      registry = Registry()
      completion_receipt.publish("QRun-IO", self.receipt(), registry.writer, registry.reader)
      completion_receipt.publish("QRun-IO", self.receipt(), registry.writer, registry.reader)  # identical replay is accepted
      with self.assertRaises(ValueError):
         completion_receipt.publish("QRun-IO", self.receipt() | {"archive_sha256": "d" * 64}, registry.writer, registry.reader)

   def test_for_the_owner_the_frozen_primitive_hardcodes_the_location_and_bytes_are_identical(self) -> None:
      try:
         frozen = importlib.import_module("qrun_completion_receipt")  # the vendored original (its directory is on `sys.path` through `legacy`), used only to prove equivalence
      except ModuleNotFoundError:
         self.skipTest("the frozen qrun_completion_receipt is not vendored here")
      self.assertEqual(frozen.location(7, 41, BUILD), completion_receipt.location("QRun-IO", 7, 41, BUILD))
      self.assertEqual(frozen.encode(self.receipt()), completion_receipt.encode("QRun-IO", self.receipt()))


class EnrollmentCentralTest(unittest.TestCase):
   """The registry destination is explicit, company-bound data."""

   def document(self, central: object) -> dict[str, Any]:
      return {
         "schema": 1,
         "company": "qqq",
         "repository": "QRun-IO/r",
         "repository_id": 7,
         "platform_revisions": ["a" * 40],
         "max_artifact_bytes": 1000,
         "max_runs_scanned": 10,
         "workflow_references": [f"self:QRun-IO/platform-ci/.gitea/scoped_workflows/pipeline.yaml@{'b' * 40}"],
         "destinations": [{"destination": "d", "formats": ["maven"], "channels": ["release"], "target": {"kind": "package", "owner": "QRun-IO"}}],
         "central": central,
      }

   def test_the_central_object_is_validated_and_stays_inside_the_company(self) -> None:
      parsed = parse_enrollment(self.document({"completion_owner": "QRun-IO", "allow_private": True}))
      self.assertEqual(parsed.central, CentralAuthorization("QRun-IO", True))
      for bad in ({"completion_owner": "KofTwentyTwo"}, {}, {"completion_owner": "QRun-IO", "other": 1}, {"completion_owner": "QRun-IO", "allow_private": "yes"}):
         with self.subTest(bad), self.assertRaises(EnrollmentError):
            parse_enrollment(self.document(bad))


class FakeReleases:
   """The forge release endpoints: GET by tag (404 until a record exists) and POST."""

   def __init__(self) -> None:
      self.records: dict[str, dict[str, Any]] = {}
      self.posts = 0

   def __call__(self, method: str, path: str, body: dict[str, Any] | None = None) -> tuple[int, Any]:
      if method == "GET":
         tag = path.rsplit("/", 1)[1]
         if tag not in self.records:
            raise urllib.error.HTTPError(path, 404, "nf", Message(), io.BytesIO(b""))
         return 200, self.records[tag]
      assert body is not None
      self.posts += 1
      self.records[body["tag_name"]] = {"tag_name": body["tag_name"], "draft": False}
      return 201, self.records[body["tag_name"]]


class ReleaseRecordTest(unittest.TestCase):
   """The frozen record finalizer is idempotent: a retry posts nothing and never duplicates or rewrites a record."""

   def test_a_retry_confirms_the_existing_record_without_posting_again(self) -> None:
      releases = FakeReleases()
      first = legacy.ensure_release_record(releases, "O/r", "1.2.3", "O", "maven")
      second = legacy.ensure_release_record(releases, "O/r", "1.2.3", "O", "maven")
      self.assertEqual((first, second, releases.posts), ({"release": "v1.2.3"}, {"release": "v1.2.3"}, 1))

   def test_a_draft_or_mismatched_existing_record_is_refused(self) -> None:
      releases = FakeReleases()
      releases.records["v1.2.3"] = {"tag_name": "v1.2.3", "draft": True}
      with self.assertRaises(DeliveryError):
         legacy.ensure_release_record(releases, "O/r", "1.2.3", "O", "maven")
      self.assertEqual(releases.posts, 0)

   def test_a_version_the_native_contract_does_not_know_is_refused_before_any_request(self) -> None:
      releases = FakeReleases()
      with self.assertRaises(DeliveryError):
         legacy.ensure_release_record(releases, "O/r", "1.2", "O", "maven")
      self.assertEqual(releases.posts, 0)


class CompletionConfigurationTest(unittest.TestCase):
   """The repository's own channel asks for terminal evidence, and only where it can apply."""

   CHANNELS: ClassVar[dict[str, Any]] = {"release": {"trigger": {"push_tags": ["v*"]}, "tag_policy": "stable", "release_record": True, "central_completion": True}}

   def test_release_record_needs_a_tag_policy(self) -> None:
      with self.assertRaises(ConfigurationError):
         parse_project_configuration(configuration(channels={"release": {"trigger": {"push_branches": ["main"]}, "release_record": True}}))

   def test_central_completion_is_for_maven_publications_only(self) -> None:
      container = {"artifacts": {"out": {"paths": ["out/*"], "format": "container"}}}
      publication = {"out": {"component": "checks", "artifact": "out", "destination": "dest", "channels": ["release"]}}
      with self.assertRaises(ConfigurationError):
         parse_project_configuration(configuration(channels=self.CHANNELS, component=container, publications=publication))
      parse_project_configuration(configuration(channels={"release": {"trigger": {"push_tags": ["v*"]}, "tag_policy": "stable"}}, component=container, publications=publication))


if __name__ == "__main__":
   unittest.main()