platform-ci/tests/test_plan.py

"""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 <arguments>`, 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)