platform-ci/tests/test_delivery_contracts.py

"""Publication and versioning configuration, plan requests, build inputs in the manifest, versions, and the workflow's credential boundary."""

import copy
import hashlib
import json
import re
import tempfile
import unittest
from collections.abc import Callable, Mapping
from pathlib import Path
from typing import cast, override

import yaml

from platform_ci.build_inputs import compute_build_inputs
from platform_ci.configuration import ConfigurationError, parse_project_configuration
from platform_ci.packaging import PackagingError, RunIdentity, collect_outputs, manifest_digest
from platform_ci.plan import Target
from platform_ci.reporting import render_report
from platform_ci.versions import UNDECLARED, UNVERSIONED, parse_declared_version, read_declared_version
from tests.support import EXAMPLES_DIRECTORY, PLATFORM_REVISION, SOURCE_REVISION, SOURCE_TREE, event, example_configuration, example_mapping, plan_for

ROOT = Path(__file__).resolve().parent.parent
WORKFLOW_TEXT = (ROOT / ".gitea" / "scoped_workflows" / "pipeline.yaml").read_text(encoding="utf-8")
WORKFLOW = cast("dict[str, object]", yaml.safe_load(WORKFLOW_TEXT))
JOBS = cast("dict[str, dict[str, object]]", WORKFLOW["jobs"])


def _publication_mapping() -> dict[str, object]:
   """The documented publication example as a mutable mapping."""
   return example_mapping("maven-publication.ci.yaml")


def _publications(configuration: dict[str, object]) -> dict[str, dict[str, object]]:
   """The release.publications mapping of a parsed example."""
   return cast("dict[str, dict[str, object]]", cast("dict[str, object]", configuration["release"])["publications"])


def _library_artifacts(configuration: dict[str, object]) -> dict[str, dict[str, object]]:
   """The library component's artifacts mapping."""
   return cast("dict[str, dict[str, object]]", cast("dict[str, dict[str, object]]", configuration["components"])["library"]["artifacts"])


class PublicationConfigurationTest(unittest.TestCase):
   """The approved publication and versioning fields validate strictly; nothing can carry an address or a credential."""

   def test_documented_example_is_valid_and_typed(self) -> None:
      project = example_configuration("maven-publication.ci.yaml")
      self.assertEqual([(p.name, p.format, p.platform, p.destination, p.channels, p.deploy) for p in project.publications], [("library-image", "container", "linux-amd64", "company-images", ("release",), "production-gitops"), ("library-packages", "maven", "linux-amd64", "company-packages", ("release", "snapshot"), "")])
      self.assertEqual({channel.name: channel.versioning for channel in project.channels}, {"release": "maven-release", "snapshot": "maven-snapshot"})

   def test_configuration_without_publications_is_unchanged(self) -> None:
      self.assertEqual(example_configuration("maven.ci.yaml").publications, ())

   def test_each_invalid_publication_field_fails_with_its_path(self) -> None:
      changes: dict[str, tuple[Callable[[dict[str, object]], object], str]] = {
         "unknown component": (lambda c: _publications(c)["library-packages"].update(component="ghost"), "no component named 'ghost'"),
         "unknown artifact": (lambda c: _publications(c)["library-packages"].update(artifact="ghost"), "declares no artifact 'ghost'"),
         "artifact without a format": (lambda c: _library_artifacts(c)["packages"].pop("format"), "declares no format"),
         "undeclared channel": (lambda c: _publications(c)["library-packages"].update(channels=["nightly"]), "declared release channels"),
         "duplicate channels": (lambda c: _publications(c)["library-packages"].update(channels=["snapshot", "snapshot"]), "distinct names"),
         "url destination": (lambda c: _publications(c)["library-packages"].update(destination="https://registry.example/x"), "logical destination name"),
         "credential field": (lambda c: _publications(c)["library-packages"].update({"token": "x"}), "release.publications.library-packages.token: unsupported field"),
         "owner field": (lambda c: _publications(c)["library-packages"].update(owner="Other"), "owner: unsupported field"),
         "deploy of a package": (lambda c: _publications(c)["library-packages"].update(deploy="production-gitops"), "only a container artifact"),
         "npm format from maven": (lambda c: _library_artifacts(c)["packages"].update(format="npm"), "needs build tool 'node'"),
         "unknown format": (lambda c: _library_artifacts(c)["packages"].update(format="rpm"), "unsupported artifact format 'rpm'"),
         "unknown versioning": (lambda c: cast("dict[str, dict[str, object]]", cast("dict[str, object]", c["release"])["channels"])["snapshot"].update(versioning="latest"), "unsupported rule 'latest'"),
         "artifact source": (lambda c: _library_artifacts(c)["packages"].update(source="reactor-outputs"), "source: described by the approved design but not implemented"),
         "ambiguous platform": (lambda c: cast("dict[str, object]", cast("dict[str, object]", c["components"])["library"]).update(platforms=["linux-amd64"]) or _publications(c)["library-packages"].update(platform="windows-amd64"), "does not build for 'windows-amd64'"),
      }
      for name, (change, message) in changes.items():
         with self.subTest(name):
            configuration = _publication_mapping()
            change(configuration)
            with self.assertRaisesRegex(ConfigurationError, re.escape(message)):
               parse_project_configuration(configuration)

   def test_a_component_with_several_platforms_needs_an_explicit_publication_platform(self) -> None:
      configuration = _publication_mapping()
      components = cast("dict[str, dict[str, object]]", configuration["components"])
      components["library"]["platforms"] = ["linux-amd64", "windows-amd64"]
      components["library"]["build"] = {"tool": "go", "arguments": ["build"]}
      components["library"]["toolchains"] = {"go": "1"}
      _library_artifacts(configuration)["packages"].pop("format")
      _publications(configuration).pop("library-packages")
      _publications(configuration).pop("library-image")
      _library_artifacts(configuration)["image"]["format"] = "container"
      _publications(configuration)["library-image"] = {"component": "library", "artifact": "image", "destination": "company-images", "channels": ["release"]}
      with self.assertRaisesRegex(ConfigurationError, "builds for several platforms"):
         parse_project_configuration(configuration)
      _publications(configuration)["library-image"]["platform"] = "linux-amd64"
      self.assertEqual(parse_project_configuration(configuration).publications[0].platform, "linux-amd64")


class PlanRequestTest(unittest.TestCase):
   """The plan lists requested publications only for eligible channels and exposes no address or credential."""

   def test_requests_follow_channel_eligibility(self) -> None:
      project = example_configuration("maven-publication.ci.yaml")
      develop = plan_for(project, event("push", "refs/heads/develop"))["delivery"]
      main = plan_for(project, event("push", "refs/heads/main", default_branch="main"))["delivery"]
      pull = plan_for(project, event())["delivery"]
      self.assertEqual([(p["name"], p["channels"], p["target"]) for p in develop["publications"]], [("library-packages", ["snapshot"], "library-linux-amd64")])
      self.assertEqual([(p["name"], p["channels"]) for p in main["publications"]], [("library-image", ["release"]), ("library-packages", ["release"])])
      self.assertEqual(pull["publications"], [])
      self.assertFalse(develop["implemented"])

   def test_request_data_has_no_address_credential_or_repository_identity(self) -> None:
      delivery = plan_for(example_configuration("maven-publication.ci.yaml"), event("push", "refs/heads/main", default_branch="main"))["delivery"]
      text = json.dumps(delivery)
      for forbidden in ("token", "http", "owner", "KofTwentyTwo", "QRun"):
         self.assertNotIn(forbidden, text.replace("trusted publisher", ""))

   def test_plan_is_identical_for_two_unrelated_repository_event_contexts(self) -> None:
      project = example_configuration("maven-publication.ci.yaml")
      first = plan_for(project, event("push", "refs/heads/main", default_branch="main"))
      second = plan_for(copy.deepcopy(project), event("push", "refs/heads/main", default_branch="main"))
      self.assertEqual(first, second)

   def test_report_lists_what_was_requested_and_claims_no_delivery(self) -> None:
      plan = plan_for(example_configuration("maven-publication.ci.yaml"), event("push", "refs/heads/develop"))
      text, passed = render_report(plan, "success", "success")
      self.assertTrue(passed)
      self.assertIn("library-packages (maven to company-packages on snapshot)", text)
      self.assertIn("NOT PERFORMED BY THIS WORKFLOW", text)
      self.assertNotIn("Published", text)


class BuildInputsManifestTest(unittest.TestCase):
   """A successful build's manifest records everything a later release needs to prove equivalence; a failed one is never reusable."""

   @override
   def setUp(self) -> None:
      """Create Maven-shaped output and the publication example's library target."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.source = Path(directory.name) / "checkout"
      self.output = Path(directory.name) / "bundle"
      (self.source / "target").mkdir(parents=True)
      (self.source / "target" / "widget.jar").write_bytes(b"jar")
      (self.source / "pom.xml").write_text('<project xmlns="http://maven.apache.org/POM/4.0.0"><version>1.2.3-SNAPSHOT</version></project>', encoding="utf-8")
      self.target: Target = plan_for(example_configuration("maven-publication.ci.yaml"), event())["targets"][0]
      (self.source / "image.tar").write_bytes(b"image")
      (self.source / "target" / "surefire-reports").mkdir()
      (self.source / "target" / "surefire-reports" / "TEST-a.xml").write_bytes(b"<testsuite tests='1'/>")
      self.identity = RunIdentity(SOURCE_REVISION, PLATFORM_REVISION, "42", "1", SOURCE_TREE)

   def _inputs(self, manifest: Mapping[str, object]) -> dict[str, object]:
      return cast("dict[str, object]", manifest["build_inputs"])

   def test_manifest_records_tree_version_toolchain_platform_and_digests(self) -> None:
      manifest = collect_outputs(self.target, self.source, self.output, self.identity, build_succeeded=True)
      inputs = self._inputs(manifest)
      self.assertEqual((inputs["source_tree"], inputs["version"], inputs["platform"]), (SOURCE_TREE, "1.2.3-SNAPSHOT", "linux-amd64"))
      self.assertEqual(inputs["toolchain"], {"java": "21", "runtime": self.target["image"]})
      self.assertEqual(manifest["digest"], manifest_digest(manifest))
      self.assertEqual(json.loads((self.output / "manifest.json").read_text(encoding="utf-8")), manifest)

   def test_build_inputs_are_bound_by_the_manifest_digest(self) -> None:
      manifest = collect_outputs(self.target, self.source, self.output, self.identity, build_succeeded=True)
      altered = copy.deepcopy(manifest)
      self._inputs(altered)["version"] = "9.9.9"
      self.assertNotEqual(manifest_digest(altered), manifest["digest"])

   def test_every_input_that_a_rebuild_could_change_changes_the_record(self) -> None:
      base = compute_build_inputs(self.target, SOURCE_TREE, "1.0.0")
      changed_target = copy.deepcopy(self.target)
      changed_target["commands"] = [["mvn", "-DskipITs", "verify"]]
      other_image = copy.deepcopy(self.target)
      other_image["image"] = f"registry.example/other@sha256:{'8' * 64}"
      variants = {"tree": compute_build_inputs(self.target, "e" * 40, "1.0.0"), "version": compute_build_inputs(self.target, SOURCE_TREE, "1.0.1"), "commands": compute_build_inputs(changed_target, SOURCE_TREE, "1.0.0"), "runtime image": compute_build_inputs(other_image, SOURCE_TREE, "1.0.0")}
      for name, variant in variants.items():
         with self.subTest(name):
            self.assertNotEqual(variant, base)
      self.assertEqual(compute_build_inputs(copy.deepcopy(self.target), SOURCE_TREE, "1.0.0"), base)

   def test_failed_build_records_no_inputs_and_needs_no_tree(self) -> None:
      manifest = collect_outputs(self.target, self.source, self.output, RunIdentity(SOURCE_REVISION, PLATFORM_REVISION, "42", "1"), build_succeeded=False)
      self.assertNotIn("build_inputs", manifest)

   def test_successful_build_without_a_source_tree_cannot_be_packaged(self) -> None:
      with self.assertRaisesRegex(PackagingError, "source tree id is required"):
         collect_outputs(self.target, self.source, self.output, RunIdentity(SOURCE_REVISION, PLATFORM_REVISION, "42", "1"), build_succeeded=True)
      self.assertFalse(self.output.exists())

   def test_a_malformed_tree_id_is_refused(self) -> None:
      with self.assertRaisesRegex(PackagingError, "full 40 character tree id"):
         collect_outputs(self.target, self.source, self.output, RunIdentity(SOURCE_REVISION, PLATFORM_REVISION, "42", "1", "not-a-tree"), build_succeeded=True)


class DeclaredVersionTest(unittest.TestCase):
   """Versions come from the component's own native file, as a literal only."""

   def test_maven_literal_parent_and_revision_property(self) -> None:
      ns = 'xmlns="http://maven.apache.org/POM/4.0.0"'
      self.assertEqual(parse_declared_version("maven", f"<project {ns}><version>2.0.0</version></project>".encode()), "2.0.0")
      self.assertEqual(parse_declared_version("maven", f"<project {ns}><parent><version>3.1.0</version></parent></project>".encode()), "3.1.0")
      self.assertEqual(parse_declared_version("maven", f"<project {ns}><version>${{revision}}</version><properties><revision>4.0.0-SNAPSHOT</revision></properties></project>".encode()), "4.0.0-SNAPSHOT")

   def test_expressions_and_missing_versions_are_undeclared_not_guessed(self) -> None:
      ns = 'xmlns="http://maven.apache.org/POM/4.0.0"'
      self.assertEqual(parse_declared_version("maven", f"<project {ns}><version>${{project.x}}</version></project>".encode()), UNDECLARED)
      self.assertEqual(parse_declared_version("maven", f"<project {ns}></project>".encode()), UNDECLARED)
      self.assertEqual(parse_declared_version("node", b'{"name": "x"}'), UNDECLARED)

   def test_npm_version_and_tools_without_a_version_file(self) -> None:
      self.assertEqual(parse_declared_version("node", b'{"version": "5.6.7"}'), "5.6.7")
      self.assertEqual(parse_declared_version("go", b"anything"), UNVERSIONED)

   def test_dtd_and_entities_are_refused(self) -> None:
      with self.assertRaisesRegex(Exception, "DTD"):
         parse_declared_version("maven", b'<!DOCTYPE x [<!ENTITY a "b">]><project/>')

   def test_missing_linked_or_unreadable_version_file_is_undeclared(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         self.assertEqual(read_declared_version("maven", Path(directory)), UNDECLARED)
         self.assertEqual(read_declared_version("go", Path(directory)), UNVERSIONED)


class WorkflowBoundaryTest(unittest.TestCase):
   """Build jobs never receive release credentials, and the workflow only forwards what the trusted publisher verifies itself."""

   def test_the_workflow_has_no_publish_or_deploy_job_and_only_the_isolated_tag_job_has_a_secret_or_write_scope(self) -> None:
      """UNTESTED edit. The one trusted write phase is `release-tag`: no build job, the application is never checked out there, and it alone
      references the existing signing secret and holds contents:write."""
      self.assertEqual(sorted(JOBS), ["prepare", "release-tag", "report", "secrets", "verify", "verify-native", "verify-pool", "verify-pool-docker"])
      code = re.sub(r"(?m)^\s*#.*$", "", WORKFLOW_TEXT)
      # the existing read-only checkout slot reaches ONLY the dependency-fetch step of the three container verify jobs; the signing secret only the tag job
      self.assertEqual(re.findall(r"\$\{\{\s*secrets\.([A-Za-z0-9_]+)", code), ["PUBLISHER_CHECKOUT_TOKEN"] * 3 + ["QRUN_TAG_SIGNING_KEY"])
      for name in ("verify", "verify-pool", "verify-pool-docker"):
         holders = [step.get("name") for step in cast("list[dict[str, object]]", JOBS[name]["steps"]) if "secrets." in json.dumps(step)]
         self.assertEqual(holders, ["Fetch pinned private dependencies"], name)
      self.assertNotIn("password", code.lower())
      tag_job = JOBS["release-tag"]
      self.assertEqual(tag_job["permissions"], {"contents": "write", "actions": "read"})
      self.assertEqual(tag_job["needs"], ["prepare", "report"])
      for name, job in JOBS.items():
         if name != "release-tag":
            self.assertNotIn("write", json.dumps(job.get("permissions", {})), name)
            self.assertNotIn("SIGNING_KEY", json.dumps(job), name)
      for step in cast("list[dict[str, object]]", tag_job["steps"]):
         if step.get("uses") == "builtin:checkout":  # the only checkout is the pinned platform implementation, never the application
            self.assertEqual(cast("dict[str, object]", step["with"])["repository"], "${{ env.PLATFORM_CI_REPOSITORY }}")
      self.assertEqual([step.get("run") for step in cast("list[dict[str, object]]", tag_job["steps"]) if step.get("run")], ["python3 -P -m platform_ci release-tag"])
      # the credential-like values are the job's own token: read-only for the plan and reuse steps of prepare, and the write token of the tag job only
      self.assertEqual(
         re.findall(r"\$\{\{[^}]*token[^}]*\}\}", code, re.IGNORECASE),
         ["${{ github.token }}", "${{ github.token }}", *["${{ secrets.PUBLISHER_CHECKOUT_TOKEN }}"] * 3, "${{ github.token }}"],
      )
      (plan_step,) = [step for step in cast("list[dict[str, object]]", JOBS["prepare"]["steps"]) if step.get("id") == "plan"]
      self.assertEqual(cast("dict[str, str]", plan_step["env"])["CI_FORGE_TOKEN"], "${{ github.token }}")
      (reuse,) = [step for step in cast("list[dict[str, object]]", JOBS["prepare"]["steps"]) if step.get("id") == "reuse"]
      self.assertEqual(cast("dict[str, str]", reuse["env"])["CI_FORGE_TOKEN"], "${{ github.token }}")
      self.assertEqual(JOBS["prepare"]["permissions"], {"contents": "read", "actions": "read"})
      self.assertEqual(WORKFLOW["permissions"], {"contents": "read"})

   def test_the_source_tree_flows_from_prepare_to_every_package_step(self) -> None:
      outputs = cast("dict[str, str]", JOBS["prepare"]["outputs"])
      self.assertEqual(outputs["source_tree"], "${{ steps.source.outputs.tree }}")
      for name in ("verify", "verify-native"):
         (package,) = [step for step in cast("list[dict[str, object]]", JOBS[name]["steps"]) if step.get("id") == "package"]
         self.assertEqual(cast("dict[str, str]", package["env"])["CI_SOURCE_TREE"], "${{ needs.prepare.outputs.source_tree }}")

   def test_the_publisher_is_not_reachable_from_the_workflow(self) -> None:
      self.assertNotIn("platform_ci.delivery", re.sub(r"(?m)^#.*$", "", WORKFLOW_TEXT))
      self.assertNotIn("platform_ci delivery", WORKFLOW_TEXT)


class FrozenPrimitiveIntegrityTest(unittest.TestCase):
   """The vendored publication primitives are byte-identical to the recorded originals."""

   def test_every_frozen_file_matches_its_recorded_digest(self) -> None:
      directory = ROOT / "publisher_frozen"
      receipts = cast("list[dict[str, str]]", json.loads((directory / "receipts.json").read_text(encoding="utf-8")))
      self.assertGreater(len(receipts), 10)
      for entry in receipts:
         with self.subTest(entry["file"]):
            self.assertEqual(hashlib.sha256((directory / entry["file"]).read_bytes()).hexdigest(), entry["sha256"])

   def test_only_the_bridge_imports_the_frozen_modules(self) -> None:
      frozen = {path.stem for path in (ROOT / "publisher_frozen").glob("*.py")}
      pattern = re.compile(r"^\s*(?:import|from)\s+(" + "|".join(sorted(frozen)) + r")\b", re.MULTILINE)
      offenders = [path.name for path in (ROOT / "platform_ci").rglob("*.py") if path.name != "legacy.py" and pattern.search(path.read_text(encoding="utf-8"))]
      self.assertEqual(offenders, [])

   def test_example_publication_file_is_listed_with_the_other_examples(self) -> None:
      self.assertTrue((EXAMPLES_DIRECTORY / "maven-publication.ci.yaml").is_file())