platform-ci/tests/test_pool_and_source.py

"""The native `source` tool (unchanged receipted helpers) and the pool job for runtime profiles.

Everything here runs locally. The source tests run the real helper scripts through the adapter on this machine's Node, not the qualified
runtime image. The pool tests prove planning, routing data and workflow structure only: they say nothing about whether the pool's mounts
and options work or isolate jobs on real runners, which is root's pending live qualification.
"""

import hashlib
import json
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path
from typing import Any, cast, override

import yaml

from platform_ci.build_tools import BUILD_TOOLS, SOURCE_CHECK_PROGRAM
from platform_ci.configuration import ConfigurationError, parse_project_configuration
from platform_ci.execution import ExecutionError, run_target_commands
from platform_ci.plan import Target
from platform_ci.runtimes import load_runtime_catalog, parse_runtime_catalog
from platform_ci.source_helpers import SourceHelperError, verify_source_helpers
from tests.support import JAVA_IMAGE, NODE_IMAGE, event, example_mapping, plan_for

ROOT = Path(__file__).resolve().parent.parent
HELPERS = ROOT / "source_checks"
BROWSER_IMAGE = f"registry.example/java-browser@sha256:{'7' * 64}"
POOL_CATALOG = parse_runtime_catalog(
   {
      "schema": 1,
      "runners": {"linux-amd64": {"browser": "pool-x", "docker": "pool-x", "browser+docker": "pool-x"}},
      "toolchains": {"java": {"21": {"image": JAVA_IMAGE, "browser_image": BROWSER_IMAGE}}, "node": {"22.20.0": {"image": NODE_IMAGE, "browser_image": None}}},
   }
)
WORKFLOW_TEXT = (ROOT / ".gitea" / "scoped_workflows" / "pipeline.yaml").read_text(encoding="utf-8")
JOBS = cast("dict[str, dict[str, Any]]", yaml.safe_load(WORKFLOW_TEXT)["jobs"])


def source_configuration(commands: object = (("tracked-files",),), **component: object) -> dict[str, object]:
   """A one-component `source` configuration."""
   return {"schema": 1, "components": {"source": {"toolchains": {"node": "22.20.0"}, "build": {"tool": "source", "commands": [list(command) for command in cast("tuple[tuple[str, ...], ...]", commands)]}, **component}}}


def pool_configuration(*profiles: str) -> dict[str, object]:
   """The Maven example with runtime profiles required."""
   configuration = example_mapping("maven.ci.yaml")
   cast("dict[str, dict[str, Any]]", configuration["components"])["library"]["requires"] = list(profiles)
   return configuration


class SourceConfigurationTest(unittest.TestCase):
   """`tool: source` accepts only its closed vocabulary, in the repository root."""

   def test_tracked_files_parses_in_both_command_forms(self) -> None:
      parse_project_configuration(source_configuration())
      arguments = source_configuration()
      build = cast("dict[str, dict[str, Any]]", arguments["components"])["source"]["build"]
      del build["commands"]
      build["arguments"] = ["tracked-files"]
      self.assertEqual(parse_project_configuration(arguments).components[0].build_commands, (("tracked-files",),))

   def test_unknown_subcommands_extra_arguments_and_programs_are_rejected(self) -> None:
      for name, commands in {
         "unknown subcommand": (("actionlint",),),
         "extra argument": (("tracked-files", "--fix"),),
         "program path": (("./scripts/check.sh",),),
         "interpreter": (("node", "x.mjs"),),
         "shell text": (("tracked-files; id",),),
         "empty": ((),),
      }.items():
         with self.subTest(name), self.assertRaises(ConfigurationError):
            parse_project_configuration(source_configuration(commands))

   def test_only_the_repository_root_is_accepted(self) -> None:
      with self.assertRaisesRegex(ConfigurationError, "directory must be '.'"):
         parse_project_configuration(source_configuration(directory="subproject"))
      parse_project_configuration(source_configuration(directory="."))


class SourcePlanTest(unittest.TestCase):
   """A source component plans the native helper on the qualified Node runtime and nothing else."""

   def _target(self, repository_label: str = "any") -> Target:
      del repository_label
      return plan_for(parse_project_configuration(source_configuration()), event())["targets"][0]

   def test_plans_the_receipted_program_without_install_or_extra_arguments(self) -> None:
      target = self._target()
      self.assertEqual(target["commands"], [["node", SOURCE_CHECK_PROGRAM]])
      self.assertEqual((target["image"], target["runner"], target["execution"], target["requires"]), (NODE_IMAGE, "linux-small", "container", []))
      self.assertEqual(BUILD_TOOLS["source"].install_commands, ())
      self.assertEqual(target["environment"], {})

   def test_identical_configuration_plans_identically_whatever_repository_runs_it(self) -> None:
      self.assertEqual(self._target("first"), self._target("second"))
      for module in ("build_tools.py", "plan.py", "source_helpers.py"):
         text = (ROOT / "platform_ci" / module).read_text(encoding="utf-8")
         self.assertNotRegex(text, r"KofTwentyTwo/|QRun-IO/|repository_owner")

   def test_the_shipped_catalog_resolves_source_to_the_qualified_node_image_on_the_standard_pool_only(self) -> None:
      runtime = load_runtime_catalog().resolve_runtime("node", "22.20.0", "linux-amd64")
      self.assertEqual((runtime.runner, runtime.image.split("@")[0]), ("linux-small", "git.kof22.com/koftwentytwo/platform-ci-node"))


class SourceHelperReceiptTest(unittest.TestCase):
   """The vendored helpers are byte-identical to root's receipted files and to the captured source blobs."""

   def test_helpers_match_the_receipt_and_the_captured_blobs(self) -> None:
      receipt = json.loads((HELPERS / "receipt.json").read_text(encoding="utf-8"))
      self.assertEqual(receipt["files"], {"check-source.mjs": "c1b6c2f925ac1c4c4bccc3fd8675c4bc0118ad55a03be8b39d2d10f176879677", "jsonc.mjs": "a71f0f1dc416fd4751c0106ee31b3dc86b30f44e64bf8ecb30922a5d7ac3b4e9"})
      verify_source_helpers(HELPERS)
      # git blob ids of the files captured at the reviewed revisions (identical at ab38e255, c95ea023, d4de48e2 and dfe06089)
      for name, blob in (("check-source.mjs", "f50d41480d75df904b93fdd95c00d1e8b84c2476"), ("jsonc.mjs", "f88b146a334805cc5ff9f810acfd064f6572861a")):
         data = (HELPERS / name).read_bytes()
         self.assertEqual(hashlib.sha1(b"blob " + str(len(data)).encode() + b"\0" + data).hexdigest(), blob, name)  # noqa: S324 - a git blob id, not a security digest

   def test_a_changed_or_missing_helper_stops_the_run(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         copy_directory = Path(directory) / "source_checks"
         shutil.copytree(HELPERS, copy_directory)
         verify_source_helpers(copy_directory)
         (copy_directory / "jsonc.mjs").write_bytes((HELPERS / "jsonc.mjs").read_bytes() + b"\n// edited\n")
         with self.assertRaisesRegex(SourceHelperError, "differs from its receipted revision"):
            verify_source_helpers(copy_directory)
         (copy_directory / "jsonc.mjs").unlink()
         with self.assertRaisesRegex(SourceHelperError, "missing beside the program"):
            verify_source_helpers(copy_directory)


class Recorder:
   """A command stub that records the command and the environment it would run with."""

   def __init__(self) -> None:
      """Start empty."""
      self.calls: list[tuple[list[str], dict[str, str]]] = []

   def __call__(self, command: Any, working_directory: Path, environment: Any) -> int:
      """Record and succeed."""
      del working_directory
      self.calls.append((list(command), dict(environment)))
      return 0


class SourceExecutionTest(unittest.TestCase):
   """The adapter runs exactly the platform checkout's helper in the repository root, with a scrubbed environment."""

   @override
   def setUp(self) -> None:
      """Create a checkout with the platform's helper directory beside it, as the workflow does."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.source = Path(directory.name) / "checkout"
      shutil.copytree(HELPERS, self.source / ".platform-ci" / "source_checks")
      self.target = plan_for(parse_project_configuration(source_configuration()), event())["targets"][0]

   def test_runs_the_one_program_with_no_install_no_python_override_and_no_credentials(self) -> None:
      recorder = Recorder()
      parent = {"PATH": "/usr/bin", "HOME": "/tmp", "CHECK_SOURCE_PYTHON": "/evil/python", "GITHUB_TOKEN": "secret"}  # noqa: S108
      self.assertTrue(run_target_commands(self.target, self.source, recorder, parent))
      self.assertEqual([call[0] for call in recorder.calls], [["node", SOURCE_CHECK_PROGRAM]])
      for _command, environment in recorder.calls:
         self.assertNotIn("CHECK_SOURCE_PYTHON", environment)
         self.assertNotIn("GITHUB_TOKEN", environment)

   def test_a_missing_or_altered_helper_is_an_error_not_a_pass(self) -> None:
      (self.source / ".platform-ci" / "source_checks" / "jsonc.mjs").write_text("export const jsonc = () => 1\n", encoding="utf-8")
      with self.assertRaises(SourceHelperError):
         run_target_commands(self.target, self.source, Recorder(), {"PATH": "/usr/bin"})
      shutil.rmtree(self.source / ".platform-ci")
      with self.assertRaisesRegex(ExecutionError, "is not a file inside the component directory"):
         run_target_commands(self.target, self.source, Recorder(), {"PATH": "/usr/bin"})


@unittest.skipUnless(all(shutil.which(tool) for tool in ("node", "git", "bash", "python3")), "needs node, git, bash and python3 on PATH")
class SourceRuntimeBehaviourTest(unittest.TestCase):
   """Root's offline proof cases, through the adapter, on the local Node (the qualified image itself is proven by root's receipt)."""

   @override
   def setUp(self) -> None:
      """Create a Git fixture tracking valid JSON, JSONC, shell and Python."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.source = Path(directory.name) / "fixture"
      (self.source / ".vscode").mkdir(parents=True)
      shutil.copytree(HELPERS, self.source / ".platform-ci" / "source_checks")
      self.write("config.json", '{"valid":true}\n')
      self.write(".vscode/settings.json", '{// comment\n"valid":true,}\n')
      self.write("check.sh", "#!/bin/sh\ntrue\n")
      self.write("check.py", 'raise RuntimeError("must never execute")\n')
      self.git("init", "-q")
      self.git("add", "config.json", ".vscode/settings.json", "check.sh", "check.py")
      self.target = plan_for(parse_project_configuration(source_configuration()), event())["targets"][0]

   def write(self, name: str, text: str) -> None:
      """Write a tracked fixture file."""
      (self.source / name).write_text(text, encoding="utf-8", newline="\n")

   def git(self, *arguments: str) -> None:
      """Run git in the fixture."""
      subprocess.run(["git", "-C", str(self.source), *arguments], check=True, capture_output=True)  # noqa: S603, S607

   def verdict(self) -> bool:
      """Run the adapter with the real process environment, which the sandbox filters to its allow-list."""
      return run_target_commands(self.target, self.source)

   def test_valid_files_pass_without_executing_python_or_writing_bytecode(self) -> None:
      self.assertTrue(self.verdict())
      self.assertFalse(list(self.source.rglob("__pycache__")))

   def test_invalid_json_shell_and_python_each_reject(self) -> None:
      for name, text, restore in (("config.json", "{invalid}\n", '{"valid":true}\n'), ("check.sh", "if\n", "#!/bin/sh\ntrue\n"), ("check.py", "def broken(\n", 'raise RuntimeError("must never execute")\n')):
         with self.subTest(name):
            self.write(name, text)
            self.assertFalse(self.verdict())
            self.write(name, restore)
            self.assertTrue(self.verdict())
      self.assertFalse(list(self.source.rglob("__pycache__")))