"""The `source-tools` build tool: closed subcommands with validated paths, receipted unchanged helpers, and no tool downloads. The planning and receipt tests are exact. The execution tests run the real, receipted helper scripts through the adapter on this machine's `bash`, with fixture tools standing in for the image's pinned binaries; the published image itself, and each real tool in it, is proven by root's receipt for that image, not here. """ import hashlib import json import os import shutil import stat 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_TOOLS_SUBCOMMANDS from platform_ci.configuration import ConfigurationError, parse_project_configuration from platform_ci.execution import run_target_commands from platform_ci.packaging import PackagingError, RunIdentity, collect_outputs from platform_ci.plan import Target from platform_ci.runtimes import load_runtime_catalog from platform_ci.source_helpers import SourceHelperError, verify_source_helpers from platform_ci.source_tools import HELPER_DIRECTORY, RECEIPT_PATH from tests.support import EXAMPLES_DIRECTORY, PLATFORM_REVISION, SOURCE_REVISION, SOURCE_TREE, event, example_mapping, plan_for ROOT = Path(__file__).resolve().parent.parent HELPERS = ROOT / "source_tools" PUBLISHED_IMAGE = "git.kof22.com/koftwentytwo/platform-ci-source-tools@sha256:f50fdd17ff1b4d06a3fe22f089917898c3367fc2be0c10628c32989960542b63" CAPTURED_REVISION = "ab38e255e5ae7a000c71fd2c1b4a5c2cd518de4d" ALL_SUBCOMMANDS = ("actionlint", "shellcheck-orb", "make-bats", "kustomize-render", "blueprint-lint", "policy") POLICY_REVISION = "c95ea023eae447f482ae6ce4d61efbcb9c79a017" POLICY_VALIDATOR_SHA256 = "ad9e230bafa10816c0f8651cb0ada89c54168a07f86fa72fb1a048265a29b83e" """The captured, unchanged `scripts/policy-validate.py` of the policy-verify action at `POLICY_REVISION`.""" def configuration(*commands: tuple[str, ...], **component: object) -> dict[str, object]: """A one-component `source-tools` configuration.""" body: dict[str, object] = {"toolchains": {"source-tools": "1"}, "build": {"tool": "source-tools", "commands": [list(command) for command in commands]}, **component} return {"schema": 1, "components": {"checks": body}} def planned_target(*commands: tuple[str, ...]) -> Target: """Plan one `source-tools` component against the shipped catalog and return its target.""" return plan_for(parse_project_configuration(configuration(*commands)), event(), load_runtime_catalog())["targets"][0] class SourceToolsConfigurationTest(unittest.TestCase): """Only the five subcommands parse, only two take paths, and every path must be a safe repository-relative directory.""" def test_every_subcommand_parses_in_both_command_forms(self) -> None: for command in (("actionlint",), ("shellcheck-orb",), ("make-bats",), ("kustomize-render", "deploy/a", "deploy/b"), ("blueprint-lint", "blueprints"), ("policy",)): with self.subTest(command): self.assertEqual(parse_project_configuration(configuration(command)).components[0].build_commands, (command,)) mapping = configuration() build = cast("dict[str, dict[str, Any]]", mapping["components"])["checks"]["build"] del build["commands"] build["arguments"] = ["kustomize-render", "deploy"] self.assertEqual(parse_project_configuration(mapping).components[0].build_commands, (("kustomize-render", "deploy"),)) def test_the_example_parses_and_declares_every_subcommand(self) -> None: parsed = parse_project_configuration(example_mapping("source-tools.ci.yaml")) self.assertEqual([command[0] for command in parsed.components[0].build_commands], list(ALL_SUBCOMMANDS[:5])) (policy,) = parse_project_configuration(example_mapping("source-policy.ci.yaml")).components self.assertEqual((policy.build_tool, policy.build_commands, policy.directory), ("source-tools", (("policy",),), ".")) def test_rendered_manifests_are_retained_by_the_existing_generic_artifact_paths(self) -> None: (component,) = parse_project_configuration(example_mapping("source-manifests.ci.yaml")).components self.assertEqual(component.artifacts, {"rendered-manifests": ("ci-manifests/**/*",)}) (target,) = plan_for(parse_project_configuration(example_mapping("source-manifests.ci.yaml")), event(), load_runtime_catalog())["targets"] self.assertEqual(target["artifacts"], {"rendered-manifests": ["ci-manifests/**/*"]}) def test_unknown_subcommands_programs_and_arguments_are_rejected(self) -> None: for name, command in { "tracked-files belongs to the source tool": ("tracked-files",), "unknown": ("helm-lint",), "program path": ("./scripts/check.sh",), "interpreter": ("bash", "-c", "id"), "shell text": ("actionlint; id",), "actionlint takes no arguments": ("actionlint", "x"), "shellcheck-orb takes no arguments": ("shellcheck-orb", "."), "make-bats takes no arguments": ("make-bats", "--fix"), "policy takes no arguments": ("policy", "."), "policy has no shell option": ("policy", "--shell"), "kustomize-render needs a root": ("kustomize-render",), "blueprint-lint needs a path": ("blueprint-lint",), "empty": (), }.items(): with self.subTest(name), self.assertRaises(ConfigurationError): parse_project_configuration(configuration(command)) def test_unsafe_paths_are_rejected_at_prepare(self) -> None: for path in ("/etc", "../outside", "a/../../b", "a/..", "..", ".", "./", "", "has space", "a,b", "a;id", "$HOME", "`id`", "a\nb", "ünï", "a*", "~/x"): for subcommand in ("kustomize-render", "blueprint-lint"): with self.subTest(subcommand=subcommand, path=path), self.assertRaises(ConfigurationError): parse_project_configuration(configuration((subcommand, path))) for path in ("deploy", "./deploy", "deploy/overlays/prod", "a.b/c_d-e/", "blueprints/custom"): parse_project_configuration(configuration(("kustomize-render", path))) def test_only_the_repository_root_is_accepted_and_a_check_is_not_declared_twice(self) -> None: with self.assertRaisesRegex(ConfigurationError, "directory must be '.'"): parse_project_configuration(configuration(("actionlint",), directory="sub")) with self.assertRaisesRegex(ConfigurationError, "declared twice"): parse_project_configuration(configuration(("actionlint",), ("actionlint",))) with self.assertRaises(ConfigurationError): parse_project_configuration(configuration(("actionlint",), toolchains={"node": "22.20.0"})) def test_the_existing_source_tool_is_unchanged(self) -> None: self.assertEqual(BUILD_TOOLS["source"].subcommands, (("tracked-files", ("node", ".platform-ci/source_checks/check-source.mjs")),)) self.assertEqual(BUILD_TOOLS["source"].helper_subcommands, ()) self.assertEqual(BUILD_TOOLS["source"].toolchain, "node") parse_project_configuration(example_mapping("source.ci.yaml")) with self.assertRaises(ConfigurationError): parse_project_configuration({"schema": 1, "components": {"source": {"toolchains": {"node": "22.20.0"}, "build": {"tool": "source", "commands": [["actionlint"]]}}}}) class SourceToolsPlanTest(unittest.TestCase): """The plan names the small published image on the standard pool, carries only validated data and never mentions a repository.""" def test_plans_the_published_image_on_linux_small_with_the_declared_commands(self) -> None: target = planned_target(("actionlint",), ("kustomize-render", "deploy/prod")) self.assertEqual(target["commands"], [["actionlint"], ["kustomize-render", "deploy/prod"]]) self.assertEqual((target["image"], target["runner"], target["execution"], target["requires"]), (PUBLISHED_IMAGE, "linux-small", "container", [])) self.assertEqual(target["environment"], {"PYYAML_DIR": "/opt/blueprint-tools"}) self.assertEqual(BUILD_TOOLS["source-tools"].install_commands, ()) self.assertIn("Source tools 1", target["display"]) def test_the_lightweight_source_tool_still_plans_the_node_image_alone(self) -> None: target = plan_for(parse_project_configuration(example_mapping("source.ci.yaml")), event(), load_runtime_catalog())["targets"][0] self.assertEqual(target["commands"], [["node", ".platform-ci/source_checks/check-source.mjs"]]) self.assertTrue(target["image"].startswith("git.kof22.com/koftwentytwo/platform-ci-node@")) self.assertNotIn("PYYAML_DIR", target["environment"]) def test_the_shipped_catalog_has_the_published_digest_and_no_new_runner_profile(self) -> None: runtime = load_runtime_catalog().resolve_runtime("source-tools", "1", "linux-amd64") self.assertEqual((runtime.image, runtime.runner), (PUBLISHED_IMAGE, "linux-small")) self.assertEqual(BUILD_TOOLS["source-tools"].profiles, ()) def test_plans_are_identical_whatever_repository_runs_them_and_no_name_is_branched_on(self) -> None: self.assertEqual(planned_target(("make-bats",)), planned_target(("make-bats",))) for module in ("build_tools.py", "plan.py", "source_tools.py", "execution.py"): text = (ROOT / "platform_ci" / module).read_text(encoding="utf-8") self.assertNotRegex(text, r"KofTwentyTwo/|QRun-IO/|repository_owner") class SourceToolsReceiptTest(unittest.TestCase): """The vendored helpers keep their captured bytes and provenance; the one platform-authored file is listed as such.""" def test_helpers_match_the_receipt_and_the_captured_blobs(self) -> None: receipt = json.loads(RECEIPT_PATH.read_text(encoding="utf-8")) self.assertEqual(receipt["source_revision"], CAPTURED_REVISION) self.assertEqual(receipt["runtime"], PUBLISHED_IMAGE) verify_source_helpers(HELPERS, RECEIPT_PATH) self.assertEqual(receipt["platform_authored"], ["scripts/require-installed-tool.sh", "scripts/policy-verify.sh"]) self.assertEqual(sorted(receipt["files"]), sorted([*receipt["git_blobs"], *receipt["platform_authored"]])) for name, digest in receipt["files"].items(): self.assertEqual(hashlib.sha256((HELPERS / name).read_bytes()).hexdigest(), digest, name) for name, blob in receipt["git_blobs"].items(): 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 self.assertNotIn(b"\r", data, name) def test_the_policy_validator_is_the_captured_file_and_its_launcher_adds_only_isolation(self) -> None: receipt = json.loads(RECEIPT_PATH.read_text(encoding="utf-8")) self.assertEqual(receipt["files"]["scripts/policy-validate.py"], POLICY_VALIDATOR_SHA256) self.assertEqual(receipt["additional_captures"]["scripts/policy-validate.py"]["source_revision"], POLICY_REVISION) captured = ROOT.parent / "migration" / "current-ci-settings" / "platform-actions" / POLICY_REVISION / "scripts" / "policy-validate.py" if captured.exists(): self.assertEqual((HELPERS / "scripts" / "policy-validate.py").read_bytes(), captured.read_bytes()) launcher = (HELPERS / "scripts" / "policy-verify.sh").read_text(encoding="utf-8") self.assertIn('python3 -P -s -B "$here/policy-validate.py"', launcher) self.assertIn("unset QQQ_POLICY_SHELL", launcher) for forbidden in ("curl", "pip", "install-tool", "http"): self.assertNotIn(forbidden, launcher.replace("install-tool.sh", ""), forbidden) def test_the_downloading_installer_is_not_vendored_and_no_helper_calls_it(self) -> None: self.assertFalse((HELPERS / "scripts" / "install-tool.sh").exists()) for script in (HELPERS / "scripts").iterdir(): if script.name != "require-installed-tool.sh": self.assertNotIn("curl", script.read_text(encoding="utf-8"), script.name) def test_a_changed_or_missing_helper_stops_the_run(self) -> None: with tempfile.TemporaryDirectory() as directory: copy_directory = Path(directory) / "source_tools" shutil.copytree(HELPERS, copy_directory) verify_source_helpers(copy_directory, RECEIPT_PATH) (copy_directory / "scripts" / "manifest-verify.sh").write_bytes((HELPERS / "scripts" / "manifest-verify.sh").read_bytes() + b"\ntrue\n") with self.assertRaisesRegex(SourceHelperError, "differs from its receipted revision"): verify_source_helpers(copy_directory, RECEIPT_PATH) (copy_directory / "scripts" / "manifest-verify.sh").unlink() with self.assertRaisesRegex(SourceHelperError, "missing beside the program"): verify_source_helpers(copy_directory, RECEIPT_PATH) def test_the_vocabulary_is_data_and_each_script_is_receipted(self) -> None: receipt = json.loads(RECEIPT_PATH.read_text(encoding="utf-8")) self.assertEqual([entry.name for entry in SOURCE_TOOLS_SUBCOMMANDS], list(ALL_SUBCOMMANDS)) for entry in SOURCE_TOOLS_SUBCOMMANDS: self.assertIn(f"scripts/{entry.script}", receipt["files"]) class Recorder: """A command stub that records each command with 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 SourceToolsExecutionTest(unittest.TestCase): """The adapter verifies the receipt, then starts each unchanged helper once with variables built only from validated plan data.""" @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 / HELPER_DIRECTORY) self.target = planned_target(*[(name,) if name not in {"kustomize-render", "blueprint-lint"} else (name, "deploy/prod", "deploy/stage") for name in ALL_SUBCOMMANDS]) def test_each_subcommand_starts_its_helper_once_with_validated_inputs_and_no_credentials(self) -> None: recorder = Recorder() parent = {"PATH": "/usr/bin", "HOME": "/tmp", "GITHUB_TOKEN": "secret", "INPUT_ROOTS": "/etc", "INSTALL_TOOL": "/evil", "QQQ_POLICY_SHELL": "true", "PYTHONPATH": "/evil"} # noqa: S108 self.assertTrue(run_target_commands(self.target, self.source, recorder, parent)) scripts = (self.source / HELPER_DIRECTORY).resolve() / "scripts" self.assertEqual( [call[0] for call in recorder.calls], [["bash", str(scripts / script)] for script in ("actionlint-verify.sh", "source-checks.sh", "source-checks.sh", "manifest-verify.sh", "blueprint-lint.sh", "policy-verify.sh")], ) environments = [call[1] for call in recorder.calls] for environment in environments: self.assertNotIn("GITHUB_TOKEN", environment) self.assertEqual(environment["INSTALL_TOOL"], str(scripts / "require-installed-tool.sh")) self.assertEqual(environment["PYYAML_DIR"], "/opt/blueprint-tools") self.assertEqual([environment.get("INPUT_CHECKS") for environment in environments], [None, "shellcheck-orb", "make-bats", None, None, None]) trust = {"GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "safe.directory", "GIT_CONFIG_VALUE_0": self.source.resolve().as_posix()} for index, environment in enumerate(environments): trusted = {key: environment.get(key) for key in trust} == trust self.assertEqual(trusted, ALL_SUBCOMMANDS[index] in {"make-bats", "policy"}, ALL_SUBCOMMANDS[index]) self.assertEqual([name for name in ("QQQ_POLICY_SHELL", "PYTHONPATH") if name in environments[5]], []) self.assertEqual(environments[3]["INPUT_ROOTS"], "deploy/prod,deploy/stage") self.assertNotIn("INPUT_ROOTS", environments[4]) self.assertEqual(environments[4]["INPUT_PATHS"], "deploy/prod,deploy/stage") self.assertNotIn("INPUT_PATHS", environments[0]) def test_a_missing_or_altered_helper_is_an_error_not_a_pass(self) -> None: recorder = Recorder() with (self.source / HELPER_DIRECTORY / "scripts" / "source-checks.sh").open("ab") as handle: handle.write(b"\ntrue\n") with self.assertRaises(SourceHelperError): run_target_commands(self.target, self.source, recorder, {"PATH": "/usr/bin"}) self.assertEqual(recorder.calls, []) shutil.rmtree(self.source / HELPER_DIRECTORY) with self.assertRaisesRegex(SourceHelperError, "not a directory"): run_target_commands(self.target, self.source, recorder, {"PATH": "/usr/bin"}) self.assertEqual(recorder.calls, []) def test_a_command_outside_the_vocabulary_in_a_hand_made_plan_is_refused(self) -> None: forged = cast("Target", {**self.target, "commands": [["actionlint", "x"]]}) with self.assertRaises(SourceHelperError): run_target_commands(forged, self.source, Recorder(), {"PATH": "/usr/bin"}) forged = cast("Target", {**self.target, "commands": [["kustomize-render", "../../etc"]]}) with self.assertRaises(SourceHelperError): run_target_commands(forged, self.source, Recorder(), {"PATH": "/usr/bin"}) def test_stops_at_the_first_failing_check(self) -> None: calls: list[str] = [] def failing(command: Any, working_directory: Path, environment: Any) -> int: del working_directory, environment calls.append(str(command[1])) return 1 self.assertFalse(run_target_commands(self.target, self.source, failing, {"PATH": "/usr/bin"})) self.assertEqual(len(calls), 1) def usable_bash() -> str | None: """Return a POSIX bash that can open this checkout's files (not Windows' WSL launcher), or None.""" git = shutil.which("git") candidates = [shutil.which("bash"), str(Path(git).resolve().parent.parent / "bin" / "bash.exe") if git else None] probe = (HELPERS / "tools.env").as_posix() for candidate in candidates: if candidate and Path(candidate).exists(): result = subprocess.run([candidate, "-c", 'test -f "$0"', probe], capture_output=True, check=False) # noqa: S603 if result.returncode == 0: return candidate return None BASH = usable_bash() def run_with_bash(command: Any, working_directory: Path, environment: Any) -> int: """Start the command with the usable bash; on Windows, write paths with forward slashes so a POSIX shell reads them.""" assert BASH is not None arguments = [BASH if index == 0 else str(argument).replace("\\", "/") for index, argument in enumerate(command)] forwarded = {name: value.replace("\\", "/") if name in {"INSTALL_TOOL", "SOURCE_TOOLS_DIR", "HOME"} else value for name, value in environment.items()} return subprocess.run(arguments, cwd=working_directory, env=forwarded, check=False).returncode # noqa: S603 @unittest.skipUnless(BASH is not None, "needs a POSIX bash that can read this checkout") class SourceToolsHelperBehaviourTest(unittest.TestCase): """The real helpers through the adapter, with fixture binaries: failures stay failures and nothing is ever downloaded.""" @override def setUp(self) -> None: """Create a checkout, a fixture tool directory laid out like the image's, and a PATH that holds only fixture tools.""" directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) base = Path(directory.name) self.source = base / "checkout" shutil.copytree(HELPERS, self.source / HELPER_DIRECTORY) self.tools = base / "image" (self.tools / "bin").mkdir(parents=True) (self.tools / "bats" / "bin").mkdir(parents=True) tools_env = dict(line.split("=", 1) for line in (HELPERS / "tools.env").read_text(encoding="utf-8").splitlines() if re_assignment(line)) self.actionlint_version = tools_env["ACTIONLINT_VERSION"].strip("\"'") self.fixture_tool("actionlint", f'[ "$1" = --version ] && {{ echo {self.actionlint_version}; exit 0; }}\ngrep -q BAD "$@" && exit 1\nexit 0\n') self.fixture_tool("shellcheck", 'for file; do :; done\ngrep -q SMELL "$file" && echo "$file: SC0000 smell"\nexit 0\n') for name in ("yq", "jq", "envsubst"): self.fixture_tool(name, "exit 0\n") bats = self.tools / "bats" / "bin" / "bats" bats.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8", newline="\n") bats.chmod(bats.stat().st_mode | stat.S_IEXEC) self.fixture_tool("kubectl", 'case "$1" in kustomize) [ -f "$2/kustomization.yaml" ] || exit 1; grep -q BROKEN "$2/kustomization.yaml" && exit 1; echo "kind: ConfigMap";; esac\n', directory=self.tools / "pathbin") (self.source / ".github" / "workflows").mkdir(parents=True) (self.source / ".github" / "workflows" / "ci.yml").write_text("name: ok\n", encoding="utf-8") (self.source / "deploy" / "prod").mkdir(parents=True) (self.source / "deploy" / "prod" / "kustomization.yaml").write_text("resources: []\n", encoding="utf-8") (self.source / "Makefile").write_text("test-scripts test-templates:\n\t@true\n", encoding="utf-8") (self.source / "tests").mkdir() (self.source / "script.sh").write_text("#!/bin/sh\ntrue\n", encoding="utf-8", newline="\n") def fixture_tool(self, name: str, body: str, directory: Path | None = None) -> None: """Write an executable fixture binary.""" target_directory = directory or self.tools / "bin" target_directory.mkdir(parents=True, exist_ok=True) path = target_directory / name path.write_text(f"#!/bin/sh\n{body}", encoding="utf-8", newline="\n") path.chmod(path.stat().st_mode | stat.S_IEXEC) def verdict(self, *commands: tuple[str, ...], tools: Path | None = None) -> bool: """Plan the commands, point the helper's resolver at the fixture tools and run them with a PATH of fixture tools plus the system's.""" target = planned_target(*commands) target_environment = {**target["environment"], "SOURCE_TOOLS_DIR": (tools or self.tools).as_posix()} runnable = cast("Target", {**target, "environment": target_environment}) parent = {"PATH": os.pathsep.join([str(self.tools / "pathbin"), os.environ["PATH"]]), "HOME": str(self.source.parent)} return run_target_commands(runnable, self.source, run_with_bash, parent_environment=parent) def test_actionlint_passes_fails_and_still_fails_on_an_empty_glob(self) -> None: self.assertTrue(self.verdict(("actionlint",))) (self.source / ".github" / "workflows" / "ci.yml").write_text("BAD\n", encoding="utf-8") self.assertFalse(self.verdict(("actionlint",))) (self.source / ".github" / "workflows" / "ci.yml").unlink() self.assertFalse(self.verdict(("actionlint",))) def test_shellcheck_orb_keeps_any_output_a_failure(self) -> None: self.assertTrue(self.verdict(("shellcheck-orb",))) (self.source / "script.sh").write_text("#!/bin/sh\nSMELL\n", encoding="utf-8") self.assertFalse(self.verdict(("shellcheck-orb",))) def test_make_bats_runs_make_and_bats_from_the_installed_tools(self) -> None: if shutil.which("make") is None: self.skipTest("needs make on PATH") self.assertTrue(self.verdict(("make-bats",))) (self.source / "Makefile").write_text("test-scripts test-templates:\n\t@false\n", encoding="utf-8") self.assertFalse(self.verdict(("make-bats",))) def test_kustomize_render_passes_and_fails_per_root(self) -> None: self.assertTrue(self.verdict(("kustomize-render", "deploy/prod"))) self.assertFalse(self.verdict(("kustomize-render", "deploy/prod", "deploy/missing"))) (self.source / "deploy" / "prod" / "kustomization.yaml").write_text("BROKEN\n", encoding="utf-8") self.assertFalse(self.verdict(("kustomize-render", "deploy/prod"))) def test_a_tool_that_is_not_installed_fails_instead_of_downloading(self) -> None: empty = self.tools.parent / "empty-image" (empty / "bin").mkdir(parents=True) for command in (("actionlint",), ("shellcheck-orb",), ("make-bats",)): with self.subTest(command): self.assertFalse(self.verdict(command, tools=empty)) (self.tools / "bats" / "bin" / "bats").unlink() self.assertFalse(self.verdict(("make-bats",))) self.assertFalse((self.source / ".platform-ci" / "downloads").exists()) def test_the_resolver_refuses_tools_outside_the_image(self) -> None: resolver = HELPERS / "scripts" / "require-installed-tool.sh" environment = {**os.environ, "SOURCE_TOOLS_DIR": self.tools.as_posix()} destination = self.tools.parent / "dest" for tool in ("kubectl", "helm", "node", "pyyaml"): result = subprocess.run([str(BASH), resolver.as_posix(), tool, destination.as_posix()], env=environment, capture_output=True, text=True, check=False) # noqa: S603 self.assertNotEqual(result.returncode, 0, tool) self.assertIn("not part of the source-tools image", result.stderr) result = subprocess.run([str(BASH), resolver.as_posix(), "actionlint", destination.as_posix()], env=environment, capture_output=True, text=True, check=False) # noqa: S603 self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue((destination / "actionlint").exists()) def policy_runtime_available() -> bool: """Return whether this machine's bash has git and a python3 that imports PyYAML, which the unchanged validator needs.""" if BASH is None or shutil.which("git") is None: return False return subprocess.run([BASH, "-c", 'python3 -c "import yaml"'], capture_output=True, check=False).returncode == 0 # noqa: S603 @unittest.skipUnless(policy_runtime_available(), "needs bash, git and a python3 with PyYAML") class PolicySubcommandBehaviourTest(unittest.TestCase): """The real captured validator through the launcher on a real Git checkout: pass, and the legacy rejections stay rejections.""" @override def setUp(self) -> None: """Create a Git checkout of policy files with the platform helpers beside it, untracked, as the workflow has them.""" directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) self.source = Path(directory.name) / "checkout" shutil.copytree(HELPERS, self.source / HELPER_DIRECTORY) (self.source / "docs").mkdir() (self.source / "README.md").write_text("# Policy\n", encoding="utf-8") (self.source / "docs" / "settings.yaml").write_text("a: 1\n---\nb: 2\n", encoding="utf-8") (self.source / "CODEOWNERS").write_text("* @team\n", encoding="utf-8") self.git("init", "--quiet") self.track() def git(self, *arguments: str) -> None: """Run Git in the checkout.""" subprocess.run(["git", *arguments], cwd=self.source, check=True, capture_output=True) # noqa: S603, S607 def track(self) -> None: """Track every file except the platform helpers, which the real checkout keeps as an untracked nested checkout.""" self.git("add", "--all", "--", ".", ":(exclude).platform-ci") def passes(self) -> bool: """Plan the `policy` command and run it with the real helpers.""" target = planned_target(("policy",)) parent = {**os.environ, "HOME": str(self.source.parent)} return run_target_commands(target, self.source, run_with_bash, parent_environment=parent) def test_policy_and_documentation_files_pass(self) -> None: self.assertTrue(self.passes()) def test_application_manifests_unsupported_files_and_broken_documents_fail(self) -> None: for name, text in {"package.json": "{}\n", "app.py": "x = 1\n", "docs/bad.json": "{\n", "docs/bad.yaml": "a: [\n"}.items(): with self.subTest(name): path = self.source / name path.write_text(text, encoding="utf-8") self.track() self.assertFalse(self.passes()) self.git("rm", "--cached", "--quiet", "--", name) path.unlink() self.assertTrue(self.passes()) def re_assignment(line: str) -> bool: """Return whether a `tools.env` line is a plain KEY=value assignment.""" return bool(line) and not line.startswith("#") and "=" in line and line.split("=", 1)[0].isidentifier() class RenderedManifestRetentionTest(unittest.TestCase): """`kustomize-render` writes `ci-manifests/`; the existing generic artifact paths already retain it in the verified bundle.""" def collect(self, source: Path, *, succeeded: bool = True) -> dict[str, object]: """Plan the manifests example and package it as the workflow's package step does.""" (target,) = plan_for(parse_project_configuration(example_mapping("source-manifests.ci.yaml")), event(), load_runtime_catalog())["targets"] return collect_outputs(target, source, source.parent / "bundle", RunIdentity(SOURCE_REVISION, PLATFORM_REVISION, "7", "1", SOURCE_TREE), build_succeeded=succeeded) def test_nested_rendered_files_are_collected_with_size_and_digest(self) -> None: with tempfile.TemporaryDirectory() as directory: source = Path(directory) / "checkout" (source / "ci-manifests" / "prod").mkdir(parents=True) (source / "ci-manifests" / "prod" / "all.yaml").write_text("kind: ConfigMap\n", encoding="utf-8") (source / "ci-manifests" / "staging.yaml").write_text("kind: Service\n", encoding="utf-8") manifest = self.collect(source) files = cast("list[dict[str, object]]", manifest["files"]) self.assertEqual([entry["path"] for entry in files], ["artifact/rendered-manifests/ci-manifests/prod/all.yaml", "artifact/rendered-manifests/ci-manifests/staging.yaml"]) self.assertTrue(all(entry["sha256"] and entry["size"] for entry in files)) def test_a_successful_build_that_rendered_nothing_is_an_error_and_a_failed_one_keeps_no_artifact(self) -> None: with tempfile.TemporaryDirectory() as directory: source = Path(directory) / "checkout" source.mkdir() with self.assertRaisesRegex(PackagingError, "matched no files"): self.collect(source) self.assertEqual(self.collect(source, succeeded=False)["files"], []) class ExampleFileTest(unittest.TestCase): """The shipped example is valid YAML that the converter can copy.""" def test_the_example_is_valid_yaml_with_a_single_source_tools_component(self) -> None: mapping = yaml.safe_load((EXAMPLES_DIRECTORY / "source-tools.ci.yaml").read_text(encoding="utf-8")) self.assertEqual(list(mapping["components"]), ["source-tools"])