Menu
AkurAI-Build
publicLatest change d9b2b1e262d3265c1661d308de806d70a84d802a - test+docs: real disposable-data evidence for binary/schema rollback 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 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", "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']",
)
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()