platform-ci/tests/test_image_packaging.py

"""Image packaging: one native verification, then images built from exactly its outputs and retained in the SAME verified bundle and manifest.

The last class runs the real publisher and reuse path over such a bundle, so acceptance is proven against the existing OCI publisher primitive and
not a parallel artifact protocol.
"""

import hashlib
import io
import json
import shutil
import tarfile
import tempfile
import unittest
from collections.abc import Mapping, Sequence
from datetime import date
from pathlib import Path
from typing import Any, cast, override
from unittest.mock import patch

import yaml

from platform_ci.__main__ import main
from platform_ci.build_tools import DOCKER_PROFILE_ENVIRONMENT, IMAGE_OUTPUT_DIRECTORY
from platform_ci.caching import CacheError, build_cache_metadata, trivy_cache_directory, trivy_cache_key
from platform_ci.configuration import ConfigurationError, parse_project_configuration, parse_project_configuration_text
from platform_ci.delivery.errors import ReuseError
from platform_ci.execution import run_target_commands
from platform_ci.image_packaging import DIAGNOSTIC_DIRECTORY_NAME, ImagePackagingError, ImageRun, package_images, verify_image_binding
from platform_ci.packaging import MANIFEST_FILE_NAME, RunIdentity, collect_outputs
from platform_ci.plan import Target, plan_targets
from platform_ci.runtimes import RuntimeCatalogError, load_runtime_catalog, parse_runtime_catalog
from tests.support import JAVA_IMAGE, NODE_IMAGE, PLATFORM_REVISION, SOURCE_REVISION, SOURCE_TREE, component_mapping, event, example_mapping, plan_for
from tests.synthetic_services import oci_archive
from tests.test_delivery_publisher import PublisherTestCase
from tests.test_pool_jobs import DOCKER_VARIABLES, WORKFLOW_TEXT

JAR_CONTENT = b"the one jar the single verification produced"
IMAGE_RUNTIME = f"registry.example/toolchain-image@sha256:{'3' * 64}"
IMAGE_CATALOG = parse_runtime_catalog(
   {
      "schema": 1,
      "runners": {"linux-amd64": {"image": "test-image-pool"}},
      "toolchains": {"java": {"21": {"image": JAVA_IMAGE, "image_image": IMAGE_RUNTIME}}, "node": {"22.20.0": {"image": NODE_IMAGE, "image_image": None}}},
   }
)
RUNTIME_ONLY_DOCKERFILE = 'FROM scratch\nCOPY target/app.jar /app/app.jar\nUSER 65532:65532\nCMD ["/app/app.jar"]\n'
RECOMPILING_DOCKERFILE = "FROM maven:3 AS build\nCOPY . .\nRUN mvn -B package\nFROM scratch\nCOPY --from=build target/app.jar /app/app.jar\n"
UNRELATED_DOCKERFILE = "FROM scratch\nCOPY LICENSE /LICENSE\n"


def _configuration_with_images(images: dict[str, object]) -> dict[str, object]:
   """Return the Maven example with the given `images` declared on its only component."""
   mapping = example_mapping("maven.ci.yaml")
   component_mapping(mapping, "library")["images"] = images
   return mapping


def _minimal_oci() -> bytes:
   """Return a minimal OCI layout archive that is not a real image (enough for the layout check)."""
   buffer = io.BytesIO()
   with tarfile.open(fileobj=buffer, mode="w") as archive:
      for name, content in (("oci-layout", b'{"imageLayoutVersion":"1.0.0"}'), ("index.json", b"{}"), ("blobs/sha256/" + "0" * 64, b"layer")):
         member = tarfile.TarInfo(name)
         member.size = len(content)
         archive.addfile(member, io.BytesIO(content))
   return buffer.getvalue()


class _Tools:
   """A stand-in for the processes of one job: records every command and the bytes of the tested jar when BuildKit ran."""

   def __init__(self, *, scan_exit: int = 0, boundary_exit: int = 0, archive: bytes | None = None, jar: str = "target/app.jar") -> None:
      self.commands: list[list[str]] = []
      self.native_builds = 0
      self.context_jar: bytes | None = None
      self.scan_exit, self.boundary_exit = scan_exit, boundary_exit
      self.archive = archive if archive is not None else _minimal_oci()
      self.jar = jar

   def native(self, command: Sequence[str], working_directory: Path, _environment: Mapping[str, str]) -> int:
      """The verification's own runner: `mvn` compiles and tests once and leaves the jar and the report."""
      self.commands.append(list(command))
      if command[0] == "mvn":
         self.native_builds += 1
         (working_directory / "target" / "surefire-reports").mkdir(parents=True, exist_ok=True)
         (working_directory / "target" / "surefire-reports" / "TEST-a.xml").write_text('<testsuite tests="1"/>', encoding="utf-8")
         (working_directory / self.jar).write_bytes(JAR_CONTENT)
      return 0

   def packaging(self, command: Sequence[str], working_directory: Path, _environment: Mapping[str, str]) -> int:
      """The image operation's runner: boundary check, BuildKit, Trivy and the evidence check; a native build would be counted."""
      self.commands.append(list(command))
      program = command[0]
      if program == "mvn":
         self.native_builds += 1
      elif program == "bash":
         return self.boundary_exit
      elif program == "buildctl":
         jar = working_directory / self.jar
         self.context_jar = jar.read_bytes() if jar.is_file() else None
         destination = next(token for token in command if token.startswith("type=oci,dest=")).removeprefix("type=oci,dest=")
         Path(destination).write_bytes(self.archive)
      elif program == "trivy" and "json" in command:
         Path(command[command.index("--output") + 1]).write_text('{"ArtifactType":"container_image","Results":[]}', encoding="utf-8")
         return self.scan_exit
      elif program == "trivy":
         Path(command[command.index("--output") + 1]).write_text('{"bomFormat":"CycloneDX","components":[]}', encoding="utf-8")
      return 0

   def programs(self) -> list[str]:
      """Return the program of every command started so far."""
      return [command[0] for command in self.commands]


class ImageConfigurationTest(unittest.TestCase):
   """The schema accepts only images that can be built from the component's own tested output, and declares their outputs as ordinary ones."""

   def test_images_imply_the_image_profile_and_declare_their_outputs_as_ordinary_artifacts(self) -> None:
      """The archive is a `container` artifact and the evidence a required report, so the one bundle retains both."""
      configuration = parse_project_configuration(_configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}}))
      target = plan_for(configuration, event(), IMAGE_CATALOG)["targets"][0]
      self.assertEqual((target["requires"], target["execution"], target["image"], target["runner"]), (["image"], "pool", IMAGE_RUNTIME, "test-image-pool"))
      self.assertEqual(target["commands"], [["mvn", "--batch-mode", "verify"]], "image packaging adds no native command")
      component = configuration.components[0]
      self.assertEqual(component.artifacts["image-app"], (f"{IMAGE_OUTPUT_DIRECTORY}/app/image.tar",))
      self.assertEqual(component.artifact_formats["image-app"], "container")
      evidence = component.reports["image-app-evidence"]
      self.assertEqual((evidence.format, evidence.required), ("file", True))
      self.assertIn(f"{IMAGE_OUTPUT_DIRECTORY}/app/trivy.json", evidence.paths)
      self.assertIn(f"{IMAGE_OUTPUT_DIRECTORY}/app/sbom.cdx.json", evidence.paths)

   def test_a_component_without_images_is_unchanged(self) -> None:
      """No images means no profile, the standard pool and nothing generated."""
      configuration = parse_project_configuration(example_mapping("maven.ci.yaml"))
      target = plan_for(configuration, event(), IMAGE_CATALOG)["targets"][0]
      self.assertEqual((target["requires"], target["execution"], target["images"]), ([], "container", {}))
      self.assertNotIn("image-app", configuration.components[0].artifacts)

   def test_rejected_declarations(self) -> None:
      """The artifact must belong to the component; a dockerfile and a recipe are exclusive; no other profile mixes in; no name is replaced."""
      cases: list[dict[str, object]] = [
         {"app": {"artifact": "missing", "dockerfile": "Dockerfile"}},
         {"app": {"artifact": "image-app", "dockerfile": "Dockerfile"}},
         {"app": {"artifact": "packages"}},
         {"app": {"artifact": "packages", "dockerfile": "Dockerfile", "recipe": "node-webapp-runner", "node_version": "22", "health_port": "3000"}},
         {"app": {"artifact": "packages", "dockerfile": "../Dockerfile"}},
         {"app": {"artifact": "packages", "recipe": "unknown", "node_version": "22", "health_port": "3000"}},
         {"app": {"artifact": "packages", "dockerfile": "Dockerfile", "node_version": "22"}},
      ]
      for images in cases:
         with self.subTest(images=images), self.assertRaises(ConfigurationError):
            parse_project_configuration(_configuration_with_images(images))
      declared = _configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}})
      component_mapping(declared, "library")["requires"] = ["image"]
      del component_mapping(declared, "library")["images"]
      with self.assertRaisesRegex(ConfigurationError, "implied by `images`"):
         parse_project_configuration(declared)
      clash = _configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}})
      component_mapping(clash, "library")["artifacts"] = {"packages": {"paths": ["target/*.jar"]}, "image-app": {"paths": ["x"]}}
      with self.assertRaisesRegex(ConfigurationError, "would replace a declared one"):
         parse_project_configuration(clash)

   def test_unqualified_image_runtime_blocks_planning_by_name(self) -> None:
      """Node has no image runtime holding npm, BuildKit and Trivy, and a missing pool label blocks too: nothing is substituted."""
      node = example_mapping("node.ci.yaml")
      for component in cast("dict[str, dict[str, object]]", node["components"]).values():
         component["artifacts"] = {"dist": {"paths": ["dist/*.js"]}}
         component["images"] = {"app": {"artifact": "dist", "dockerfile": "Dockerfile"}}
      with self.assertRaisesRegex(RuntimeCatalogError, "image runtime is unqualified"):
         plan_for(parse_project_configuration(node), event(), IMAGE_CATALOG)
      without_pool = parse_runtime_catalog({"schema": 1, "toolchains": {"java": {"21": {"image": JAVA_IMAGE, "image_image": IMAGE_RUNTIME}}}})
      with self.assertRaisesRegex(RuntimeCatalogError, "image runtime profile has no qualified runner pool"):
         plan_for(parse_project_configuration(_configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}})), event(), without_pool)

   def test_image_composes_with_browser_and_docker_without_dropping_either(self) -> None:
      """Every declared capability is kept in `requires`; docker selects the docker job, a browser-only requirement the browser job, and docker adds no image."""
      catalog = parse_runtime_catalog(
         {
            "schema": 1,
            "runners": {"linux-amd64": {"image": "pool-i", "docker+image": "pool-di", "browser+image": "pool-bi", "browser+docker+image": "pool-bdi"}},
            "toolchains": {"java": {"21": {"image": JAVA_IMAGE, "image_image": IMAGE_RUNTIME, "image_browser_image": f"registry.example/browser-image@sha256:{'5' * 64}"}}},
         }
      )
      browser_image = f"registry.example/browser-image@sha256:{'5' * 64}"
      expected = {
         (): (["image"], "pool", "pool-i", IMAGE_RUNTIME),
         ("docker",): (["docker", "image"], "pool-docker", "pool-di", IMAGE_RUNTIME),
         ("browser",): (["browser", "image"], "pool", "pool-bi", browser_image),
         ("browser", "docker"): (["browser", "docker", "image"], "pool-docker", "pool-bdi", browser_image),
      }
      for requires, (profiles, execution, runner, image) in expected.items():
         with self.subTest(requires):
            mapping = _configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}})
            component_mapping(mapping, "library")["requires"] = list(requires)
            (target,) = plan_for(parse_project_configuration(mapping), event(), catalog)["targets"]
            self.assertEqual((target["requires"], target["execution"], target["runner"], target["image"]), (profiles, execution, runner, image))
            self.assertEqual("DOCKER_HOST" in target["environment"], "docker" in requires)

   def test_unqualified_combinations_are_null_in_the_shipped_catalog_and_block_by_name(self) -> None:
      """No published image has a browser and BuildKit and Trivy, so those combinations fail naming what is missing; nothing is dropped or substituted."""
      shipped = load_runtime_catalog()
      self.assertIsNone(shipped.image_browser_images["java"]["21"])
      self.assertIsNone(shipped.image_browser_images["node"]["22.20.0"])
      self.assertEqual({profile: shipped.runners["linux-amd64"][profile] for profile in ("docker+image", "browser+image", "browser+docker+image")}, {"docker+image": "container-build", "browser+image": None, "browser+docker+image": None})
      for requires in (("browser",), ("browser", "docker")):
         with self.subTest(requires):
            mapping = _configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}})
            component_mapping(mapping, "library")["requires"] = list(requires)
            with self.assertRaisesRegex(RuntimeCatalogError, "image\\+browser runtime is unqualified"):
               plan_for(parse_project_configuration(mapping), event(), shipped)
      mapping = _configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}})
      component_mapping(mapping, "library")["requires"] = ["docker"]
      (target,) = plan_for(parse_project_configuration(mapping), event(), shipped)["targets"]
      self.assertEqual((target["runner"], target["image"], target["execution"]), ("container-build", shipped.image_images["java"]["21"], "pool-docker"))

   def test_shipped_catalog_uses_the_container_build_pool_and_the_verified_container_image(self) -> None:
      """Java and Node use their published packaging runtimes without substituting a different native toolchain."""
      catalog = load_runtime_catalog()
      self.assertIn("platform-ci-container@sha256:", str(catalog.image_images["java"]["21"]))
      for version, name in (("20.20.2", "node20"), ("22.20.0", "node22"), ("24.22.2", "node24")):
         self.assertIn(f"platform-ci-{name}-image@sha256:", str(catalog.image_images["node"][version]))
      self.assertEqual(catalog.runners["linux-amd64"]["image"], "container-build")
      runtime = catalog.resolve_runtime("java", "21", "linux-amd64", ("image",))
      self.assertEqual((runtime.runner, runtime.image), ("container-build", catalog.image_images["java"]["21"]))


class ImagePackagingTest(unittest.TestCase):
   """One native verification, then image packaging that uses its output, never builds the source again and is retained in the one bundle."""

   @override
   def setUp(self) -> None:
      """Create a checkout and plan the target; the Dockerfile is set by each test."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.root = Path(directory.name)
      self.source, self.home, self.bundle = self.root / "checkout", self.root / "home", self.root / "bundle"
      self.source.mkdir()
      self.home.mkdir()
      configuration = parse_project_configuration(_configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}}))
      self.target: Target = plan_for(configuration, event(), IMAGE_CATALOG)["targets"][0]
      self.tools = _Tools()

   def verify(self, dockerfile: str) -> None:
      """Run the target's native commands once, as the pool job's verify step does."""
      (self.source / "Dockerfile").write_text(dockerfile, encoding="utf-8")
      self.assertTrue(run_target_commands(self.target, self.source, self.tools.native, {"HOME": str(self.home)}))

   def images(self) -> list[str]:
      """Package the images and return the program names the image operation started."""
      before = len(self.tools.commands)
      package_images(self.target, self.source, ImageRun(SOURCE_REVISION, PLATFORM_REVISION, self.tools.packaging, {}))
      return [command[0] for command in self.tools.commands[before:]]

   def collect(self) -> dict[str, object]:
      """Collect the one bundle, as the ordinary package step does, and check the image binding."""
      manifest = collect_outputs(self.target, self.source, self.bundle, RunIdentity(SOURCE_REVISION, PLATFORM_REVISION, "7", "1", SOURCE_TREE), build_succeeded=True)
      verify_image_binding(self.target, self.source, manifest)
      return manifest

   def test_one_native_verification_and_the_image_is_retained_in_the_same_manifest(self) -> None:
      """`mvn` ran once; BuildKit saw the tested jar; boundary, Trivy, evidence and SBOM ran; the one manifest lists jar, archive and evidence."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      started = self.images()
      self.assertEqual(self.tools.native_builds, 1)
      self.assertEqual(started, ["bash", "buildctl", "trivy", "node", "trivy"])
      boundary = self.tools.commands[len(self.tools.commands) - 5]
      self.assertTrue(boundary[1].endswith("image-boundary.sh"))
      manifest = self.collect()
      files = {str(entry["path"]): str(entry["sha256"]) for entry in cast("list[dict[str, object]]", manifest["files"])}
      self.assertEqual(hashlib.sha256(self.tools.context_jar or b"").hexdigest(), files["artifact/packages/target/app.jar"])
      self.assertIn(f"artifact/image-app/{IMAGE_OUTPUT_DIRECTORY}/app/image.tar", files)
      for name in ("trivy.json", "sbom.cdx.json", "SOURCE_COMMIT", "tested-output.json", "SHA256SUMS"):
         self.assertIn(f"report/image-app-evidence/{IMAGE_OUTPUT_DIRECTORY}/app/{name}", files)
      self.assertEqual((self.source / IMAGE_OUTPUT_DIRECTORY / "app" / "SOURCE_COMMIT").read_text(encoding="utf-8").strip(), SOURCE_REVISION)
      self.assertEqual(json.loads((self.bundle / MANIFEST_FILE_NAME).read_text(encoding="utf-8")), manifest)

   def test_trivy_is_the_captured_scan_and_the_helpers_are_the_receipted_ones(self) -> None:
      """The scan fails on HIGH and CRITICAL; the evidence program and the boundary script are the vendored, receipted files."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      self.images()
      scan = next(command for command in self.tools.commands if command[0] == "trivy")
      self.assertEqual((scan[scan.index("--severity") + 1], scan[scan.index("--exit-code") + 1], scan[scan.index("--scanners") + 1]), ("HIGH,CRITICAL", "1", "vuln"))
      self.assertTrue(next(command for command in self.tools.commands if command[0] == "node")[1].endswith("check-image-evidence.mjs"))

   def test_the_image_is_scanned_once_and_the_sbom_is_converted_from_that_report(self) -> None:
      """One `trivy image` scan lists every package; the SBOM is `trivy convert` of that same report; the freshness update is never skipped."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      self.images()
      trivy = [command for command in self.tools.commands if command[0] == "trivy"]
      self.assertEqual([command[1] for command in trivy], ["image", "convert"])
      scan, convert = trivy
      report = scan[scan.index("--output") + 1]
      self.assertIn("--list-all-pkgs", scan)
      self.assertEqual((scan[scan.index("--severity") + 1], scan[scan.index("--exit-code") + 1], scan[scan.index("--format") + 1]), ("HIGH,CRITICAL", "1", "json"))
      self.assertEqual(convert[1:5], ["convert", "--format", "cyclonedx", "--output"])
      self.assertEqual(convert[-1], report, "the SBOM is converted from the scan's own report")
      self.assertEqual(sum(1 for command in self.tools.commands if "--input" in command), 1, "the image is read once")
      for command in trivy:
         self.assertNotIn("--skip-db-update", command)

   def test_the_scan_uses_the_cache_directory_the_workflow_restores_and_saves(self) -> None:
      """`--cache-dir` is the proven `<home>/.cache/trivy`; a home inside the checkout cannot be proven safe, so the scanner's default is left alone."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      package_images(self.target, self.source, ImageRun(SOURCE_REVISION, PLATFORM_REVISION, self.tools.packaging, {"HOME": str(self.home)}))
      scan = next(command for command in self.tools.commands if command[:2] == ["trivy", "image"])
      self.assertEqual(scan[scan.index("--cache-dir") + 1], (self.home / ".cache" / "trivy").as_posix())
      shutil.rmtree(self.source / IMAGE_OUTPUT_DIRECTORY)
      self.tools.commands.clear()
      package_images(self.target, self.source, ImageRun(SOURCE_REVISION, PLATFORM_REVISION, self.tools.packaging, {"HOME": str(self.source)}))
      self.assertNotIn("--cache-dir", next(command for command in self.tools.commands if command[:2] == ["trivy", "image"]))

   def test_a_blocking_scan_still_fails_closed_after_the_conversion_change(self) -> None:
      """HIGH/CRITICAL findings stop packaging before the evidence check and before any SBOM conversion."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      self.tools = _Tools(scan_exit=1)
      with self.assertRaisesRegex(ImagePackagingError, "HIGH/CRITICAL"):
         self.images()
      self.assertEqual([command[0] for command in self.tools.commands if command[0] in ("node", "trivy")], ["trivy"])
      self.assertEqual({path.name for path in (self.source / DIAGNOSTIC_DIRECTORY_NAME / "app").iterdir()}, {"trivy.json", "SOURCE_COMMIT"})
      self.assertFalse((self.source / IMAGE_OUTPUT_DIRECTORY).exists())

   def test_a_failed_boundary_check_builds_nothing(self) -> None:
      """The captured execution-boundary check runs before any repository code; its failure stops before BuildKit."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      self.tools = _Tools(boundary_exit=1)
      with self.assertRaisesRegex(ImagePackagingError, "boundary check failed"):
         self.images()
      self.assertEqual(self.tools.programs(), ["bash"])
      self.assertFalse((self.source / IMAGE_OUTPUT_DIRECTORY).exists())

   def test_a_dockerfile_that_recompiles_is_refused_and_is_not_build_reuse(self) -> None:
      """A Dockerfile whose RUN builds the source is rejected before BuildKit starts; it cannot satisfy build reuse."""
      self.verify(RECOMPILING_DOCKERFILE)
      with self.assertRaisesRegex(ImagePackagingError, "builds the source again"):
         self.images()
      self.assertNotIn("buildctl", self.tools.programs())
      self.assertEqual(self.tools.native_builds, 1)
      self.assertFalse((self.source / IMAGE_OUTPUT_DIRECTORY).exists())

   def test_a_dockerfile_that_copies_no_tested_output_is_refused(self) -> None:
      """An image that would not contain the tested jar is not an image of this verification."""
      self.verify(UNRELATED_DOCKERFILE)
      with self.assertRaisesRegex(ImagePackagingError, "copies none of the tested artifact files"):
         self.images()
      self.assertNotIn("buildctl", self.tools.programs())

   def test_a_tested_output_changed_after_the_image_was_built_is_refused_at_packaging(self) -> None:
      """The bundle must list exactly the bytes the image was built from; a jar swapped before `package` fails the binding."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      self.images()
      (self.source / "target" / "app.jar").write_bytes(b"swapped after the image was built")
      with self.assertRaisesRegex(ImagePackagingError, "differ from the files the image was built from"):
         self.collect()

   def test_an_archive_swapped_before_packaging_is_refused(self) -> None:
      """The archive the bundle retains must be the one that was scanned."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      self.images()
      archive = self.source / IMAGE_OUTPUT_DIRECTORY / "app" / "image.tar"
      manifest = collect_outputs(self.target, self.source, self.bundle, RunIdentity(SOURCE_REVISION, PLATFORM_REVISION, "7", "1", SOURCE_TREE), build_succeeded=True)
      archive.write_bytes(b"a different archive")
      with self.assertRaisesRegex(ImagePackagingError, "does not retain the image archive"):
         verify_image_binding(self.target, self.source, manifest)

   def test_a_missing_image_fails_the_package_step_closed(self) -> None:
      """If image packaging did not run, the required image artifact is missing and the bundle cannot be collected."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      with self.assertRaisesRegex(Exception, "matched no files"):
         collect_outputs(self.target, self.source, self.bundle, RunIdentity(SOURCE_REVISION, PLATFORM_REVISION, "7", "1", SOURCE_TREE), build_succeeded=True)

   def test_a_blocking_scan_leaves_no_output_but_keeps_the_report(self) -> None:
      """A failed Trivy scan removes the image output so it cannot be retained, and keeps only the scanner's report and the commit."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      self.tools = _Tools(scan_exit=1)
      with self.assertRaisesRegex(ImagePackagingError, "Trivy found blocking"):
         self.images()
      self.assertFalse((self.source / IMAGE_OUTPUT_DIRECTORY).exists())
      self.assertTrue((self.source / DIAGNOSTIC_DIRECTORY_NAME / "app" / "trivy.json").is_file())
      self.assertFalse((self.source / DIAGNOSTIC_DIRECTORY_NAME / "app" / "image.tar").exists())

   def test_a_bad_source_revision_is_refused(self) -> None:
      """Only a full commit id can be recorded as the image's source."""
      self.verify(RUNTIME_ONLY_DOCKERFILE)
      with self.assertRaisesRegex(ImagePackagingError, "full commit"):
         package_images(self.target, self.source, ImageRun("main", PLATFORM_REVISION, self.tools.packaging, {}))

   def test_a_recipe_is_rendered_outside_the_context_and_recorded(self) -> None:
      """A central recipe is rendered from the shipped template, labelled and recorded, and copies only tested output."""
      node = example_mapping("node.ci.yaml")
      for component in cast("dict[str, dict[str, object]]", node["components"]).values():
         component["artifacts"] = {"webapp": {"paths": [".next/standalone/server.js", ".next/static/a.js", "public/a.txt"]}}
         component["images"] = {"web": {"artifact": "webapp", "recipe": "node-webapp-runner", "node_version": "22", "health_port": "3000"}}
      catalog = parse_runtime_catalog({"schema": 1, "runners": {"linux-amd64": {"image": "test-image-pool"}}, "toolchains": {"node": {"22.20.0": {"image": NODE_IMAGE, "image_image": IMAGE_RUNTIME}}}})
      self.target = plan_for(parse_project_configuration(node), event(), catalog)["targets"][0]
      for relative in (".next/standalone/server.js", ".next/static/a.js", "public/a.txt", "reports/junit.xml"):
         (self.source / relative).parent.mkdir(parents=True, exist_ok=True)
         (self.source / relative).write_text("verified", encoding="utf-8")
      self.images()
      build = next(command for command in self.tools.commands if command[0] == "buildctl")
      self.assertIn("label:org.kof22.ci.recipe=node-webapp-runner", build)
      self.assertFalse((self.source / "Dockerfile").exists(), "the rendered Dockerfile lives outside the context")
      record = cast("dict[str, str]", json.loads((self.source / IMAGE_OUTPUT_DIRECTORY / "web" / "recipe.json").read_text(encoding="utf-8")))
      self.assertEqual((record["recipe"], record["node_version"], record["engine"]), ("node-webapp-runner", "22", PLATFORM_REVISION))
      manifest = self.collect()
      self.assertIn(f"report/image-web-evidence/{IMAGE_OUTPUT_DIRECTORY}/web/recipe.json", {str(entry["path"]) for entry in cast("list[dict[str, object]]", manifest["files"])})
      template = (Path(__file__).parent.parent / "platform_ci" / "image_recipes" / "node-webapp-runner.Dockerfile").read_text(encoding="utf-8")
      self.assertFalse([line for line in template.splitlines() if line.startswith("RUN ")], "the recipe has no RUN step, so it cannot build the source")


IMAGE_PUBLICATION_CONFIGURATION = (
   "schema: 1\ncomponents:\n  library:\n    toolchains: {java: '21'}\n    build: {tool: maven, arguments: [verify]}\n"
   "    artifacts:\n      packages: {format: maven, paths: [pom.xml, 'target/widget-*.jar']}\n"
   "    images:\n      widget: {artifact: packages, dockerfile: Dockerfile}\n"
   "release:\n  channels:\n    snapshot:\n      trigger: {push_branches: [main]}\n      versioning: maven-snapshot\n"
   "  publications:\n    widget-packages: {component: library, artifact: packages, destination: company-packages, channels: [snapshot]}\n"
   "    widget-image: {component: library, artifact: image-widget, destination: company-images, channels: [snapshot], deploy: prod-gitops}\n"
)


class ImageCapabilityEnvironmentAndCacheTest(unittest.TestCase):
   """Image packaging never receives the docker contract, and the scanner's database cache follows the dependency cache's rules."""

   @staticmethod
   def docker_image_target() -> Target:
      """Plan a component that needs `docker` and packages an image."""
      mapping = _configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}})
      component_mapping(mapping, "library")["requires"] = ["docker"]
      return plan_for(parse_project_configuration(mapping), event(), load_runtime_catalog())["targets"][0]

   def test_the_native_build_gets_the_docker_contract_but_image_packaging_does_not(self) -> None:
      """The boundary check refuses a Docker variable, and BuildKit is not the test daemon: the contract goes to the verification only."""
      target = self.docker_image_target()
      self.assertEqual({name: target["environment"][name] for name in DOCKER_VARIABLES}, dict(DOCKER_PROFILE_ENVIRONMENT))
      seen: list[Mapping[str, str]] = []
      with tempfile.TemporaryDirectory() as directory:
         source = Path(directory) / "checkout"
         (source / "target").mkdir(parents=True)
         (source / "target" / "app.jar").write_bytes(JAR_CONTENT)
         (source / "Dockerfile").write_text(RUNTIME_ONLY_DOCKERFILE, encoding="utf-8")
         tools = _Tools()

         def recorder(command: Sequence[str], working_directory: Path, environment: Mapping[str, str]) -> int:
            seen.append(dict(environment))
            return tools.packaging(command, working_directory, environment)

         environment = {"CI_TARGET": json.dumps(target), "CI_SOURCE_REVISION": SOURCE_REVISION, "CI_PLATFORM_REVISION": PLATFORM_REVISION, "HOME": str(Path(directory) / "home"), "DOCKER_HOST": "unix:///var/run/docker.sock"}
         with patch("platform_ci.__main__.run_process", recorder):
            self.assertEqual(main(["package-images", "--source", str(source)], environment), 0)
      self.assertEqual(len(seen), 5)
      for received in seen:
         self.assertTrue(set(DOCKER_VARIABLES).isdisjoint(received))
         self.assertEqual(received["HOME"], environment["HOME"])

   def test_cache_metadata_adds_a_daily_scanner_key_only_for_image_targets(self) -> None:
      """Exact key (UTC day and runtime image), the scanner's own directory, no fallback; other targets get nothing new."""
      target = plan_for(parse_project_configuration(_configuration_with_images({"app": {"artifact": "packages", "dockerfile": "Dockerfile"}})), event(), IMAGE_CATALOG)["targets"][0]
      plain = plan_for(parse_project_configuration(example_mapping("maven.ci.yaml")), event(), IMAGE_CATALOG)["targets"][0]
      with tempfile.TemporaryDirectory() as directory:
         source, home = Path(directory) / "checkout", Path(directory) / "home"
         source.mkdir()
         home.mkdir()
         (source / "pom.xml").write_text("<project/>", encoding="utf-8")
         environment = {"HOME": str(home), "CI_EVENT_NAME": "push", "CI_EVENT_REF": "refs/heads/main", "CI_DEFAULT_BRANCH": "main"}
         monday = build_cache_metadata(target, source, environment, date(2026, 11, 2))
         tuesday = build_cache_metadata(target, source, environment, date(2026, 11, 3))
         self.assertEqual(monday.trivy_path, (home / ".cache" / "trivy").as_posix())
         self.assertNotEqual(monday.trivy_key, tuesday.trivy_key)
         self.assertIn("20261102", monday.trivy_key)
         self.assertEqual(monday.key, tuesday.key, "the dependency cache key does not change with the day")
         self.assertEqual(trivy_cache_key(target, date(2026, 11, 2)), monday.trivy_key)
         other_image = cast("Target", {**target, "image": f"registry.example/other@sha256:{'8' * 64}"})
         self.assertNotEqual(trivy_cache_key(other_image, date(2026, 11, 2)), monday.trivy_key)
         self.assertEqual((build_cache_metadata(plain, source, environment).trivy_key, build_cache_metadata(plain, source, environment).trivy_path), ("", ""))
         with self.assertRaises(CacheError):
            trivy_cache_directory(source, source)

   def test_the_scanner_cache_is_saved_only_after_a_successful_default_branch_push_and_never_falls_back(self) -> None:
      """Both pool jobs restore the exact key; saving is a final, push-only step that needs the whole job to have succeeded."""
      jobs = cast("dict[str, dict[str, Any]]", yaml.safe_load(WORKFLOW_TEXT)["jobs"])
      for name in ("verify-pool", "verify-pool-docker"):
         steps = {step.get("id", step.get("name")): step for step in jobs[name]["steps"]}
         restore, check = steps["trivy-restore"], steps["trivy-directory"]
         save = steps["Save vulnerability database cache after a successful job"]
         self.assertEqual(jobs[name]["steps"][-1], save, "saving is the last step, so success() covers the whole job")
         self.assertNotIn("restore-keys", restore["with"])
         self.assertTrue(restore["uses"].startswith("https://github.com/actions/cache/restore@"))
         self.assertTrue(save["uses"].startswith("https://github.com/actions/cache/save@"))
         for condition in (check["if"], save["if"]):
            for required in ("success()", "github.event_name == 'push'", "trivy_save == 'true'", "steps.images.outcome == 'success'", "trivy-restore.outputs.cache-hit != 'true'"):
               self.assertIn(required, condition)
         self.assertEqual((restore["with"]["key"], save["with"]["key"]), ("${{ steps.cache.outputs.trivy_key }}",) * 2)
         self.assertTrue(all(step.get("continue-on-error") for step in (restore, check, save)))
      self.assertNotIn("trivy", json.dumps(jobs["verify"]), "the standard job has no image packaging")


class ImagePublicationTest(PublisherTestCase):
   """The real publisher and reuse path accept the image from the one verified bundle: no second artifact protocol."""

   @override
   def setUp(self) -> None:
      """Use the image runtime catalog and the image-publishing configuration."""
      super().setUp()
      self.runtimes = IMAGE_CATALOG
      self.config = IMAGE_PUBLICATION_CONFIGURATION

   def build_with_image(self, sha: str, run_id: int, *, tamper_archive: bool = False) -> tuple[Path, str, _Tools]:
      """Run the real flow for a commit (one verification, image packaging, collect, bind) and register the retained bundle with the forge."""
      self.forge.add_run(run_id, sha, conclusion="success", status="completed")
      project = parse_project_configuration_text(self.repository.snapshots[sha][".ci.yaml"].decode())
      (target,) = plan_targets(project, self.runtimes)
      archive, digest = oci_archive()
      tools = _Tools(archive=archive, jar=f"target/widget-{self.version}.jar")
      source = self.root / f"build-{run_id}"
      source.mkdir()
      (source / "pom.xml").write_bytes(self.repository.snapshots[sha]["pom.xml"])
      (source / "Dockerfile").write_text(f"FROM scratch\nCOPY target/widget-{self.version}.jar /app.jar\nUSER 65532:65532\n", encoding="utf-8")
      self.assertTrue(run_target_commands(target, source, tools.native, {"HOME": str(self.root)}))
      package_images(target, source, ImageRun(sha, PLATFORM_REVISION, tools.packaging, {}))
      output = self.root / f"bundle-{run_id}"
      manifest = collect_outputs(target, source, output, RunIdentity(sha, PLATFORM_REVISION, str(run_id), "1", self.repository.trees[sha]), build_succeeded=True)
      verify_image_binding(target, source, manifest)
      if tamper_archive:
         retained = output / "artifact" / "image-widget" / IMAGE_OUTPUT_DIRECTORY / "widget" / "image.tar"
         retained.write_bytes(retained.read_bytes() + b"x")
      self.forge.add_artifact(run_id, f"verified-{target['name']}", output)
      return source, digest, tools

   def test_the_image_from_the_one_bundle_is_published_by_digest_and_reused_without_rebuilding(self) -> None:
      """A later commit with the same tree republishes the EARLIER run's tested image bytes; nothing builds, tests or starts a process."""
      built = self.commit()
      _source, digest, tools = self.build_with_image(built, 100)
      self.assertEqual(tools.native_builds, 1)
      self.repository.commit({}, "same tree, later commit")
      with self.no_process_may_start():
         result = self.execute()
      self.assertEqual(result["status"], "published")
      images = [item for item in cast("list[dict[str, Any]]", result["published"]) if item["format"] == "container"]
      self.assertEqual((images[0]["digest"], images[0]["reused"], images[0]["build_sha"] == built), (digest, True, True))
      self.assertIn(digest, self.world.image_registry("koftwentytwo/widget").manifests)
      (deployment,) = cast("list[dict[str, Any]]", result["deployments"])
      self.assertEqual((deployment["status"], deployment["pins"]), ("committed", (("web", digest),)))
      self.assertEqual(tools.native_builds, 1, "publication did not build or test again")

   def test_a_tampered_image_archive_in_the_bundle_is_not_published(self) -> None:
      """The publisher re-hashes the retained archive; a swapped one is refused and nothing is written to the registry."""
      built = self.commit()
      self.build_with_image(built, 100, tamper_archive=True)
      self.repository.commit({}, "same tree, later commit")
      with self.assertRaisesRegex(ReuseError, "differs from the manifest"):
         self.execute()
      self.assertFalse(self.world.image_registry("koftwentytwo/widget").manifests)


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