#!/usr/bin/env python3 """Tests for the production topology, secrets, backup, and rollback runbook. A runbook is only worth reading if its commands are the ones the tools actually accept, so this suite parses every documented ``python3 scripts/...`` command out of the markdown and checks each flag against the real CLI help. It also fails if the document starts carrying a real secret, or if it keeps claiming the server holds no journal after ``server.mjs`` starts writing one. """ from __future__ import annotations import re import shlex import subprocess import sys import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] RUNBOOK = ROOT / "docs" / "PRODUCTION-RUNBOOK.md" SERVER = ROOT / "server.mjs" BACKUP = ROOT / "scripts" / "operator_backup.py" FENCE = re.compile(r"```bash\n(.*?)```", re.S) SECRET_ASSIGNMENT = re.compile( r"(?i)(?:token|secret|password|passwd|api[_-]?key|salt)\s*=\s*(?!<|\$|\"|\'|\.\.\.)[^\s`\"']{6,}" ) def text() -> str: return RUNBOOK.read_text(encoding="utf-8") def sections() -> dict[str, str]: body = text() headings = list(re.finditer(r"^## (.+)$", body, re.M)) out: dict[str, str] = {} for index, match in enumerate(headings): end = headings[index + 1].start() if index + 1 < len(headings) else len(body) out[match.group(1).strip().lower()] = body[match.end():end] return out def documented_commands() -> list[str]: commands: list[str] = [] for block in FENCE.findall(text()): for line in block.splitlines(): line = line.strip() if line.startswith("python3 scripts/"): commands.append(line) return commands def help_text(script: str, subcommand: str | None) -> str: argv = [sys.executable, script] + ([subcommand] if subcommand else []) + ["--help"] result = subprocess.run(argv, cwd=ROOT, capture_output=True, text=True) assert result.returncode == 0, f"{' '.join(argv)} failed: {result.stderr}" return result.stdout class ProductionRunbookTests(unittest.TestCase): def test_runbook_covers_topology_network_boundaries_secrets_backup_restore_and_rollback(self) -> None: self.assertTrue(RUNBOOK.is_file(), "docs/PRODUCTION-RUNBOOK.md must exist") titles = sections().keys() for required in ("topology", "network boundaries", "secrets", "backup", "restore", "rollback"): self.assertTrue( any(title.startswith(required) for title in titles), f"runbook is missing a '{required}' section; found {sorted(titles)}", ) def test_backup_section_names_what_is_never_archived_and_why(self) -> None: backup = next(body for title, body in sections().items() if title.startswith("backup")) self.assertIn("scripts/operator_backup.py", backup) for required in ("model weights", "transient inference", "secret", "tombstone"): self.assertIn(required, backup.lower(), f"backup section must name {required}") self.assertRegex(backup, r"manifest\.jsonl") def test_every_documented_script_command_is_accepted_by_the_real_cli(self) -> None: commands = documented_commands() self.assertGreaterEqual(len(commands), 4, f"expected documented commands, found {commands}") for command in commands: with self.subTest(command=command): tokens = shlex.split(command) script = tokens[1] rest = tokens[2:] subcommand = rest[0] if rest and not rest[0].startswith("-") else None combined = help_text(script, subcommand) + help_text(script, None) for token in rest: if token.startswith("--"): self.assertIn(token, combined, f"{script} does not accept {token}") for token in rest[1:]: if not token.startswith("--"): continue def test_documented_backup_commands_are_the_ones_the_tool_runs(self) -> None: commands = " ".join(documented_commands()) for flag in ("--source", "--archive", "--manifest", "--target"): self.assertIn(flag, commands, f"the runbook must show the real {flag} flag") self.assertIn("verify", commands) self.assertIn("restore", commands) def test_rollback_is_documented_as_a_tested_fixture_path_and_a_human_gated_live_drill(self) -> None: rollback = next(body for title, body in sections().items() if title.startswith("rollback")) self.assertIn("scripts/deploy_staging.py rollback", rollback) self.assertRegex(rollback.lower(), r"human[- ]gate") self.assertIn("#53", rollback) self.assertIn("tests/staging-deploy.test.py", rollback) # The live drill must be explicit about not being run by automation (wrapped lines are fine). self.assertRegex( rollback.lower(), r"(?s)not .{0,80}automatically|does not run .{0,80}automatically", "the runbook must say the live drill is not run automatically", ) def test_network_boundaries_match_how_the_server_and_worker_actually_bind(self) -> None: server = SERVER.read_text(encoding="utf-8") # The app's own default is not loopback, so the runbook must say the deployment sets it. self.assertRegex(server, r"HOST\|\|'0\.0\.0\.0'") staging_env = (ROOT / "deploy" / "timmy-staging.env.example").read_text(encoding="utf-8") self.assertIn("HOST=127.0.0.1", staging_env) worker = (ROOT / "deploy" / "worker" / "run_worker_container.sh").read_text(encoding="utf-8") self.assertIn("Refusing to publish the worker outside loopback", worker) boundaries = next(body for title, body in sections().items() if title.startswith("network boundaries")) self.assertIn("0.0.0.0", boundaries, "the runbook must name the app's real default bind address") self.assertIn("HOST=127.0.0.1", boundaries) self.assertIn("deploy/timmy-staging.env.example", boundaries) self.assertIn("loopback", boundaries.lower()) self.assertIn("8080", boundaries) def test_runbook_carries_no_literal_secret_values(self) -> None: findings = [match.group(0) for match in SECRET_ASSIGNMENT.finditer(text())] self.assertEqual(findings, [], f"runbook looks like it contains a real secret: {findings}") self.assertIn("mode-600", text()) self.assertIn("never commit", text().lower()) def test_runbook_claims_match_a_stateless_server(self) -> None: server = SERVER.read_text(encoding="utf-8") for call in ("writeFile", "appendFile", "createWriteStream", "mkdirSync"): self.assertNotIn(call, server, "server.mjs started persisting data; the runbook's data-lifetime claim must be updated") body = text().lower() self.assertIn("no server-side copy of the journal", body) self.assertIn("/api/healthz", body) # The health probe is allowed to report readiness, never configuration or secrets. health = re.search(r"'/api/healthz'[^\n]*", server) self.assertIsNotNone(health, "server.mjs must serve /api/healthz") health_text = health.group(0) if health else "" for forbidden in ("token", "secret", "env:", "password"): self.assertNotIn(forbidden, health_text.lower()) # The response may report readiness only, so a new field is a deliberate decision. for field in ("ok", "release", "commit", "visionEnabled", "agentEnabled"): self.assertIn(field, health_text) def test_backup_tool_is_gated_by_its_own_suite(self) -> None: package = (ROOT / "package.json").read_text(encoding="utf-8") self.assertIn("tests/backup-restore.test.py", package) self.assertIn("tests/production-runbook.test.py", package) self.assertTrue(BACKUP.is_file()) if __name__ == "__main__": unittest.main(verbosity=2)