"""Plan preparation: Maven and Node plans, one verification per event, and no repository-aware behavior.""" import json import re import shutil import subprocess import tempfile import unittest from dataclasses import replace from pathlib import Path from typing import cast import yaml from platform_ci.__main__ import main from platform_ci.configuration import load_project_configuration, parse_project_configuration from platform_ci.plan import build_matrix, to_compact_json from platform_ci.workspace import PipelineError from tests.support import ( EXAMPLES_DIRECTORY, JAVA_IMAGE, NODE_IMAGE, PLATFORM_REVISION, SOURCE_REVISION, TEST_RUNTIME_CATALOG, build_mapping, event, example_configuration, example_mapping, first_target, plan_for, ) PULL_REQUEST = event("pull_request", "refs/pull/7/merge") PUSH_TO_DEFAULT_BRANCH = event("push", "refs/heads/main") PUSH_TO_RELEASE_BRANCH = event("push", "refs/heads/develop") PUSH_TO_FEATURE_BRANCH = event("push", "refs/heads/feature-x") PUSH_TO_TAG = event("push", "refs/tags/v1.0.0") WORKFLOW_PATH = Path(__file__).resolve().parent.parent / ".gitea" / "scoped_workflows" / "pipeline.yaml" class PlanPreparationTest(unittest.TestCase): """The plan is deterministic data derived only from configuration and event.""" def test_maven_plan(self) -> None: """Maven runs the declared arguments in the toolchain image.""" target = plan_for(example_configuration("maven.ci.yaml"), PULL_REQUEST)["targets"][0] self.assertEqual(target["name"], "library-linux-amd64") self.assertEqual(target["display"], "library / Linux / Java 21") self.assertEqual(target["image"], JAVA_IMAGE) self.assertEqual(target["commands"], [["mvn", "--batch-mode", "verify"]]) def test_node_plan(self) -> None: """Node installs from the lock file, then runs the declared arguments.""" target = plan_for(example_configuration("node.ci.yaml"), PULL_REQUEST)["targets"][0] self.assertEqual(target["display"], "website / Linux / Node 22.20.0") self.assertEqual(target["image"], NODE_IMAGE) self.assertEqual(target["commands"], [["npm", "ci"], ["npm", "run", "verify"]]) def test_node_commands_install_once_then_run_each_configured_command(self) -> None: """`npm ci` appears exactly once, first; each configured command follows once, in order, prefixed with npm.""" commands = first_target("node-commands.ci.yaml")["commands"] self.assertEqual(commands, [["npm", "ci"], ["npm", "run", "lint"], ["npm", "test"], ["npm", "run", "build"]]) self.assertEqual(commands.count(["npm", "ci"]), 1) def test_maven_commands_each_get_the_maven_executable(self) -> None: """Several Maven commands each run once as `mvn `, with no install step.""" mapping = example_mapping("maven.ci.yaml") build = build_mapping(mapping, "library") del build["arguments"] build["commands"] = [["clean"], ["--batch-mode", "verify"]] commands = plan_for(parse_project_configuration(mapping), PULL_REQUEST)["targets"][0]["commands"] self.assertEqual(commands, [["mvn", "clean"], ["mvn", "--batch-mode", "verify"]]) def test_report_json_contract(self) -> None: """Legacy and typed reports both reach the plan JSON as `format`, `paths` and `required`, ready for per-format checks.""" legacy = first_target("maven.ci.yaml")["reports"] self.assertEqual(legacy, {"junit": {"format": "junit", "paths": ["**/target/surefire-reports/*.xml"], "required": True}}) target = first_target("maven-reports.ci.yaml") payload = json.loads(to_compact_json(target)) self.assertEqual( payload["reports"], { "coverage": {"format": "file", "paths": ["target/site/jacoco/jacoco.xml"], "required": True}, "junit": {"format": "junit", "paths": ["**/target/surefire-reports/*.xml", "**/target/failsafe-reports/*.xml"], "required": False}, }, ) def test_plan_is_deterministic(self) -> None: """The same inputs give byte-identical output, and targets are ordered by component name.""" configuration = example_configuration("maven.ci.yaml") first = to_compact_json(plan_for(configuration, PULL_REQUEST)) self.assertEqual(first, to_compact_json(plan_for(configuration, PULL_REQUEST))) self.assertEqual(json.loads(first)["source_revision"], SOURCE_REVISION) self.assertEqual(json.loads(first)["platform_revision"], PLATFORM_REVISION) def test_requested_events_verify_once(self) -> None: """A pull request and selected, default, release and tag pushes each require one verification with one matrix entry.""" configuration = example_configuration("maven.ci.yaml") contexts = { "pull request": PULL_REQUEST, "default branch": PUSH_TO_DEFAULT_BRANCH, "release branch": PUSH_TO_RELEASE_BRANCH, "selected branch": event("push", "refs/heads/staging"), "selected tag": PUSH_TO_TAG, "manual call": event("workflow_dispatch", "refs/heads/feature-x"), } for description, context in contexts.items(): with self.subTest(description): plan = plan_for(configuration, context) self.assertTrue(plan["verification"]["required"]) self.assertEqual(len(build_matrix(plan)["include"]), 1) def test_unselected_feature_push_is_not_requested_and_matrix_is_not_empty(self) -> None: """A feature push requests no verification, yet the matrix still parses and the reason names the configuration.""" plan = plan_for(example_configuration("maven.ci.yaml"), PUSH_TO_FEATURE_BRANCH) self.assertFalse(plan["verification"]["required"]) self.assertIn("does not select", plan["verification"]["reason"]) self.assertEqual(len(build_matrix(plan)["include"]), 1) def test_default_branch_comes_from_event_metadata(self) -> None: """A configuration that lists no branches still verifies whatever branch the event says is the default.""" configuration = example_configuration("node.ci.yaml") trunk = plan_for(configuration, event("push", "refs/heads/trunk", default_branch="trunk")) self.assertTrue(trunk["verification"]["required"]) self.assertEqual(trunk["verification"]["reason"], "push to the default branch") self.assertFalse(plan_for(configuration, event("push", "refs/heads/feature-x", default_branch="trunk"))["verification"]["required"]) def test_release_eligible_push_verifies_even_when_push_selection_is_empty(self) -> None: """Release eligibility forces verification regardless of `verification.push`.""" mapping = example_mapping("maven.ci.yaml") mapping["verification"] = {"push": {"branches": [], "tags": []}} plan = plan_for(parse_project_configuration(mapping), PUSH_TO_RELEASE_BRANCH) self.assertTrue(plan["verification"]["required"]) self.assertIn("release-eligible", plan["verification"]["reason"]) def test_tags_obey_configuration(self) -> None: """A tag verifies only when a pattern matches; omitting the setting verifies every tag.""" configuration = example_configuration("maven.ci.yaml") self.assertTrue(plan_for(configuration, event("push", "refs/tags/v2.1.0"))["verification"]["required"]) self.assertFalse(plan_for(configuration, event("push", "refs/tags/nightly"))["verification"]["required"]) mapping = example_mapping("maven.ci.yaml") del mapping["verification"] self.assertTrue(plan_for(parse_project_configuration(mapping), event("push", "refs/tags/nightly"))["verification"]["required"]) def test_missing_metadata_and_unknown_refs_fail(self) -> None: """A branch push without default branch metadata, or a push to neither branch nor tag, cannot be decided.""" configuration = example_configuration("maven.ci.yaml") with self.assertRaisesRegex(PipelineError, "default branch metadata"): plan_for(configuration, event("push", "refs/heads/feature-x", default_branch="")) with self.assertRaisesRegex(PipelineError, "neither a branch nor a tag"): plan_for(configuration, event("push", "refs/notes/x")) self.assertTrue(plan_for(configuration, event("pull_request", "refs/pull/1/merge", default_branch=""))["verification"]["required"]) def test_release_eligibility_is_data_only(self) -> None: """An eligible push has the same targets as an ineligible one and states that delivery is not implemented.""" configuration = example_configuration("maven.ci.yaml") eligible = plan_for(configuration, PUSH_TO_RELEASE_BRANCH) ineligible = plan_for(configuration, PUSH_TO_DEFAULT_BRANCH) self.assertEqual(eligible["targets"], ineligible["targets"]) self.assertEqual(eligible["delivery"]["eligible_channels"], ["snapshot"]) self.assertEqual(ineligible["delivery"]["eligible_channels"], []) self.assertFalse(eligible["delivery"]["implemented"]) def test_revisions_must_be_full_hashes(self) -> None: """The unpinned platform placeholder cannot produce a plan.""" with self.assertRaises(PipelineError): plan_for(example_configuration("maven.ci.yaml"), replace(event(), platform_revision="unpinned")) class RepositoryIndependenceTest(unittest.TestCase): """Repository identity cannot change behavior because it never reaches the implementation.""" def test_identical_configuration_gives_identical_plan_for_unrelated_repositories(self) -> None: """Two checkouts with unrelated names and the same `.ci.yaml` produce the same plan and workflow outputs.""" configuration_text = (EXAMPLES_DIRECTORY / "maven.ci.yaml").read_text(encoding="utf-8") results: list[tuple[str, str]] = [] for repository_name in ("alpha-service", "zeta-unrelated-tool"): with self.subTest(repository_name), tempfile.TemporaryDirectory() as parent: checkout = Path(parent) / repository_name checkout.mkdir() (checkout / ".ci.yaml").write_text(configuration_text, encoding="utf-8") for context in (PUSH_TO_RELEASE_BRANCH, PUSH_TO_FEATURE_BRANCH): plan = plan_for(load_project_configuration(checkout), context) results.append((to_compact_json(plan), to_compact_json(build_matrix(plan)))) self.assertEqual(results[:2], results[2:]) def test_shared_implementation_never_reads_repository_identity(self) -> None: """No shared file mentions the repository or owner context a name-based branch would need.""" root = Path(__file__).resolve().parent.parent shared_files = [*(root / "platform_ci").glob("*.py"), WORKFLOW_PATH] for path in shared_files: text = path.read_text(encoding="utf-8") for forbidden in ("GITHUB_REPOSITORY", "repository_owner", "REPOSITORY_NAME"): self.assertNotIn(forbidden, text, f"{path.name} must not branch on repository identity") if path != WORKFLOW_PATH: self.assertNotIn("github.repository", text, f"{path.name} must not branch on repository identity") # The identity is authorization and provenance only (reuse scope, dependency checkout, version finalization, release tag): it reaches # steps solely as the `CI_REPOSITORY` env var, and only these modules read it. No condition, matrix or runner label may use it. workflow_text = WORKFLOW_PATH.read_text(encoding="utf-8") env_uses = re.findall(r"^ +CI_REPOSITORY: \$\{\{ github\.repository \}\}$", workflow_text, flags=re.MULTILINE) all_uses = re.findall(r"\$\{\{[^}]*github\.repository[^}]*\}\}", workflow_text) self.assertEqual(len(env_uses), len(all_uses), "github.repository may appear only as a CI_REPOSITORY step env value") # Six consumers: dependency checkout in each of the three container verify jobs, the `reuse` step, the release-tag step, and the prepare `plan` step. The prepare `plan` # step is the SIXTH: `__main__._release_activity_reader` scopes the job's read-only forge client to this run's own repository (with CI_REPOSITORY_ID and CI_RUN_ID) to read the # server's start time of this run's prepare job and the commit history of the source revision for a develop SNAPSHOT's effective version. It is the same authorization/provenance use as # the `reuse` step (a `RepositoryScope`), never a branch: the check below keeps every use out of conditions and routing. self.assertEqual(len(all_uses), 6, "the three dependency checkouts, reuse, release-tag and the prepare plan step are the only consumers") (prepare_job_text,) = re.findall(r"(?ms)^ prepare:\n.*?^ verify:\n", workflow_text) self.assertIn("CI_REPOSITORY: ${{ github.repository }}", prepare_job_text) main_text = (root / "platform_ci" / "__main__.py").read_text(encoding="utf-8") reader_text = main_text[main_text.index("def _release_activity_reader") : main_text.index("def _optional_tree")] self.assertIn("RepositoryScope(repository,", reader_text) for routing in ("repository ==", "repository !=", "repository in", "startswith(", "split("): self.assertNotIn(routing, reader_text, "the repository identity is only the forge client's scope, never compared or parsed to choose behavior") for line in workflow_text.splitlines(): if re.match(r"\s*(if|runs-on):", line) or "matrix" in line: self.assertNotIn("github.repository", line, "no condition, matrix or runner label may use the repository identity") readers = sorted(path.relative_to(root / "platform_ci").as_posix() for path in (root / "platform_ci").rglob("*.py") if "CI_REPOSITORY" in path.read_text(encoding="utf-8")) self.assertEqual(readers, ["__main__.py", "delivery/finalize_version.py", "delivery/release_tag.py", "dependency_checkout.py"]) class WorkflowLocationTest(unittest.TestCase): """There is exactly one workflow, in the scoped-workflow directory, and it keeps a callable entry.""" def test_only_the_scoped_workflow_exists(self) -> None: """The superseded top-level workflow path must not coexist with the scoped one.""" workflows_directory = WORKFLOW_PATH.parents[1] / "workflows" self.assertTrue(WORKFLOW_PATH.is_file()) self.assertEqual(list(workflows_directory.glob("*")) if workflows_directory.exists() else [], []) self.assertEqual(sorted(path.name for path in WORKFLOW_PATH.parent.glob("*")), ["pipeline.yaml"]) def test_scoped_workflow_keeps_a_callable_entry(self) -> None: """The same file is triggered by push and pull request and callable for trigger-only callers.""" triggers = cast("dict[object, object]", yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8"))[True]) self.assertEqual(set(triggers), {"push", "pull_request", "workflow_dispatch", "workflow_call"}) class ForkGuardTest(unittest.TestCase): """A fork pull request is rejected by the first step of prepare, before any source is checked out.""" def _guard(self) -> dict[str, object]: """Return prepare's first step, after proving it precedes every checkout.""" steps = cast("list[dict[str, object]]", yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8"))["jobs"]["prepare"]["steps"]) self.assertNotIn("uses", steps[0]) self.assertEqual(steps[1].get("uses"), "builtin:checkout") return steps[0] def test_guard_compares_event_repositories_and_never_names_one(self) -> None: """The head and base repositories come from the event through the environment, not from names in the workflow.""" guard = self._guard() self.assertEqual(guard["if"], "${{ github.event_name == 'pull_request' }}") self.assertEqual(guard["env"], {"HEAD_REPOSITORY": "${{ github.event.pull_request.head.repo.full_name }}", "BASE_REPOSITORY": "${{ github.event.pull_request.base.repo.full_name }}"}) self.assertNotIn("${{", cast("str", guard["run"])) @unittest.skipIf(shutil.which("sh") is None, "needs a POSIX shell") def test_guard_script_fails_for_a_fork_or_unknown_head_and_passes_for_the_same_repository(self) -> None: """The script's exit code is the verdict: nonzero for a fork or a missing head, zero for the same repository.""" script = cast("str", self._guard()["run"]) for head, base, expected in (("owner/repo", "owner/repo", 0), ("fork-owner/repo", "owner/repo", 1), ("", "owner/repo", 1)): with self.subTest(head=head): result = subprocess.run(["sh", "-c", script], env={"HEAD_REPOSITORY": head, "BASE_REPOSITORY": base}, capture_output=True, check=False) # noqa: S603, S607 self.assertEqual(result.returncode, expected) class PrepareCommandTest(unittest.TestCase): """The prepare command publishes the plan as workflow outputs.""" def test_prepare_writes_outputs_and_rejects_invalid_configuration(self) -> None: """Valid configuration yields verify, matrix and plan outputs; invalid configuration exits nonzero.""" with tempfile.TemporaryDirectory() as directory: checkout = Path(directory) outputs_file = checkout / "outputs" environment = { "CI_EVENT_NAME": "push", "CI_EVENT_REF": "refs/heads/feature-x", "CI_DEFAULT_BRANCH": "main", "CI_SOURCE_REVISION": SOURCE_REVISION, "CI_PLATFORM_REVISION": PLATFORM_REVISION, "GITHUB_OUTPUT": str(outputs_file), } (checkout / ".ci.yaml").write_text((EXAMPLES_DIRECTORY / "node.ci.yaml").read_text(encoding="utf-8"), encoding="utf-8") self.assertEqual(main(["prepare", "--source", str(checkout)], environment, TEST_RUNTIME_CATALOG), 0) outputs = dict(line.split("=", 1) for line in outputs_file.read_text(encoding="utf-8").splitlines()) self.assertEqual(outputs["verify"], "false") self.assertEqual(len(json.loads(outputs["matrix"])["include"]), 1) self.assertEqual(json.loads(json.loads(outputs["matrix"])["include"][0]["target"])["build_tool"], "node") (checkout / ".ci.yaml").write_text("schema: 1\ncomponents: {}\n", encoding="utf-8") self.assertEqual(main(["prepare", "--source", str(checkout)], environment, TEST_RUNTIME_CATALOG), 1)