platform-ci/tests/test_pipeline_workflow.py

"""The shared workflow wires prepare, verify, secrets and report as designed; static checks only, nothing is run."""

import re
import unittest
from pathlib import Path
from typing import cast

import yaml

WORKFLOW_TEXT = (Path(__file__).resolve().parent.parent / ".gitea" / "scoped_workflows" / "pipeline.yaml").read_text(encoding="utf-8")
WORKFLOW = cast("dict[str, object]", yaml.safe_load(WORKFLOW_TEXT))
JOBS = cast("dict[str, dict[str, object]]", WORKFLOW["jobs"])
UPLOAD_ACTION = "self:KofTwentyTwo/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02"


def _steps(job: str) -> list[dict[str, object]]:
   return cast("list[dict[str, object]]", JOBS[job]["steps"])


def _step(job: str, name_prefix: str) -> dict[str, object]:
   return next(step for step in _steps(job) if str(step.get("name", "")).startswith(name_prefix))


class WorkflowTest(unittest.TestCase):
   """Security work is once per run, conditional on verification, and required by the final report."""

   def test_secrets_job_runs_once_when_verification_is_required(self) -> None:
      job = JOBS["secrets"]
      self.assertEqual(job["needs"], ["prepare"])
      self.assertEqual(job["if"], "${{ needs.prepare.outputs.verify == 'true' }}")
      self.assertNotIn("strategy", job)

   def test_security_image_is_the_qualified_immutable_runtime(self) -> None:
      container = cast("dict[str, str]", JOBS["secrets"]["container"])
      self.assertEqual(container["image"], "git.kof22.com/koftwentytwo/platform-ci-security@sha256:7b4707350ab57512ca0612fb616f4d1de769c740e7aa887e9f0c78b734d4bcd5")
      self.assertNotIn("ROOT_SECURITY_IMAGE_REQUIRED", WORKFLOW_TEXT)

   def test_source_checkout_is_exact_and_complete(self) -> None:
      checkout = cast("dict[str, object]", _steps("secrets")[0]["with"])
      self.assertEqual(checkout["ref"], "${{ needs.prepare.outputs.source_revision }}")
      self.assertEqual(checkout["fetch-depth"], 0)

   def test_scan_uses_the_cli_with_run_facts_and_no_repository_lookup(self) -> None:
      scan = _step("secrets", "Scan")
      self.assertTrue(str(scan["run"]).startswith("python3 -P -m platform_ci scan --source . "))
      environment = cast("dict[str, str]", scan["env"])
      self.assertEqual(sorted(environment), ["CI_PLATFORM_REVISION", "CI_RUN_ATTEMPT", "CI_RUN_ID", "CI_SOURCE_REVISION"])
      self.assertNotIn("curl", WORKFLOW_TEXT)
      self.assertNotIn("wget", WORKFLOW_TEXT)

   def test_artifact_guard_and_pinned_upload_apply_to_security_too(self) -> None:
      verify_guard = _step("verify", "Require private artifact service")["run"]
      self.assertEqual(_step("secrets", "Require private artifact service")["run"], verify_guard)
      upload = _step("secrets", "Retain")
      self.assertEqual(upload["uses"], UPLOAD_ACTION)
      self.assertEqual(_step("verify", "Retain")["uses"], UPLOAD_ACTION)
      self.assertEqual(cast("dict[str, object]", upload["with"])["path"], "security-output/")
      # the artifact is the redacted manifest directory; the raw Gitleaks report is never named as an upload path
      self.assertNotIn("gitleaks.json", str(upload["with"]))

   def test_report_needs_all_four_jobs_and_receives_each_result(self) -> None:
      report = JOBS["report"]
      self.assertEqual(report["needs"], ["prepare", "verify", "verify-pool", "verify-pool-docker", "verify-native", "secrets"])
      self.assertEqual(report["if"], "${{ always() }}")
      environment = cast("dict[str, str]", _step("report", "Report")["env"])
      self.assertEqual(environment["CI_SECRETS_RESULT"], "${{ needs.secrets.result }}")
      self.assertEqual(environment["CI_VERIFY_RESULT"], "${{ needs.verify.result }}")
      self.assertEqual(environment["CI_POOL_VERIFY_RESULT"], "${{ needs.verify-pool.result }}")
      self.assertEqual(environment["CI_DOCKER_POOL_VERIFY_RESULT"], "${{ needs.verify-pool-docker.result }}")
      self.assertEqual(environment["CI_NATIVE_VERIFY_RESULT"], "${{ needs.verify-native.result }}")
      self.assertEqual(environment["CI_PREPARE_RESULT"], "${{ needs.prepare.result }}")

   def test_fork_guard_helper_pin_and_no_repository_identity_selection_remain(self) -> None:
      self.assertIn("Reject fork pull requests", str(_steps("prepare")[0]["name"]))
      self.assertIn("PLATFORM_CI_REVISION: unpinned", WORKFLOW_TEXT)
      for context in re.findall(r"\$\{\{[^}]*\}\}", WORKFLOW_TEXT):
         self.assertNotRegex(context, r"repository_owner|github\.actor")
         if "github.repository" in context:
            self.assertEqual(context, "${{ github.repository }}")  # the reuse client's authorization identity only (see test_plan)


if __name__ == "__main__":
   unittest.main()