platform-ci/tests/test_caching.py

"""Dependency caching: exact keys from descriptors only, safe download directories only, and a workflow that cannot skip the build."""

import copy
import json
import tempfile
import unittest
from pathlib import Path
from typing import cast, override

import yaml

from platform_ci.__main__ import main
from platform_ci.caching import CacheError, build_cache_metadata, cache_directory, cache_unavailable_reason, dependency_descriptors, save_allowed
from platform_ci.plan import Target, to_compact_json
from platform_ci.workspace import PipelineError
from tests.support import event, example_configuration, plan_for

WORKFLOW_PATH = Path(__file__).resolve().parent.parent / ".gitea" / "scoped_workflows" / "pipeline.yaml"
CACHE_ACTION_REVISION = "5a3ec84eff668545956fd18022155c47e93e2684"
# the shape of a runner-provided native cache endpoint (private HTTP on a pod address); tests only, never read by the platform
NATIVE_CACHE_URL = "http://10.1.2.3:8088/"
REQUIRED_GATE_LINES = [
   "set -eu",
   """case "$ACTIONS_RUNTIME_URL" in https://git.kof22.com|https://git.kof22.com/*) ;; *) echo 'Runtime endpoint is not the private HTTPS forge'; exit 1 ;; esac""",
   """case "${ACTIONS_RESULTS_URL:-}" in ''|https://git.kof22.com|https://git.kof22.com/*) ;; *) echo 'Results endpoint is not the private HTTPS forge'; exit 1 ;; esac""",
]


class CacheTestCase(unittest.TestCase):
   """A checkout, a separate home directory, and the Maven and Node example targets."""

   @override
   def setUp(self) -> None:
      """Create a Maven checkout, a Node checkout and a home directory outside both."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.root = Path(directory.name).resolve()
      self.home = self.root / "home"
      self.home.mkdir()
      self.maven_source = self.root / "maven"
      self.node_source = self.root / "node"
      (self.maven_source / "src" / "main" / "java").mkdir(parents=True)
      (self.maven_source / "src" / "main" / "java" / "App.java").write_text("class App {}", encoding="utf-8")
      (self.maven_source / "pom.xml").write_text("<project>root</project>", encoding="utf-8")
      (self.node_source / "src").mkdir(parents=True)
      (self.node_source / "src" / "index.js").write_text("run()", encoding="utf-8")
      (self.node_source / "package.json").write_text('{"name":"site"}', encoding="utf-8")
      (self.node_source / "package-lock.json").write_text('{"lockfileVersion":3}', encoding="utf-8")
      self.maven_target = plan_for(example_configuration("maven.ci.yaml"), event())["targets"][0]
      self.node_target = plan_for(example_configuration("node.ci.yaml"), event())["targets"][0]

   def environment(self, name: str = "push", ref: str = "refs/heads/main", default_branch: str = "main") -> dict[str, str]:
      """Return the event facts and home directory the metadata command reads."""
      return {"HOME": str(self.home), "CI_EVENT_NAME": name, "CI_EVENT_REF": ref, "CI_DEFAULT_BRANCH": default_branch, "ACTIONS_CACHE_URL": NATIVE_CACHE_URL}

   def maven_key(self, target: Target | None = None) -> str:
      """Return the Maven example's current cache key."""
      return build_cache_metadata(target or self.maven_target, self.maven_source, self.environment()).key

   def node_key(self, target: Target | None = None) -> str:
      """Return the Node example's current cache key."""
      return build_cache_metadata(target or self.node_target, self.node_source, self.environment()).key

   def link(self, link: Path, destination: Path) -> None:
      """Create a symbolic link or skip the test where the platform refuses."""
      try:
         link.symlink_to(destination, target_is_directory=destination.is_dir())
      except (OSError, NotImplementedError):
         self.skipTest("this platform does not allow the test to create symbolic links")


class CacheKeyTest(CacheTestCase):
   """The key follows dependency descriptors and runtime identity, never application source."""

   def test_source_edits_keep_the_key(self) -> None:
      """Editing or adding application source changes neither key."""
      maven_before, node_before = self.maven_key(), self.node_key()
      (self.maven_source / "src" / "main" / "java" / "App.java").write_text("class App { int changed; }", encoding="utf-8")
      (self.maven_source / "src" / "main" / "java" / "Other.java").write_text("class Other {}", encoding="utf-8")
      (self.node_source / "src" / "index.js").write_text("run(changed)", encoding="utf-8")
      (self.node_source / "README.md").write_text("docs", encoding="utf-8")
      self.assertEqual(self.maven_key(), maven_before)
      self.assertEqual(self.node_key(), node_before)

   def test_maven_dependency_edits_change_the_key(self) -> None:
      """The root pom, any reactor module pom and `.mvn` configuration and extensions each change the key."""
      keys = {self.maven_key()}
      (self.maven_source / "pom.xml").write_text("<project>edited</project>", encoding="utf-8")
      keys.add(self.maven_key())
      (self.maven_source / "module-a").mkdir()
      (self.maven_source / "module-a" / "pom.xml").write_text("<project>a</project>", encoding="utf-8")
      keys.add(self.maven_key())
      (self.maven_source / ".mvn").mkdir()
      (self.maven_source / ".mvn" / "maven.config").write_text("-T1", encoding="utf-8")
      keys.add(self.maven_key())
      (self.maven_source / ".mvn" / "extensions.xml").write_text("<extensions/>", encoding="utf-8")
      keys.add(self.maven_key())
      (self.maven_source / ".mvn" / "jvm.config").write_text("-Xmx1g", encoding="utf-8")
      keys.add(self.maven_key())
      self.assertEqual(len(keys), 6)

   def test_node_dependency_edits_change_the_key(self) -> None:
      """The lock file, package.json and a workspace package.json each change the key."""
      keys = {self.node_key()}
      (self.node_source / "package-lock.json").write_text('{"lockfileVersion":3,"packages":{}}', encoding="utf-8")
      keys.add(self.node_key())
      (self.node_source / "package.json").write_text('{"name":"site","dependencies":{"a":"1"}}', encoding="utf-8")
      keys.add(self.node_key())
      (self.node_source / "packages" / "ui").mkdir(parents=True)
      (self.node_source / "packages" / "ui" / "package.json").write_text("{}", encoding="utf-8")
      keys.add(self.node_key())
      self.assertEqual(len(keys), 4)

   def test_toolchain_image_platform_and_component_change_the_key(self) -> None:
      """The same descriptors under another toolchain, runtime image, platform or component never share a key."""
      baseline = self.maven_key()
      keys: set[str] = set()
      for field, value in (("toolchain_version", "25"), ("image", "registry.example/toolchain-java@sha256:" + "9" * 64), ("platform", "linux-arm64"), ("component", "other")):
         changed = cast("dict[str, object]", copy.deepcopy(self.maven_target))
         changed[field] = value
         keys.add(self.maven_key(cast("Target", changed)))
      self.assertNotIn(baseline, keys)
      self.assertEqual(len(keys), 4)
      self.assertNotEqual(self.maven_key(), self.node_key())

   def test_key_is_exact_and_names_no_repository(self) -> None:
      """The key is a single exact string made of the plan's own values and a descriptor digest, with no broad prefix to fall back to."""
      key = self.maven_key()
      self.assertTrue(key.startswith(f"platform-ci-deps-v1-maven-{self.maven_target['toolchain_version']}-linux-amd64-"))
      self.assertIn(f"-{self.maven_target['component']}-", key)
      self.assertNotIn(self.root.name, key)
      self.assertEqual(key, self.maven_key())

   def test_excluded_and_unrelated_files_are_not_descriptors(self) -> None:
      """Build output, installed dependencies, Git metadata, the platform checkout, settings and npmrc never reach the key."""
      maven_before, node_before = self.maven_key(), self.node_key()
      for relative in ("target/pom.xml", ".git/pom.xml", ".platform-ci/pom.xml", "ci-output/pom.xml", ".mvn/settings.xml", "settings.xml", "src/.mvn/settings.xml"):
         (self.maven_source / relative).parent.mkdir(parents=True, exist_ok=True)
         (self.maven_source / relative).write_text("<x>changed</x>", encoding="utf-8")
      for relative in ("node_modules/dep/package.json", ".git/package.json", ".platform-ci/package.json", "ci-output/package.json", ".npmrc"):
         (self.node_source / relative).parent.mkdir(parents=True, exist_ok=True)
         (self.node_source / relative).write_text("//registry/:_authToken=secret", encoding="utf-8")
      self.assertEqual(self.maven_key(), maven_before)
      self.assertEqual(self.node_key(), node_before)
      self.assertEqual(set(dependency_descriptors(self.maven_target, self.maven_source)), {"pom.xml"})
      self.assertEqual(set(dependency_descriptors(self.node_target, self.node_source)), {"package.json", "package-lock.json"})

   def test_missing_root_descriptor_is_an_error(self) -> None:
      """A component without a root pom or lock file gets no key, so the cache is never keyed on nothing."""
      (self.maven_source / "pom.xml").unlink()
      (self.node_source / "package-lock.json").unlink()
      with self.assertRaisesRegex(CacheError, "no pom.xml"):
         self.maven_key()
      with self.assertRaisesRegex(CacheError, "no package-lock.json"):
         self.node_key()


class CacheSafetyTest(CacheTestCase):
   """Symbolic links, unsafe directories and excluded names are rejected, and only download directories are selected."""

   def test_symlinked_descriptor_is_rejected(self) -> None:
      """A descriptor that links elsewhere cannot be trusted to describe this checkout."""
      outside = self.root / "outside.xml"
      outside.write_text("<project>elsewhere</project>", encoding="utf-8")
      (self.maven_source / "module-a").mkdir()
      self.link(self.maven_source / "module-a" / "pom.xml", outside)
      with self.assertRaisesRegex(CacheError, "symbolic link"):
         self.maven_key()

   def test_symlinked_directory_is_rejected_but_skipped_names_are_not_followed(self) -> None:
      """A linked directory could hide descriptors, so it is an error; a linked `node_modules` is skipped and never read."""
      (self.root / "elsewhere").mkdir()
      self.link(self.node_source / "linked", self.root / "elsewhere")
      with self.assertRaisesRegex(CacheError, "linked is a symbolic link"):
         self.node_key()
      (self.node_source / "linked").unlink()
      self.link(self.node_source / "node_modules", self.root / "elsewhere")
      self.assertTrue(self.node_key())

   def test_component_directory_outside_the_checkout_is_rejected(self) -> None:
      """The target's directory is resolved like every other stage, so it cannot leave the checkout."""
      target = copy.deepcopy(self.maven_target)
      target["directory"] = ".."
      with self.assertRaisesRegex(PipelineError, "not a directory inside the source checkout"):
         self.maven_key(target)

   def test_only_artifact_download_directories_are_selected(self) -> None:
      """Maven selects the local repository and npm its download cache, never build output, node_modules, settings or credentials."""
      maven = build_cache_metadata(self.maven_target, self.maven_source, self.environment())
      node = build_cache_metadata(self.node_target, self.node_source, self.environment())
      self.assertEqual(Path(maven.path), self.home / ".m2" / "repository")
      self.assertEqual(Path(node.path), self.home / ".npm" / "_cacache")
      for selected in (maven.path, node.path):
         self.assertNotIn(Path(self.maven_source).as_posix(), selected)
         self.assertNotIn(Path(self.node_source).as_posix(), selected)
         for excluded in ("target", "node_modules", "settings.xml", ".npmrc", ".git"):
            self.assertNotIn(excluded, Path(selected).parts)
      self.assertNotEqual(Path(maven.path), self.home / ".m2")

   def test_unsafe_cache_directories_are_rejected(self) -> None:
      """Relative homes, `..`, homes inside or around the checkout, excluded names and linked directories fail."""
      with self.assertRaisesRegex(CacheError, "absolute"):
         cache_directory("maven", Path("relative-home"), self.maven_source)
      with self.assertRaisesRegex(CacheError, "absolute"):
         cache_directory("maven", self.home / ".." / "home", self.maven_source)
      with self.assertRaisesRegex(CacheError, "overlaps the source checkout"):
         cache_directory("node", self.node_source / "home", self.node_source)
      with self.assertRaisesRegex(CacheError, "overlaps the source checkout"):
         cache_directory("maven", self.root, self.root / ".m2" / "repository" / "checkout")
      for excluded in ("node_modules", "target", ".git"):
         with self.subTest(excluded), self.assertRaisesRegex(CacheError, "excluded path name"):
            cache_directory("maven", self.root / excluded, self.maven_source)
      with self.assertRaisesRegex(CacheError, "no dependency cache"):
         cache_directory("gradle", self.home, self.maven_source)

   def test_symlinked_cache_directory_is_rejected(self) -> None:
      """A `.m2` or repository directory that is a link could point the cache anywhere."""
      (self.root / "elsewhere").mkdir()
      self.link(self.home / ".m2", self.root / "elsewhere")
      with self.assertRaisesRegex(CacheError, "symbolic link"):
         cache_directory("maven", self.home, self.maven_source)

   def test_missing_home_is_an_error(self) -> None:
      """Without HOME the download directory is unknown, so there is no cache."""
      with self.assertRaisesRegex(CacheError, "HOME is not set"):
         build_cache_metadata(self.maven_target, self.maven_source, {})


class CacheSaveEligibilityTest(unittest.TestCase):
   """Only a push to the event's own default branch may write the cache."""

   def test_only_default_branch_push_may_save(self) -> None:
      """Pull requests, forks' events, tags, feature branches, other events and unknown metadata never save."""
      self.assertTrue(save_allowed("push", "refs/heads/main", "main"))
      self.assertTrue(save_allowed("push", "refs/heads/trunk", "trunk"))
      for name, ref, default_branch in (
         ("pull_request", "refs/pull/7/merge", "main"),
         ("pull_request", "refs/heads/main", "main"),
         ("workflow_call", "refs/heads/main", "main"),
         ("push", "refs/heads/feature-x", "main"),
         ("push", "refs/heads/develop", "main"),
         ("push", "refs/tags/v1.0.0", "main"),
         ("push", "refs/heads/main", ""),
         ("push", "refs/heads/mainline", "main"),
      ):
         with self.subTest(name=name, ref=ref, default_branch=default_branch):
            self.assertFalse(save_allowed(name, ref, default_branch))


class CacheMetadataCommandTest(CacheTestCase):
   """The command publishes the key, path and save flag as workflow outputs, and fails closed."""

   def _run(self, environment: dict[str, str], source: Path, target: Target) -> tuple[int, dict[str, str]]:
      """Run the command and return its exit code and the outputs it wrote."""
      outputs_file = self.root / "outputs"
      outputs_file.unlink(missing_ok=True)
      full = {**environment, "CI_TARGET": to_compact_json(target), "GITHUB_OUTPUT": str(outputs_file)}
      code = main(["cache-metadata", "--source", str(source)], full)
      lines = outputs_file.read_text(encoding="utf-8").splitlines() if outputs_file.exists() else []
      return code, dict(line.split("=", 1) for line in lines)

   def test_outputs_match_the_library_and_save_flag_follows_the_event(self) -> None:
      """A default branch push may save; a pull request gets the same key but may not."""
      code, outputs = self._run(self.environment(), self.maven_source, self.maven_target)
      self.assertEqual(code, 0)
      self.assertEqual(outputs, {"available": "true", "key": self.maven_key(), "path": (self.home / ".m2" / "repository").as_posix(), "save": "true"})
      code, pull_request = self._run(self.environment("pull_request", "refs/pull/7/merge"), self.maven_source, self.maven_target)
      self.assertEqual(code, 0)
      self.assertEqual((pull_request["available"], pull_request["key"], pull_request["save"]), ("true", outputs["key"], "false"))

   def test_runner_provided_endpoint_is_used_as_given_without_any_allowlist(self) -> None:
      """Private HTTP pod addresses and HTTPS addresses alike are available; the platform neither rewrites nor restricts them."""
      for endpoint in (NATIVE_CACHE_URL, "http://10.9.8.7:8088", "https://cache.internal.example/"):
         with self.subTest(endpoint):
            self.assertIsNone(cache_unavailable_reason({"ACTIONS_CACHE_URL": endpoint}))
            code, outputs = self._run({**self.environment(), "ACTIONS_CACHE_URL": endpoint}, self.maven_source, self.maven_target)
            self.assertEqual((code, outputs["available"]), (0, "true"))
            self.assertNotIn(endpoint, "".join(outputs.values()))

   def test_absent_or_unusable_endpoint_reports_unavailable_and_publishes_nothing_to_restore_or_save(self) -> None:
      """Without a usable runner-provided endpoint the command succeeds with available=false, a note, no key or path, and never touches the checkout."""
      (self.maven_source / "pom.xml").unlink()
      for endpoint in (None, "", "   ", "not a url", "ftp://10.1.2.3:8088/", "http://"):
         with self.subTest(endpoint):
            environment = {name: value for name, value in self.environment().items() if name != "ACTIONS_CACHE_URL"}
            if endpoint is not None:
               environment["ACTIONS_CACHE_URL"] = endpoint
            code, outputs = self._run(environment, self.maven_source, self.maven_target)
            self.assertEqual(code, 0)
            self.assertEqual(set(outputs), {"available", "save", "note"})
            self.assertEqual((outputs["available"], outputs["save"]), ("false", "false"))
            self.assertIn("cache unavailable", outputs["note"])
            self.assertIn("build and tests still run", outputs["note"])

   def test_unsafe_input_exits_nonzero_without_outputs(self) -> None:
      """A rejected path yields an error and no key, so the workflow's cache steps do not run."""
      (self.maven_source / "pom.xml").unlink()
      code, outputs = self._run(self.environment(), self.maven_source, self.maven_target)
      self.assertEqual((code, outputs), (1, {}))
      code, _ = self._run({"HOME": str(self.home)}, self.node_source, self.node_target)
      self.assertEqual(code, 1)

   def test_target_json_is_what_selects_directory_and_image(self) -> None:
      """The matrix target's directory chooses what is hashed and its image chooses the key."""
      (self.maven_source / "service").mkdir()
      (self.maven_source / "service" / "pom.xml").write_text("<project>service</project>", encoding="utf-8")
      nested = copy.deepcopy(self.maven_target)
      nested["directory"] = "service"
      self.assertEqual(set(dependency_descriptors(nested, self.maven_source)), {"pom.xml"})
      self.assertNotEqual(self.maven_key(nested), self.maven_key())


class CacheWorkflowTest(unittest.TestCase):
   """The workflow wires the native cache actions safely and no cache result can skip verification."""

   @override
   def setUp(self) -> None:
      """Parse the one workflow and index the verify job's steps by name."""
      self.text = WORKFLOW_PATH.read_text(encoding="utf-8")
      workflow = cast("dict[str, dict[str, dict[str, object]]]", yaml.safe_load(self.text))
      self.steps = cast("list[dict[str, object]]", workflow["jobs"]["verify"]["steps"])
      self.other_steps = [step for job in ("prepare", "report") for step in cast("list[dict[str, object]]", workflow["jobs"][job]["steps"])]

   def _index(self, key: str, value: str) -> int:
      """Return the position of the one verify step whose field has the value."""
      (position,) = [position for position, step in enumerate(self.steps) if step.get(key) == value]
      return position

   def _action(self, name: str) -> dict[str, object]:
      """Return the one verify step that uses the named cache action."""
      (step,) = [step for step in self.steps if str(step.get("uses", "")).startswith(f"https://github.com/actions/cache/{name}@")]
      return step

   def test_restore_and_save_are_separate_native_actions_pinned_to_the_reviewed_revision(self) -> None:
      """Both come from the public upstream at the one reviewed commit, never a tag or branch, and only in the verify job."""
      for name in ("restore", "save"):
         self.assertEqual(self._action(name)["uses"], f"https://github.com/actions/cache/{name}@{CACHE_ACTION_REVISION}")
      self.assertNotIn("actions/cache@", self.text)
      self.assertFalse([step for step in self.other_steps if "cache" in str(step.get("uses", ""))])

   def test_keys_are_exact_with_no_fallback_and_only_computed_values_are_passed(self) -> None:
      """No restore-keys, no miss tolerance flags and no hard-coded path: the metadata step is the only source."""
      for name in ("restore", "save"):
         self.assertEqual(self._action(name)["with"], {"key": "${{ steps.cache.outputs.key }}", "path": "${{ steps.cache.outputs.path }}"})
      for name in ("restore", "save"):
         self.assertTrue({"restore-keys", "fail-on-cache-miss", "lookup-only", "enableCrossOsArchive"}.isdisjoint(cast("dict[str, object]", self._action(name)["with"])))

   def test_artifact_and_results_gate_is_unchanged_and_does_not_judge_the_cache_endpoint(self) -> None:
      """The private HTTPS runtime and results gate is exactly as before and runs first; the native cache endpoint is not held to it."""
      check = self._index("name", "Require private artifact service")
      self.assertEqual([line.strip() for line in cast("str", self.steps[check]["run"]).splitlines() if line.strip()], REQUIRED_GATE_LINES)
      self.assertNotIn("ACTIONS_CACHE_URL", self.text)
      self.assertNotIn("8088", self.text)
      self.assertLess(check, self._index("id", "cache"))
      self.assertLess(check, self.steps.index(self._action("restore")))
      self.assertLess(check, self.steps.index(self._action("save")))

   def test_restore_precedes_the_build_and_save_follows_it(self) -> None:
      """Order is metadata, restore, build, save; the build is neither reordered nor given a condition."""
      build = self._index("id", "build")
      self.assertLess(self._index("id", "cache"), self.steps.index(self._action("restore")))
      self.assertLess(self.steps.index(self._action("restore")), build)
      self.assertLess(build, self.steps.index(self._action("save")))
      self.assertNotIn("if", self.steps[build])
      self.assertNotIn("continue-on-error", self.steps[build])
      self.assertEqual(self.steps[build]["run"], "python3 -P -m platform_ci verify --source .")

   def test_save_runs_only_after_a_successful_build_on_a_default_branch_push(self) -> None:
      """The condition needs build success, the helper's default-branch verdict, a push, and no exact hit already."""
      condition = cast("str", self._action("save")["if"])
      for required in ("success()", "steps.build.outcome == 'success'", "steps.cache.outputs.save == 'true'", "github.event_name == 'push'", "steps.restore.outputs.cache-hit != 'true'"):
         self.assertIn(required, condition)
      self.assertNotIn("pull_request", condition)
      self.assertNotIn("always()", condition)

   def test_both_cache_actions_require_the_available_output_and_the_metadata_step_names_no_endpoint(self) -> None:
      """No endpoint means available=false, so neither action runs and nothing is retried; the workflow adds no endpoint setting or credential."""
      for name in ("restore", "save"):
         self.assertIn("steps.cache.outputs.available == 'true'", cast("str", self._action(name)["if"]))
         self.assertNotIn("always()", cast("str", self._action(name)["if"]))
      metadata_step = self.steps[self._index("id", "cache")]
      self.assertEqual(metadata_step["run"], "python3 -P -m platform_ci cache-metadata --source .")
      self.assertEqual(set(cast("dict[str, str]", metadata_step["env"])), {"CI_TARGET", "CI_EVENT_NAME", "CI_EVENT_REF", "CI_DEFAULT_BRANCH"})
      # the `secrets` credential context; `needs.secrets.result` is the secret-scan job's name, not a credential
      self.assertNotRegex(self.text, r"\$\{\{[^}]*(?<![.\w])secrets\.")

   def test_cache_problems_never_fail_or_skip_verification(self) -> None:
      """Cache steps tolerate their own failure, no other step reads a cache result, and no step skips on a hit."""
      for step in (self.steps[self._index("id", "cache")], self.steps[self._index("id", "cache-directory")], self._action("restore"), self._action("save")):
         self.assertIs(step["continue-on-error"], True)
      for step in self.steps:
         if step.get("id") in ("cache", "cache-directory", "restore") or str(step.get("uses", "")).startswith("https://github.com/actions/cache/"):
            continue
         self.assertNotIn("steps.cache", str(step), step.get("name"))
         self.assertNotIn("steps.restore", str(step), step.get("name"))

   def test_workflow_names_no_repository_and_caches_no_credentials(self) -> None:
      """The cache namespace comes from the service; no node_modules, target, settings or credential path appears in cache wiring."""
      for forbidden in ("GITHUB_REPOSITORY", "repository_owner", "REPOSITORY_NAME"):
         self.assertNotIn(forbidden, self.text)
      # only the prepare job's reuse step may carry the repository identity, and it is not part of any cache wiring
      self.assertNotIn("github.repository", json.dumps(self.steps))
      for name in ("restore", "save"):
         self.assertNotIn("node_modules", json.dumps(self._action(name)))
         self.assertNotIn("settings.xml", json.dumps(self._action(name)))