timmy-talking-turd/tests/production-runbook.test.py

#!/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)