platform-ci/tests/test_runtimes.py

"""Runtime catalog resolution, and proof that the workflow and recipes never install tools or use mutable images."""

import re
import tempfile
import unittest
from pathlib import Path
from typing import cast, override

import yaml

from platform_ci.__main__ import main
from platform_ci.configuration import load_project_configuration
from platform_ci.runtimes import DEFAULT_CATALOG_PATH, RuntimeCatalogError, is_immutable_image_reference, load_runtime_catalog, parse_runtime_catalog
from tests.support import EXAMPLES_DIRECTORY, JAVA_IMAGE, NODE_IMAGE, PLATFORM_REVISION, SOURCE_REVISION, TEST_RUNTIME_CATALOG, event, example_configuration, plan_for

ROOT = Path(__file__).resolve().parent.parent
WORKFLOW_PATH = ROOT / ".gitea" / "scoped_workflows" / "pipeline.yaml"
CONTROL_IMAGE = "git.kof22.com/koftwentytwo/platform-ci-control@sha256:11f8cb328e7fc7aca8e0c39ed979b33cb1db411a59460baae3e0d48281a15e3c"
DIGEST = "sha256:" + "3" * 64


def catalog_with(java_image: object) -> dict[str, object]:
   """Return raw catalog data whose only entry is Java 21 with the given image value."""
   return {"schema": 1, "toolchains": {"java": {"21": {"image": java_image}}}}


class ImmutableReferenceTest(unittest.TestCase):
   """Only a reference that names a sha256 digest is an exact image."""

   def test_digest_references_are_accepted(self) -> None:
      """A digest may follow a repository, a registry port, or a repository and tag."""
      for reference in (f"registry.example/team/tool@{DIGEST}", f"registry.example:5000/team/tool:21@{DIGEST}", f"tool@{DIGEST}"):
         with self.subTest(reference):
            self.assertTrue(is_immutable_image_reference(reference))

   def test_mutable_or_malformed_references_are_rejected(self) -> None:
      """Tags, short or uppercase digests, and other algorithms can all be moved or misread."""
      for reference in ("maven:3.9.11-eclipse-temurin-21", "node:22.20.0-bookworm", "tool:latest", "tool@sha256:" + "3" * 63, "tool@sha256:" + "A" * 64, "tool@sha512:" + "3" * 64, f"tool@{DIGEST} ", ""):
         with self.subTest(reference):
            self.assertFalse(is_immutable_image_reference(reference))


class CatalogResolutionTest(unittest.TestCase):
   """A toolchain version resolves to its exact image, or fails with a message that says why."""

   def test_resolves_the_exact_recorded_image(self) -> None:
      """The recorded digest reference is returned unchanged."""
      self.assertEqual(TEST_RUNTIME_CATALOG.resolve("java", "21"), JAVA_IMAGE)
      self.assertEqual(TEST_RUNTIME_CATALOG.resolve("node", "22.20.0"), NODE_IMAGE)

   def test_a_tag_only_image_is_rejected_when_the_catalog_is_read(self) -> None:
      """The catalog cannot hold a mutable tag, so a mutable tag can never be resolved."""
      for image in ("maven:3.9.11-eclipse-temurin-21", "registry.example/java@sha256:abc", 21, True):
         with self.subTest(image), self.assertRaisesRegex(RuntimeCatalogError, r"toolchains\.java\.21\.image: expected null or a reference pinned by @sha256"):
            parse_runtime_catalog(catalog_with(image))

   def test_unqualified_runtime_fails_clearly(self) -> None:
      """A listed entry with no digest says it is unqualified instead of picking an image."""
      catalog = parse_runtime_catalog(catalog_with(None))
      with self.assertRaisesRegex(RuntimeCatalogError, r"java 21 runtime is unqualified.*no immutable image digest"):
         catalog.resolve("java", "21")

   def test_unsupported_version_names_the_supported_ones(self) -> None:
      """A version that was never listed is unsupported, and the message lists what is."""
      with self.assertRaisesRegex(RuntimeCatalogError, r"java 17 has no runtime in runtime/catalog.yaml; supported java versions: 21"):
         TEST_RUNTIME_CATALOG.resolve("java", "17")

   def test_unlisted_toolchain_is_unsupported(self) -> None:
      """A toolchain with no entries says nothing is supported rather than raising a lookup error."""
      with self.assertRaisesRegex(RuntimeCatalogError, r"supported java versions: none"):
         parse_runtime_catalog({"schema": 1, "toolchains": {}}).resolve("java", "21")

   def test_malformed_catalogs_are_rejected(self) -> None:
      """Unknown toolchains, unquoted versions, missing image fields, extra fields and other schemas are errors."""
      cases: dict[str, tuple[object, str]] = {
         "unknown toolchain": ({"schema": 1, "toolchains": {"cobol": {}}}, r"catalog\.toolchains\.cobol: unsupported entry"),
         "unquoted version": ({"schema": 1, "toolchains": {"java": {21: {"image": None}}}}, r"keys must be quoted strings"),
         "bad version": ({"schema": 1, "toolchains": {"java": {"latest": {"image": None}}}}, r"expected a quoted version"),
         "missing image": ({"schema": 1, "toolchains": {"java": {"21": {}}}}, r"image: required field is missing"),
         "extra field": ({"schema": 1, "toolchains": {"java": {"21": {"image": None, "tag": "21"}}}}, r"21\.tag: unsupported entry"),
         "other schema": ({"schema": 2, "toolchains": {}}, r"only schema 1 is supported"),
         "not a mapping": ([], r"catalog: expected a mapping"),
      }
      for description, (raw, message) in cases.items():
         with self.subTest(description), self.assertRaisesRegex(RuntimeCatalogError, message):
            parse_runtime_catalog(raw)

   def test_a_duplicate_version_is_rejected_instead_of_keeping_the_last(self) -> None:
      """The catalog file is read with the same duplicate-key rule as `.ci.yaml`."""
      with tempfile.TemporaryDirectory() as directory:
         path = Path(directory) / "catalog.yaml"
         path.write_text('schema: 1\ntoolchains:\n  java:\n    "21": {image: null}\n    "21": {image: null}\n', encoding="utf-8")
         with self.assertRaisesRegex(RuntimeCatalogError, "duplicate key '21'"):
            load_runtime_catalog(path)

   def test_a_missing_catalog_is_an_error(self) -> None:
      """No catalog is never treated as an empty, passing one."""
      with tempfile.TemporaryDirectory() as directory, self.assertRaisesRegex(RuntimeCatalogError, "not found"):
         load_runtime_catalog(Path(directory) / "catalog.yaml")


class ShippedCatalogTest(unittest.TestCase):
   """The catalog committed with the platform revision is valid and honest about what is qualified."""

   def test_every_example_toolchain_is_listed_and_every_entry_is_null_or_a_digest(self) -> None:
      """Loading validates every entry; the examples' versions must be listed so root only fills in values."""
      catalog = load_runtime_catalog(DEFAULT_CATALOG_PATH)
      self.assertIn("21", catalog.images["java"])
      self.assertIn("22.20.0", catalog.images["node"])
      for versions in catalog.images.values():
         for image in versions.values():
            self.assertTrue(image is None or is_immutable_image_reference(image))

   def test_prepare_fails_clearly_and_publishes_nothing_for_an_unqualified_runtime(self) -> None:
      """An unqualified toolchain stops `prepare` with exit 1 before any matrix output is written."""
      with tempfile.TemporaryDirectory() as directory:
         checkout = Path(directory)
         outputs_file = checkout / "outputs"
         (checkout / ".ci.yaml").write_text((EXAMPLES_DIRECTORY / "maven.ci.yaml").read_text(encoding="utf-8"), encoding="utf-8")
         environment = {
            "CI_EVENT_NAME": "pull_request",
            "CI_EVENT_REF": "refs/pull/7/merge",
            "CI_SOURCE_REVISION": SOURCE_REVISION,
            "CI_PLATFORM_REVISION": PLATFORM_REVISION,
            "GITHUB_OUTPUT": str(outputs_file),
         }
         unqualified = parse_runtime_catalog(catalog_with(None))
         with self.assertLogs("platform_ci.__main__", level="ERROR") as logged:
            self.assertEqual(main(["prepare", "--source", str(checkout)], environment, unqualified), 1)
         self.assertIn("java 21 runtime is unqualified", logged.output[0])
         self.assertFalse(outputs_file.exists())


class RepositoryIndependenceTest(unittest.TestCase):
   """Toolchain data decides the image; the repository never does."""

   def test_identical_configuration_resolves_identical_images_for_unrelated_repositories(self) -> None:
      """Two checkouts with unrelated names and the same `.ci.yaml` get the same image from the same catalog."""
      configuration_text = (EXAMPLES_DIRECTORY / "node.ci.yaml").read_text(encoding="utf-8")
      images: list[str] = []
      for repository_name in ("alpha-service", "zeta-unrelated-tool"):
         with tempfile.TemporaryDirectory() as parent:
            checkout = Path(parent) / repository_name
            checkout.mkdir()
            (checkout / ".ci.yaml").write_text(configuration_text, encoding="utf-8")
            images.append(plan_for(load_project_configuration(checkout), event())["targets"][0]["image"])
      self.assertEqual(images, [NODE_IMAGE, NODE_IMAGE])

   def test_the_matrix_image_is_the_catalog_entry_for_each_example(self) -> None:
      """The image that reaches `container.image` is exactly what the catalog records for the declared version."""
      for example, expected in (("maven.ci.yaml", JAVA_IMAGE), ("node.ci.yaml", NODE_IMAGE)):
         with self.subTest(example):
            self.assertEqual(plan_for(example_configuration(example), event())["targets"][0]["image"], expected)


class WorkflowRuntimeTest(unittest.TestCase):
   """The workflow uses qualified runtimes and installs nothing while it runs."""

   @override
   def setUp(self) -> None:
      """Parse the one workflow."""
      self.workflow = cast("dict[str, dict[str, object]]", yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8")))

   def _job(self, name: str) -> dict[str, object]:
      """Return one job's mapping."""
      return cast("dict[str, object]", self.workflow["jobs"][name])

   def _steps(self, name: str) -> list[dict[str, object]]:
      """Return one job's steps."""
      return cast("list[dict[str, object]]", self._job(name)["steps"])

   def test_prepare_and_report_use_the_same_published_control_image(self) -> None:
      """Both jobs name the one published control digest, never a tag."""
      images = {cast("dict[str, str]", self._job(name)["container"])["image"] for name in ("prepare", "report")}
      self.assertEqual(images, {CONTROL_IMAGE})
      self.assertTrue(is_immutable_image_reference(CONTROL_IMAGE))

   def test_every_job_is_linux_small_and_its_container_is_only_an_image(self) -> None:
      """Kubernetes job-image-pull identity pulls the image; credentials, options and volumes are unsupported and absent."""
      for name in ("prepare", "verify", "report"):
         self.assertEqual(self._job(name)["runs-on"], "linux-small", name)
         self.assertEqual(set(cast("dict[str, str]", self._job(name)["container"])), {"image"}, name)
      # the pool job is the one place container options and mounts may appear (see tests/test_pool_and_source.py)
      for name in (job for job in self.workflow["jobs"] if job not in ("verify-pool", "verify-pool-docker")):
         self.assertTrue({"options", "volumes"}.isdisjoint(cast("dict[str, object]", self._job(name).get("container", {}))), name)

   def test_verify_uses_the_image_that_prepare_resolved_from_the_catalog(self) -> None:
      """The tool image is never written in the workflow."""
      self.assertEqual(cast("dict[str, str]", self._job("verify")["container"])["image"], "${{ matrix.image }}")

   def test_no_step_installs_or_downloads_a_tool(self) -> None:
      """Tools come from the image: no run step fetches, unpacks or installs one."""
      forbidden = re.compile(r"\b(curl|wget|apt|apt-get|apk|yum|dnf|pip|pip3|uv|conda|brew|tar|unzip|sudo)\b|npm\s+(i|install)\b|nodejs\.org|actions-node")
      for name in ("prepare", "verify", "report"):
         for step in self._steps(name):
            run = cast("str", step.get("run", ""))
            self.assertIsNone(forbidden.search(run), f"{name}: {step.get('name')}")
            self.assertNotIn("install", cast("str", step.get("name", "")).lower())

   def test_checkouts_pass_only_inputs_the_platform_relies_on(self) -> None:
      """Gitea fails on an unknown `builtin:checkout` input, so only repository, ref and path may be passed; no sparse inputs."""
      for name in ("prepare", "verify", "report"):
         for step in self._steps(name):
            if step.get("uses") == "builtin:checkout":
               self.assertLessEqual(set(cast("dict[str, object]", step.get("with", {}))), {"repository", "ref", "path"}, f"{name}: {step.get('name')}")
      self.assertNotIn("sparse-checkout", WORKFLOW_PATH.read_text(encoding="utf-8"))

   def test_source_step_fails_fast_and_trusts_only_the_exact_workspace_per_command(self) -> None:
      """A runner-owned workspace is trusted for the one command only, and a failed rev-parse cannot be hidden by echo."""
      (step,) = [step for step in self._steps("prepare") if step.get("id") == "source"]
      lines = [line.strip() for line in cast("str", step["run"]).splitlines() if line.strip()]
      self.assertEqual(
         lines,
         [
            "set -eu",
            'revision="$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD)"',
            'echo "revision=$revision" >> "$GITHUB_OUTPUT"',
            'tree="$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse \'HEAD^{tree}\')"',
            'echo "tree=$tree" >> "$GITHUB_OUTPUT"',
         ],
      )
      self.assertNotIn("safe.directory=*", WORKFLOW_PATH.read_text(encoding="utf-8"))
      self.assertNotIn("--global", WORKFLOW_PATH.read_text(encoding="utf-8"))

   def test_every_job_runs_python_from_the_image_with_the_checkout_isolated(self) -> None:
      """Every command uses the image's interpreter in isolated mode, so a source file cannot shadow the platform."""
      for name in ("prepare", "verify", "report"):
         for step in self._steps(name):
            run = cast("str", step.get("run", ""))
            if "platform_ci" in run:
               self.assertIn("python3 -P -m platform_ci", run)


class RecipeTest(unittest.TestCase):
   """The recipes use only digest-pinned images, install only at build time, and keep the control image minimal."""

   @override
   def setUp(self) -> None:
      """Read both recipes."""
      self.control = (ROOT / "runtime" / "control.Dockerfile").read_text(encoding="utf-8")
      self.toolchain = (ROOT / "runtime" / "toolchain.Dockerfile").read_text(encoding="utf-8")

   def _instructions(self, text: str) -> str:
      """Return the recipe without comment lines, with line continuations joined."""
      return re.sub(r"\\\n\s*", " ", "\n".join(line for line in text.splitlines() if not line.lstrip().startswith("#")))

   def test_every_base_is_an_argument_or_an_earlier_stage_and_every_default_is_a_digest(self) -> None:
      """No literal or tag-only image: a FROM names an argument or stage, and an image default carries a sha256 digest."""
      for text in (self.control, self.toolchain):
         stages: set[str] = set()
         for source, alias in re.findall(r"(?m)^FROM\s+(\S+)(?:\s+AS\s+(\S+))?", text):
            self.assertTrue(source.startswith("${") or source in stages, source)
            stages.add(alias)
         for name, default in re.findall(r"(?m)^ARG\s+(\w*IMAGE\w*)=(\S*)$", text):
            self.assertRegex(default, r"@sha256:[0-9a-f]{64}$", name)

   def test_the_control_base_is_the_pinned_ubuntu_release(self) -> None:
      """The control image starts from Ubuntu 26.04 by digest and checks the pin again at build time."""
      self.assertRegex(self.control, r"(?m)^ARG UBUNTU_BASE_IMAGE=ubuntu:26\.04@sha256:[0-9a-f]{64}$")
      self.assertIn("@sha256:[0-9a-f]{64}$", self.control)

   def test_control_installs_only_pinned_packages_without_recommendations(self) -> None:
      """The package set is exactly Python, PyYAML, Git and CA certificates; versions are exact; apt state is removed."""
      instructions = self._instructions(self.control)
      (install,) = re.findall(r"apt-get install ([^;]*);", instructions)
      self.assertIn("--no-install-recommends", install)
      packages = [word for word in install.split() if not word.startswith("--")]
      self.assertEqual([package.split("=")[0].strip('"') for package in packages], ["python3", "python3-yaml", "git", "ca-certificates"])
      for package in packages[:3]:
         self.assertIn("=${", package)
      self.assertIn("rm -rf /var/lib/apt/lists/*", instructions)
      self.assertNotRegex(instructions, r"\b(pip3?|curl|wget|build-essential|gcc|g\+\+|make|imagemagick)\b|-dev\b")

   def test_control_build_checks_the_python_floor_and_the_actual_pyyaml_version(self) -> None:
      """A base that yields Python before 3.13 or a different PyYAML fails the build."""
      instructions = self._instructions(self.control)
      self.assertIn("sys.version_info >= (3, 13)", instructions)
      self.assertIn('yaml.__version__ == os.environ["PYYAML_EXPECTED"]', instructions)
      self.assertIn('PYYAML_EXPECTED="${PYYAML_VERSION}"', instructions)

   def test_the_recipe_pyyaml_version_matches_the_declared_requirement_and_package(self) -> None:
      """One PyYAML is promised: requirements.txt, the recipe's expected version and the Ubuntu package version agree."""
      requirement = re.search(r"(?m)^PyYAML==(\S+)$", (ROOT / "requirements.txt").read_text(encoding="utf-8"))
      default = re.search(r"(?m)^ARG PYYAML_VERSION=(\S+)$", self.control)
      package = re.search(r"(?m)^ARG PYTHON3_YAML_VERSION=(\S+)$", self.control)
      assert requirement is not None
      assert default is not None
      assert package is not None
      self.assertEqual(requirement.group(1), default.group(1))
      self.assertTrue(package.group(1).startswith(default.group(1) + "-"), package.group(1))

   def test_toolchain_targets_inherit_the_control_image_and_install_nothing(self) -> None:
      """Both targets build on the shared base, which is the control image; neither installs packages or downloads."""
      instructions = self._instructions(self.toolchain)
      self.assertRegex(instructions, r"(?m)^FROM \$\{CONTROL_IMAGE\} AS base$")
      for target in ("maven", "node"):
         self.assertRegex(instructions, rf"(?m)^FROM base AS {target}$")
      self.assertIn("CONTROL_IMAGE must be pinned by @sha256 digest", instructions)
      self.assertNotRegex(instructions, r"\b(apt|apt-get|apk|pip3?|curl|wget|dnf|yum)\b")

   def test_toolchain_copies_only_tool_distributions_never_a_foreign_system_tree(self) -> None:
      """Every copy from a source image is a named tool path; no root, /usr, /usr/local, /lib or /etc is taken wholesale."""
      copies = re.findall(r"(?m)^COPY --from=(\S+) (\S+) (\S+)$", self._instructions(self.toolchain))
      self.assertEqual(
         sorted(copies),
         sorted(
            [
               ("actions-node-source", "/usr/local/bin/node", "/opt/actions-node/bin/node"),
               ("maven-source", "/opt/java/openjdk/", "/opt/java/openjdk/"),
               ("maven-source", "/usr/share/maven/", "/out/maven/"),
               ("node-source", "/usr/local/bin/node", "/usr/local/bin/node"),
               ("patched-tools", "/out/maven/", "/usr/share/maven/"),
               ("patched-tools", "/out/npm/", "/usr/local/lib/node_modules/npm/"),
            ]
         ),
      )

   def test_toolchain_source_images_are_pinned_and_checked_in_the_stage_that_uses_them(self) -> None:
      """Each target rejects a tag-only source image; the action runtime version is checked after copying."""
      for argument in ("MAVEN_IMAGE", "NODE_IMAGE", "ACTIONS_NODE_IMAGE"):
         self.assertIn(f"{argument} must be pinned by @sha256 digest", self.toolchain)
      self.assertIn('test "$(/opt/actions-node/bin/node --version)" = "v${ACTIONS_NODE_VERSION}"', self.toolchain)

   def test_the_action_runtime_is_provided_once_and_is_last_on_path(self) -> None:
      """The runtime is added in the shared base only, appended to PATH, so a toolchain's own node wins."""
      instructions = self._instructions(self.toolchain)
      self.assertEqual(instructions.count("COPY --from=actions-node-source"), 1)
      self.assertEqual(len(re.findall(r"(?m)^ENV PATH=\"\$\{PATH\}:/opt/actions-node/bin\"$", instructions)), 1)
      for target in ("maven", "node"):
         self.assertNotIn("actions-node", instructions.split(f"FROM base AS {target}")[1].split("FROM ")[0])

   def test_each_target_smoke_checks_its_tools_at_build_time(self) -> None:
      """A broken copy fails the image build, not a pipeline."""
      instructions = self._instructions(self.toolchain)
      maven = instructions.split("FROM base AS maven")[1].split("FROM base AS node")[0]
      node = instructions.split("FROM base AS node")[1]
      for command in ("java -version", "mvn --version", "keytool -list -cacerts"):
         self.assertIn(command, maven)
      for command in ("node --version", "npm --version"):
         self.assertIn(command, node)