Menu
AkurAI-Build
publicLatest change 266a1aa21e346770b48e96f008ab6c2a5a603961 - Complete PR interfaces, isolated CI and crash-safe merge recovery 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: native verify, package and deployment remain
main-only. verify-pr uses a pinned Docker image without host caches or secrets
and runs on feature/merge refs. Per-ref trust is enforced in runner.rs.
"""
import json
import os
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
PIPELINE_PATH = REPO_ROOT / ".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-pr", "verify", "package", "deploy-production"}, set(self.jobs)
)
def test_native_and_deployment_jobs_do_not_run_on_pr_refs(self):
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_pr_checks_use_immutable_isolation_without_shared_caches(self):
job = self.jobs["verify-pr"]
self.assertEqual(job["executor"], "docker")
self.assertRegex(job["image"], r"@sha256:[0-9a-f]{64}$")
self.assertFalse(job.get("cache"))
self.assertFalse(job.get("secrets"))
self.assertFalse(job.get("environment"))
for branch in ("feature/work", "fix/work", "refs/merge-queue/1"):
self.assertTrue(job_runs_on_branch(job, branch))
self.assertIn("verify-pr", self.jobs["package"]["needs"])
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']",
)
class BinaryRollbackSchemaCompatibility(unittest.TestCase):
"""Real, disposable-data evidence for docs/pr-queue-recovery.md section 1.
Exercises actual `akurai migrate` binaries (not a description of the
schema) against a throwaway SQLite database to answer: does an older
binary tolerate a newer, additively-migrated schema at runtime, the way
the recovery doc's binary-rollback claim depends on?
Needs a pre-built OLD `akurai` binary from a version prior to the
current schema (this repo has migrations 001-010 checked in; any tag
built before the schema version its own binary code expects will do).
Skips itself, does not fail, if that binary is not supplied — building
two full release binaries in a plain `python3 tests/...` invocation
would make every unrelated test run slow and network/toolchain
dependent, so this is opt-in evidence, not part of the default `verify`
run. It IS runnable on demand exactly as documented below and was run
for real to produce the evidence quoted in docs/pr-queue-recovery.md.
Usage:
AKURAI_OLD_BINARY=/path/to/old/target/release/akurai \\
AKURAI_NEW_BINARY=/path/to/new/target/release/akurai \\
python3 tests/test_pr_pipeline.py -v BinaryRollbackSchemaCompatibility
"""
def setUp(self):
self.old_binary = os.environ.get("AKURAI_OLD_BINARY")
self.new_binary = os.environ.get("AKURAI_NEW_BINARY")
if not self.old_binary or not self.new_binary:
self.skipTest(
"set AKURAI_OLD_BINARY and AKURAI_NEW_BINARY to run this "
"real binary/schema compatibility check (see class docstring)"
)
for path in (self.old_binary, self.new_binary):
if not os.access(path, os.X_OK):
self.skipTest(f"{path} is not an executable file")
def _run(self, binary, data_dir, *args):
env = dict(os.environ)
env["AKURAI_DB_KEY"] = (data_dir / "db.key").read_text().strip()
result = subprocess.run(
[binary, "--data", str(data_dir), *args],
capture_output=True,
text=True,
env=env,
timeout=30,
)
return result
def test_old_binary_fails_closed_against_newer_schema(self):
# Uses a real repo-local gitignored .tmp scratch dir per the repo
# scratch policy, not shared /tmp, and cleans up after itself.
scratch_root = REPO_ROOT / ".tmp" / "tmp"
scratch_root.mkdir(parents=True, exist_ok=True)
data_dir = Path(tempfile.mkdtemp(prefix="rollback-compat-", dir=scratch_root))
try:
keygen = subprocess.run(
[self.old_binary, "--data", str(data_dir), "keygen", "--out", str(data_dir / "db.key")],
capture_output=True,
text=True,
timeout=30,
)
self.assertEqual(0, keygen.returncode, keygen.stderr)
old_migrate = self._run(self.old_binary, data_dir, "migrate")
self.assertEqual(0, old_migrate.returncode, old_migrate.stderr)
old_applied = json.loads(old_migrate.stdout)["data"]["migrations_applied"]
self.assertGreater(old_applied, 0, "old binary should apply its own schema fresh")
new_migrate = self._run(self.new_binary, data_dir, "migrate")
self.assertEqual(
0, new_migrate.returncode, f"new binary must upgrade cleanly: {new_migrate.stderr}"
)
# This is the actual rollback scenario: an operator reinstalls
# the old binary after a bad deploy, but the schema has already
# been upgraded by the new binary and was never rolled back
# (there is no down-migration; see doc section 3).
old_after_upgrade = self._run(self.old_binary, data_dir, "migrate")
self.assertNotEqual(
0,
old_after_upgrade.returncode,
"old binary unexpectedly succeeded against an upgraded schema; "
"if this starts passing, the compatibility claim in "
"docs/pr-queue-recovery.md changed and must be revisited",
)
payload = json.loads(old_after_upgrade.stdout)
self.assertIn("schema", payload["error"]["message"])
# And confirm the failure was clean: the new binary still reads
# the same disposable database fine afterward (old binary's
# rejection did not corrupt or partially rewrite anything).
new_after_old_failure = self._run(self.new_binary, data_dir, "migrate")
self.assertEqual(0, new_after_old_failure.returncode, new_after_old_failure.stderr)
finally:
shutil.rmtree(data_dir, ignore_errors=True)
if __name__ == "__main__":
unittest.main()