Menu
AkurAI-Build
publicLatest change 941c1e29871d5e7548906ba3b75ddd925cc9326e - ci: fail closed on PR branches, fix false rollback claims, drop PyYAML by Ólafur Búi Ólafsson
#!/usr/bin/env python3
"""Validates .akurai.yml branch policy: stdlib-only, no third-party imports.
Runnable with the site-packages import mechanism disabled entirely
(`python3 -S tests/test_pr_pipeline.py`) so a clean CI host with no PyYAML
installed still executes this regression check. It is wired into the
`verify` job in .akurai.yml.
Because pulling in PyYAML would defeat the -S requirement, this file parses
the small subset of YAML actually used by .akurai.yml (a top-level `jobs:`
list of flat string/list-of-string mappings) with a hand-rolled indentation
parser instead of a general YAML library. This is deliberately narrow: it is
not a YAML parser, only enough structure to read job name/needs/branches so
the test exercises the real checked-in pipeline file, not a hardcoded copy.
The branch-gate predicate mirrors src/runner.rs::execute_job bit for bit:
if !spec.branches.is_empty() && !spec.branches.iter().any(|allowed| allowed == branch) {
// job skipped
}
CURRENT POLICY UNDER TEST: verify, package, and deploy-production are all
branches: [main] only. There is no per-ref execution isolation in the
runner today (native jobs are gated on repository-wide trust, not per-ref),
so a `branches` allow-list cannot safely be relaxed for PR/feature refs
without other runner-side controller changes landing first (see the
PENDING INTEGRATION comment in .akurai.yml above the verify job). This test
asserts the fail-closed state, not a future desired state.
"""
import unittest
from pathlib import Path
PIPELINE_PATH = Path(__file__).resolve().parent.parent / ".akurai.yml"
def _strip_comment(line: str) -> str:
# No YAML strings in this file contain '#', so a plain split is safe here.
return line.split("#", 1)[0]
def load_jobs() -> dict:
"""Minimal indentation-based reader for .akurai.yml's `jobs:` list.
Handles exactly what .akurai.yml uses: `- name: x`, `key: value`,
`key: [a, b]`, and `key:` followed by an indented `- item` list.
"""
lines = [
_strip_comment(raw).rstrip()
for raw in PIPELINE_PATH.read_text().splitlines()
]
jobs = {}
current = None
list_key = None
for raw in lines:
if not raw.strip():
continue
indent = len(raw) - len(raw.lstrip(" "))
stripped = raw.strip()
if stripped.startswith("- name:") and indent <= 2:
name = stripped.split(":", 1)[1].strip()
current = {"name": name}
jobs[name] = current
list_key = None
continue
if current is None:
continue
if stripped.startswith("- ") and list_key:
current.setdefault(list_key, []).append(stripped[2:].strip())
continue
if ":" in stripped:
key, _, value = stripped.partition(":")
key = key.strip()
value = value.strip()
if value.startswith("[") and value.endswith("]"):
items = [v.strip() for v in value[1:-1].split(",") if v.strip()]
current[key] = items
list_key = None
elif value:
current[key] = value
list_key = None
else:
list_key = key
return jobs
def job_runs_on_branch(job: dict, branch: str) -> bool:
"""Mirror runner.rs execute_job's allow-list check bit for bit."""
branches = job.get("branches") or []
if branches and branch not in branches:
return False
return True
class PipelineBranchPolicy(unittest.TestCase):
def setUp(self):
self.jobs = load_jobs()
def test_pipeline_job_set(self):
self.assertEqual(
{"verify", "package", "deploy-production"}, set(self.jobs)
)
def test_no_job_runs_on_pr_branches_until_isolation_lands(self):
# Fail-closed state: no per-ref execution isolation exists in the
# runner yet (native jobs are gated repository-wide, not per-ref),
# so every job stays branches: [main] until that lands. See the
# PENDING INTEGRATION comment in .akurai.yml.
for job_name in ("verify", "package", "deploy-production"):
job = self.jobs[job_name]
for branch in ("feature/pr-ci-safety", "feature/anything", "fix/1"):
self.assertFalse(
job_runs_on_branch(job, branch),
f"{job_name} must not run on untrusted branch {branch} "
"until per-ref execution isolation exists",
)
def test_main_still_runs_everything(self):
for job_name in ("verify", "package", "deploy-production"):
self.assertTrue(
job_runs_on_branch(self.jobs[job_name], "main"),
f"{job_name} must still run on main",
)
def test_package_requires_verify(self):
self.assertIn("verify", self.jobs["package"].get("needs", []))
def test_deploy_requires_package(self):
self.assertIn("package", self.jobs["deploy-production"].get("needs", []))
def test_all_jobs_branches_are_main_only(self):
# Explicit assertion of the current fail-closed allow-list, so a
# future accidental widening (e.g. dropping `branches` again) fails
# this test instead of silently reopening the trust-boundary gap
# from the rejected c473d59 revision.
for job_name in ("verify", "package", "deploy-production"):
self.assertEqual(
["main"],
self.jobs[job_name].get("branches"),
f"{job_name} branches allow-list must stay ['main']",
)
if __name__ == "__main__":
unittest.main()