jesse-uxandGrok (agent loop)
5c0d895391
Product chose to keep the scheduled backup, but the current recovery point must stay the one created by hand. Daily runs still encrypt and restore-verify, then record a separate latest-backup note. Co-Authored-By: Grok (agent loop)
841 lines
32 KiB
Python
841 lines
32 KiB
Python
"""Contracts for the closed-beta safety net. Fixtures are fictional."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
WORKFLOW = ROOT / ".gitea" / "workflows" / "backup-production-daily.yml"
|
|
RECOVERY_WORKFLOW = ROOT / ".gitea" / "workflows" / "create-production-recovery.yml"
|
|
RECOVERY_SCRIPT = ROOT / "deploy" / "run-production-recovery.sh"
|
|
PRUNE_SCRIPT = ROOT / "deploy" / "prune-production-daily-backups.sh"
|
|
HEALTH_SCRIPT = ROOT / "deploy" / "monitor" / "healthcheck-alert.sh"
|
|
DIGEST_SCRIPT = ROOT / "deploy" / "monitor" / "error-digest.sh"
|
|
FIND_SCRIPT = ROOT / "deploy" / "find-request-log.sh"
|
|
NOTIFY_SCRIPT = ROOT / "deploy" / "monitor" / "notify-feishu.sh"
|
|
SERVER_COMPOSE = ROOT / "deploy" / "docker-compose.server.yml"
|
|
POSTGRES_COMPOSE = ROOT / "deploy" / "docker-compose.postgres.yml"
|
|
|
|
|
|
def resolve_bash() -> str:
|
|
found = shutil.which("bash")
|
|
git_bash = Path(r"D:\Program Files\Git\bin\bash.exe")
|
|
if os.name == "nt" and git_bash.exists():
|
|
lowered = (found or "").lower()
|
|
if not found or "system32" in lowered or "windowsapps" in lowered:
|
|
return str(git_bash)
|
|
if found:
|
|
return found
|
|
if git_bash.exists():
|
|
return str(git_bash)
|
|
return "bash"
|
|
|
|
|
|
BASH = resolve_bash()
|
|
|
|
|
|
def bash_path(path: Path) -> str:
|
|
if os.name != "nt":
|
|
return str(path)
|
|
completed = subprocess.run(
|
|
[BASH, "-lc", 'cygpath -u "$1"', "cygpath", str(path)],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
return completed.stdout.strip()
|
|
|
|
|
|
def run_bash(script: Path, extra: dict[str, str], args: list[str] | None = None) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env.pop("FEISHU_WEBHOOK_URL", None)
|
|
env.pop("FEISHU_SIGN_SECRET", None)
|
|
env.update(extra)
|
|
env["JYOTISHA_PYTHON"] = bash_path(Path(sys.executable))
|
|
return subprocess.run(
|
|
[BASH, bash_path(script), *(args or [])],
|
|
cwd=ROOT,
|
|
env=env,
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
|
|
|
|
def write_text(path: Path, text: str) -> None:
|
|
path.write_text(text, encoding="utf-8", newline="\n")
|
|
|
|
|
|
class HealthHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None:
|
|
if self.path != "/health":
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
return
|
|
kind = getattr(self.server, "kind", "ok")
|
|
if kind == "timeout":
|
|
time.sleep(3)
|
|
commit = getattr(self.server, "commit", "a" * 40)
|
|
if kind == "500":
|
|
status, body = 500, b'{"status":"ok"}'
|
|
elif kind == "missing":
|
|
status, body = 200, b"{}"
|
|
elif kind == "degraded":
|
|
status, body = 200, json.dumps({"status": "degraded", "deployment": {"gitCommit": commit}}).encode()
|
|
else:
|
|
status, body = 200, json.dumps({"status": "ok", "deployment": {"gitCommit": commit}}).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, fmt: str, *args: object) -> None:
|
|
return
|
|
|
|
|
|
class HookHandler(BaseHTTPRequestHandler):
|
|
def do_POST(self) -> None:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
body = self.rfile.read(length)
|
|
self.server.bodies.append(body)
|
|
self.close_connection = True
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", "2")
|
|
self.send_header("Connection", "close")
|
|
self.end_headers()
|
|
self.wfile.write(b"ok")
|
|
|
|
def log_message(self, fmt: str, *args: object) -> None:
|
|
return
|
|
|
|
|
|
def serve(handler: type[BaseHTTPRequestHandler]) -> tuple[ThreadingHTTPServer, threading.Thread]:
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
return server, thread
|
|
|
|
|
|
def alert_lines(state: Path) -> list[str]:
|
|
log = state / "alerts.log"
|
|
if not log.exists():
|
|
return []
|
|
return [line for line in log.read_text(encoding="utf-8").splitlines() if line]
|
|
|
|
|
|
def test_daily_workflow_is_scheduled_from_main_and_reuses_recovery() -> None:
|
|
text = WORKFLOW.read_text(encoding="utf-8")
|
|
assert "\tif:" not in text
|
|
assert "if: gitea.ref == 'refs/heads/main'" in text
|
|
assert 'cron: "0 20 * * *"' in text
|
|
assert "workflow_dispatch:" in text
|
|
assert 'git show "$GITEA_SHA:deploy/run-production-recovery.sh"' in text
|
|
assert "RECOVERY_BASENAME_PREFIX='production-daily'" in text
|
|
assert "prune-production-daily-backups.sh" in text
|
|
assert "production-daily-" in text
|
|
assert 'artifact_retention="${ARTIFACT_RETENTION_DAYS:-14}"' in text
|
|
assert 'local_retention="${LOCAL_RETENTION_DAYS:-7}"' in text
|
|
assert 'process.env["INPUT_RETENTION-DAYS"]=process.env.ARTIFACT_RETENTION_DAYS' in text
|
|
assert "set +x" in text
|
|
assert "pg_dump" not in text
|
|
assert "CHECK_RELEASE_GATE" not in text
|
|
assert "observed_staging_sha" not in text
|
|
assert "STAGING_URL" not in text
|
|
assert "echo \"$SSH_PRIVATE_KEY" not in text
|
|
assert "PRODUCTION_BACKUP_ENCRYPTION_KEY" not in text
|
|
recovery = RECOVERY_WORKFLOW.read_text(encoding="utf-8")
|
|
assert "observed_staging_sha" in recovery
|
|
assert 'process.env["INPUT_RETENTION-DAYS"]="30"' in recovery
|
|
|
|
|
|
def test_recovery_script_keeps_manual_names_unless_daily_prefix() -> None:
|
|
text = RECOVERY_SCRIPT.read_text(encoding="utf-8")
|
|
assert 'backup_prefix="production-pre-migration"' in text
|
|
assert 'production-daily) backup_prefix="production-daily"' in text
|
|
assert "openssl enc -aes-256-cbc -salt -pbkdf2" in text
|
|
assert "usage_percent < 70" in text
|
|
assert "flock -n 9" in text
|
|
assert '[ "$DEPLOY_PATH" = "/opt/jyotisha-production" ]' in text
|
|
assert 'echo "baseline_updated=false" >&2' in text
|
|
assert 'echo "baseline_updated=true" >&2' in text
|
|
assert "daily-backup-latest.env" in text
|
|
|
|
|
|
def install_fake_recovery_bin(bin_dir: Path) -> None:
|
|
bin_dir.mkdir()
|
|
write_text(
|
|
bin_dir / "sudo",
|
|
"#!/usr/bin/env bash\nset -euo pipefail\nif [[ \"${1:-}\" == \"-n\" ]]; then shift; fi\nexec \"$@\"\n",
|
|
)
|
|
write_text(
|
|
bin_dir / "df",
|
|
"#!/usr/bin/env bash\nprintf '%s\\n' 'Filesystem 1024-blocks Used Available Capacity Mounted' 'tmpfs 100 10 90 10% /'\n",
|
|
)
|
|
# Windows cannot chmod pytest's temp directories. These stand-ins only
|
|
# exist on the test PATH. The production script still calls the real commands.
|
|
write_text(
|
|
bin_dir / "chmod",
|
|
"#!/usr/bin/env bash\nexit 0\n",
|
|
)
|
|
# Git Bash on this Windows machine has no flock. The production script still calls it.
|
|
write_text(
|
|
bin_dir / "flock",
|
|
"#!/usr/bin/env bash\nexit 0\n",
|
|
)
|
|
write_text(
|
|
bin_dir / "install",
|
|
"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
dirs=()
|
|
while [[ $# -gt 0 ]]; do
|
|
case "$1" in
|
|
-d) shift ;;
|
|
-m) shift; shift ;;
|
|
--) shift; break ;;
|
|
-*) shift ;;
|
|
*) dirs+=("$1"); shift ;;
|
|
esac
|
|
done
|
|
mkdir -p "${dirs[@]}"
|
|
""",
|
|
)
|
|
write_text(
|
|
bin_dir / "docker",
|
|
"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
cmd="${1:-}"
|
|
shift || true
|
|
case "$cmd" in
|
|
ps)
|
|
printf '%s\\n' fixturecontainer
|
|
;;
|
|
inspect)
|
|
printf '%s\\n' sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
|
|
;;
|
|
run)
|
|
exit 0
|
|
;;
|
|
exec)
|
|
tool=""
|
|
sql=""
|
|
prev=""
|
|
for arg in "$@"; do
|
|
if [[ "$prev" == "-c" ]]; then
|
|
sql="$arg"
|
|
fi
|
|
case "$arg" in
|
|
pg_dump|pg_restore|psql) tool="$arg" ;;
|
|
esac
|
|
prev="$arg"
|
|
done
|
|
case "$tool" in
|
|
pg_dump)
|
|
printf '%s\\n' fixture-dump
|
|
;;
|
|
pg_restore)
|
|
cat >/dev/null
|
|
;;
|
|
psql)
|
|
case "$sql" in
|
|
*role_table_grants*) printf '%s\\n' 'reader|SELECT|profiles' ;;
|
|
*column_privileges*) printf '%s\\n' 'reader|SELECT|profiles|id' ;;
|
|
*pg_get_userbyid*) printf '%s\\n' 'profiles|app' ;;
|
|
*identity.users*) printf '%s\\n' '1|2|3|4' ;;
|
|
*"CREATE DATABASE"*|*"DROP DATABASE"*) ;;
|
|
*)
|
|
printf 'unexpected psql\\n' >&2
|
|
exit 1
|
|
;;
|
|
esac
|
|
;;
|
|
*)
|
|
printf 'unexpected docker exec\\n' >&2
|
|
exit 1
|
|
;;
|
|
esac
|
|
;;
|
|
*)
|
|
printf 'unexpected docker %s\\n' "$cmd" >&2
|
|
exit 1
|
|
;;
|
|
esac
|
|
""",
|
|
)
|
|
subprocess.run(
|
|
[BASH, "-lc", 'chmod +x "$1/sudo" "$1/df" "$1/docker" "$1/chmod" "$1/install" "$1/flock"', "chmod", bash_path(bin_dir)],
|
|
check=True,
|
|
)
|
|
|
|
|
|
def write_recovery_env(deploy: Path) -> None:
|
|
write_text(
|
|
deploy / ".env.production.database",
|
|
"\n".join(
|
|
[
|
|
"POSTGRES_DB=fixturedatabase",
|
|
"POSTGRES_USER=fixtureuser",
|
|
"POSTGRES_PASSWORD=fixturepassword",
|
|
"PRODUCTION_BACKUP_ENCRYPTION_KEY=fixtureencryptionkey",
|
|
"",
|
|
]
|
|
),
|
|
)
|
|
|
|
|
|
def run_recovery(deploy: Path, bin_dir: Path, prefix: str | None) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env.pop("FEISHU_WEBHOOK_URL", None)
|
|
env.pop("FEISHU_SIGN_SECRET", None)
|
|
env.pop("RECOVERY_BASENAME_PREFIX", None)
|
|
env["DEPLOY_PATH"] = bash_path(deploy)
|
|
env["RECOVERY_RUN_ID"] = "4242"
|
|
env["JYOTISHA_RECOVERY_FIXTURE"] = "1"
|
|
if prefix:
|
|
env["RECOVERY_BASENAME_PREFIX"] = prefix
|
|
return subprocess.run(
|
|
[
|
|
BASH,
|
|
"-lc",
|
|
'export PATH="$1:$PATH"; exec bash "$2"',
|
|
"bash",
|
|
bash_path(bin_dir),
|
|
bash_path(RECOVERY_SCRIPT),
|
|
],
|
|
cwd=ROOT,
|
|
env=env,
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
|
|
|
|
def parse_env(text: str) -> dict[str, str]:
|
|
parsed: dict[str, str] = {}
|
|
for line in text.splitlines():
|
|
key, value = line.split("=", 1)
|
|
parsed[key] = value
|
|
return parsed
|
|
|
|
|
|
def test_daily_backup_does_not_replace_the_recovery_point(tmp_path: Path) -> None:
|
|
deploy = tmp_path / "deploy"
|
|
state = deploy / ".state"
|
|
state.mkdir(parents=True)
|
|
write_recovery_env(deploy)
|
|
baseline = state / "recovery-baseline.env"
|
|
original = (
|
|
"RECOVERY_REFERENCE=gitea-actions-run-1/production-pre-migration-20261014T120000Z.dump.enc\n"
|
|
"RECOVERY_CREATED_AT=2026-10-14T12:00:00Z\n"
|
|
"RECOVERY_SHA256=" + ("ab" * 32) + "\n"
|
|
"RESTORE_VERIFIED=true\n"
|
|
"RECOVERY_FORMAT=2\n"
|
|
"VERIFY_FILE=production-pre-migration-20261014T120000Z-restore-verify.json\n"
|
|
)
|
|
write_text(baseline, original)
|
|
subprocess.run(
|
|
[BASH, "-lc", 'touch -d "@$1" "$2"', "touch", "1700000000", bash_path(baseline)],
|
|
check=True,
|
|
)
|
|
before_ns = baseline.stat().st_mtime_ns
|
|
before_bytes = baseline.read_bytes()
|
|
bin_dir = tmp_path / "bin"
|
|
install_fake_recovery_bin(bin_dir)
|
|
completed = run_recovery(deploy, bin_dir, "production-daily")
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert baseline.read_bytes() == before_bytes
|
|
assert baseline.stat().st_mtime_ns == before_ns
|
|
assert "baseline_updated" not in completed.stdout
|
|
assert "baseline_updated=false" in completed.stderr.splitlines()
|
|
assert "baseline_updated=true" not in completed.stderr
|
|
assert len(completed.stdout.splitlines()) == 7
|
|
stdout_env = parse_env(completed.stdout)
|
|
assert stdout_env["BACKUP_BASENAME"].startswith("production-daily-")
|
|
daily = state / "daily-backup-latest.env"
|
|
daily_env = parse_env(daily.read_text(encoding="utf-8"))
|
|
assert set(daily_env) == {"BACKUP_BASENAME", "BACKUP_SHA256", "VERIFIED_AT", "BACKUP_FORMAT"}
|
|
assert daily_env["BACKUP_FORMAT"] == "2"
|
|
assert daily_env["BACKUP_BASENAME"] == stdout_env["BACKUP_BASENAME"]
|
|
assert daily_env["BACKUP_SHA256"] == stdout_env["RECOVERY_SHA256"]
|
|
daily_text = daily.read_text(encoding="utf-8")
|
|
assert "fixturepassword" not in daily_text
|
|
assert "fixtureencryptionkey" not in daily_text
|
|
assert "fixturedatabase" not in daily_text
|
|
verify = deploy / "backups" / stdout_env["VERIFY_BASENAME"]
|
|
payload = json.loads(verify.read_text(encoding="utf-8"))
|
|
assert payload["checks"]["table_grants_match"] is True
|
|
assert payload["checks"]["column_privileges_match"] is True
|
|
assert payload["checks"]["table_owners_match"] is True
|
|
assert "fixturepassword" not in verify.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_manual_recovery_still_writes_format_two(tmp_path: Path) -> None:
|
|
deploy = tmp_path / "deploy"
|
|
state = deploy / ".state"
|
|
state.mkdir(parents=True)
|
|
write_recovery_env(deploy)
|
|
sentinel = state / "daily-backup-latest.env"
|
|
write_text(sentinel, "BACKUP_BASENAME=keep\nBACKUP_FORMAT=2\n")
|
|
before = sentinel.read_bytes()
|
|
bin_dir = tmp_path / "bin"
|
|
install_fake_recovery_bin(bin_dir)
|
|
completed = run_recovery(deploy, bin_dir, None)
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert "baseline_updated=true" in completed.stderr.splitlines()
|
|
assert "baseline_updated=false" not in completed.stderr
|
|
assert len(completed.stdout.splitlines()) == 7
|
|
baseline_text = (state / "recovery-baseline.env").read_text(encoding="utf-8")
|
|
baseline = parse_env(baseline_text)
|
|
assert baseline["RECOVERY_FORMAT"] == "2"
|
|
assert baseline["VERIFY_FILE"].startswith("production-pre-migration-")
|
|
assert baseline["RECOVERY_REFERENCE"].startswith("gitea-actions-run-4242/production-pre-migration-")
|
|
assert "fixturepassword" not in baseline_text
|
|
assert "fixtureencryptionkey" not in baseline_text
|
|
assert parse_env(completed.stdout)["BACKUP_BASENAME"].startswith("production-pre-migration-")
|
|
assert sentinel.read_bytes() == before
|
|
|
|
|
|
def test_release_migration_and_rollback_ignore_daily_backup_record() -> None:
|
|
readers = [
|
|
ROOT / ".gitea" / "workflows" / "deploy-production.yml",
|
|
ROOT / ".gitea" / "workflows" / "migrate-production-database.yml",
|
|
ROOT / "deploy" / "run-production-deploy.sh",
|
|
ROOT / "deploy" / "run-production-migration.sh",
|
|
]
|
|
for path in readers:
|
|
text = path.read_text(encoding="utf-8")
|
|
assert "daily-backup-latest.env" not in text, path.name
|
|
assert "recovery-baseline.env" in text
|
|
for path in (WORKFLOW, RECOVERY_WORKFLOW):
|
|
text = path.read_text(encoding="utf-8")
|
|
assert "JYOTISHA_RECOVERY_FIXTURE" not in text
|
|
assert '[[ "$(wc -l <"$ssh_root/recovery-output.env" | tr -d \' \')" == 7 ]]' in text
|
|
|
|
|
|
def test_prune_deletes_only_expired_daily_files(tmp_path: Path) -> None:
|
|
backup = tmp_path / "backups"
|
|
nested = backup / "nested"
|
|
nested.mkdir(parents=True)
|
|
now = 1_700_000_000
|
|
limit = 7 * 86400
|
|
names = {
|
|
"production-daily-20261001T000000Z.dump.enc": now - limit - 60,
|
|
"production-daily-20261001T000000Z-restore-verify.json": now - limit - 60,
|
|
"production-daily-20261007T000000Z.dump.enc": now - limit,
|
|
"production-daily-20261007T000000Z-restore-verify.json": now - limit,
|
|
"production-daily-20261014T000000Z.dump.enc": now - 3600,
|
|
"production-daily-20261014T000000Z-restore-verify.json": now - 3600,
|
|
"production-pre-migration-20261001T000000Z.dump.enc": now - limit - 3600,
|
|
"production-pre-migration-20261001T000000Z-restore-verify.json": now - limit - 3600,
|
|
"production-daily-notes.txt": now - limit - 3600,
|
|
"production-daily-20261001T000000Z.dump.enc.bak": now - limit - 3600,
|
|
"unrelated.txt": now - limit - 3600,
|
|
}
|
|
for name, epoch in names.items():
|
|
path = backup / name
|
|
write_text(path, "fictional\n")
|
|
subprocess.run(
|
|
[BASH, "-lc", 'touch -d "@$1" "$2"', "touch", str(epoch), bash_path(path)],
|
|
check=True,
|
|
)
|
|
write_text(nested / "production-daily-20261001T000000Z.dump.enc", "nested\n")
|
|
completed = run_bash(PRUNE_SCRIPT, {}, [bash_path(backup), "7", str(now)])
|
|
assert completed.returncode == 0, completed.stderr
|
|
deleted = completed.stdout.splitlines()
|
|
assert deleted == ["production-daily-20261001T000000Z.dump.enc"]
|
|
assert not (backup / "production-daily-20261001T000000Z.dump.enc").exists()
|
|
assert not (backup / "production-daily-20261001T000000Z-restore-verify.json").exists()
|
|
for name in names:
|
|
if name.startswith("production-daily-20261001T000000Z.dump") or name.startswith(
|
|
"production-daily-20261001T000000Z-restore"
|
|
):
|
|
continue
|
|
assert (backup / name).exists(), name
|
|
assert (nested / "production-daily-20261001T000000Z.dump.enc").exists()
|
|
|
|
|
|
def test_health_alerts_after_three_failures_and_recovery_once(tmp_path: Path) -> None:
|
|
server, _thread = serve(HealthHandler)
|
|
server.kind = "500"
|
|
state = tmp_path / "state"
|
|
env = {
|
|
"JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
|
|
"JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
|
|
"JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
|
|
"JYOTISHA_HEALTH_TIMEOUT": "2",
|
|
}
|
|
try:
|
|
for _ in range(2):
|
|
completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert alert_lines(state) == []
|
|
completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
|
|
assert completed.returncode == 0, completed.stderr
|
|
lines = alert_lines(state)
|
|
assert len(lines) == 1
|
|
assert "errorCode=health_down" in lines[0]
|
|
assert "count=3" in lines[0]
|
|
assert "route=/api/health" in lines[0]
|
|
server.kind = "ok"
|
|
server.commit = "b" * 40
|
|
completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
|
|
assert completed.returncode == 0, completed.stderr
|
|
lines = alert_lines(state)
|
|
assert len(lines) == 2
|
|
assert "errorCode=recovered" in lines[1]
|
|
completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert len(alert_lines(state)) == 2
|
|
finally:
|
|
server.shutdown()
|
|
|
|
|
|
def test_health_timeout_and_missing_fields_count_as_failures(tmp_path: Path) -> None:
|
|
server, _thread = serve(HealthHandler)
|
|
server.kind = "timeout"
|
|
state = tmp_path / "state"
|
|
env = {
|
|
"JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
|
|
"JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
|
|
"JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
|
|
"JYOTISHA_HEALTH_TIMEOUT": "1",
|
|
}
|
|
try:
|
|
completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert alert_lines(state) == []
|
|
server.kind = "missing"
|
|
run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
|
|
completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert any("errorCode=health_down" in line for line in alert_lines(state))
|
|
finally:
|
|
server.shutdown()
|
|
|
|
|
|
def test_commit_change_alerts_once_outside_release_window(tmp_path: Path) -> None:
|
|
server, _thread = serve(HealthHandler)
|
|
server.kind = "ok"
|
|
server.commit = "a" * 40
|
|
state = tmp_path / "state"
|
|
env = {
|
|
"JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
|
|
"JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
|
|
"JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
|
|
"JYOTISHA_HEALTH_TIMEOUT": "2",
|
|
}
|
|
try:
|
|
assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
|
|
assert alert_lines(state) == []
|
|
server.commit = "b" * 40
|
|
assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
|
|
lines = alert_lines(state)
|
|
assert len(lines) == 1
|
|
assert "errorCode=commit_changed" in lines[0]
|
|
assert "b" * 40 not in lines[0]
|
|
assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
|
|
assert len(alert_lines(state)) == 1
|
|
marker = state / "release-window"
|
|
state.mkdir(parents=True, exist_ok=True)
|
|
write_text(marker, "open\n")
|
|
server.commit = "c" * 40
|
|
assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
|
|
assert len(alert_lines(state)) == 1
|
|
finally:
|
|
server.shutdown()
|
|
|
|
|
|
def test_disk_alerts_once_and_no_webhook_stays_local(tmp_path: Path) -> None:
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
write_text(
|
|
bin_dir / "df",
|
|
"#!/usr/bin/env bash\nprintf '%s\\n' 'Filesystem 1024-blocks Used Available Capacity Mounted' \"overlay 100 90 10 ${FAKE_DISK_PERCENT:?}% /\"\n",
|
|
)
|
|
subprocess.run([BASH, "-lc", 'chmod +x "$1/df"', "chmod", bash_path(bin_dir)], check=True)
|
|
state = tmp_path / "state"
|
|
env = {
|
|
"JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
|
|
"JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
|
|
"FAKE_DISK_PERCENT": "81",
|
|
}
|
|
env.pop("FEISHU_WEBHOOK_URL", None)
|
|
env.pop("FEISHU_SIGN_SECRET", None)
|
|
completed = subprocess.run(
|
|
[BASH, "-lc", 'export PATH="$1:$PATH"; exec bash "$2" --mode disk', "bash", bash_path(bin_dir), bash_path(HEALTH_SCRIPT)],
|
|
cwd=ROOT,
|
|
env={**os.environ, **env, "JYOTISHA_PYTHON": bash_path(Path(sys.executable))},
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert "errorCode=disk_high" in alert_lines(state)[0]
|
|
assert "count=81" in alert_lines(state)[0]
|
|
env["FAKE_DISK_PERCENT"] = "81"
|
|
again = subprocess.run(
|
|
[BASH, "-lc", 'export PATH="$1:$PATH"; exec bash "$2" --mode disk', "bash", bash_path(bin_dir), bash_path(HEALTH_SCRIPT)],
|
|
cwd=ROOT,
|
|
env={**os.environ, **env, "JYOTISHA_PYTHON": bash_path(Path(sys.executable))},
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
assert again.returncode == 0, again.stderr
|
|
assert len(alert_lines(state)) == 1
|
|
env["FAKE_DISK_PERCENT"] = "10"
|
|
recovered = subprocess.run(
|
|
[BASH, "-lc", 'export PATH="$1:$PATH"; exec bash "$2" --mode disk', "bash", bash_path(bin_dir), bash_path(HEALTH_SCRIPT)],
|
|
cwd=ROOT,
|
|
env={**os.environ, **env, "JYOTISHA_PYTHON": bash_path(Path(sys.executable))},
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
assert recovered.returncode == 0, recovered.stderr
|
|
assert len(alert_lines(state)) == 2
|
|
assert "errorCode=recovered" in alert_lines(state)[1]
|
|
|
|
|
|
def test_feishu_payload_keeps_only_allowed_fields(tmp_path: Path) -> None:
|
|
hook, _thread = serve(HookHandler)
|
|
hook.bodies = []
|
|
health, _thread = serve(HealthHandler)
|
|
health.kind = "500"
|
|
state = tmp_path / "state"
|
|
secret = "fixture-sign-secret"
|
|
env_file = tmp_path / "monitor.env"
|
|
write_text(env_file, f"FEISHU_WEBHOOK_URL=http://127.0.0.1:{hook.server_address[1]}/hook\nFEISHU_SIGN_SECRET={secret}\n")
|
|
env = {
|
|
"JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
|
|
"JYOTISHA_MONITOR_ENV": bash_path(env_file),
|
|
"JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{health.server_address[1]}/health",
|
|
"JYOTISHA_HEALTH_TIMEOUT": "2",
|
|
}
|
|
try:
|
|
for _ in range(3):
|
|
completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert len(hook.bodies) == 1
|
|
payload = json.loads(hook.bodies[0].decode("utf-8"))
|
|
assert set(payload) == {"timestamp", "sign", "msg_type", "content"}
|
|
assert payload["msg_type"] == "text"
|
|
text = payload["content"]["text"]
|
|
assert secret not in text
|
|
assert secret not in hook.bodies[0].decode("utf-8")
|
|
assert "errorCode=health_down" in text
|
|
assert "@" not in text
|
|
expected = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"import base64,hashlib,hmac,os,sys; ts=sys.argv[1]; secret=sys.argv[2]; key=f'{ts}\\n{secret}'.encode(); print(base64.b64encode(hmac.new(key, digestmod=hashlib.sha256).digest()).decode())",
|
|
payload["timestamp"],
|
|
secret,
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
assert payload["sign"] == expected
|
|
finally:
|
|
hook.shutdown()
|
|
health.shutdown()
|
|
|
|
|
|
def uuid(index: int) -> str:
|
|
return f"00000000-0000-4000-8000-{index:012d}"
|
|
|
|
|
|
def test_error_digest_drops_personal_text_and_respects_thresholds(tmp_path: Path) -> None:
|
|
safe = [
|
|
f"2026-10-15T00:00:{index:02d}Z web [agent-observability] {json.dumps({'requestId': uuid(index), 'errorCode': 'settlement_failed'})}"
|
|
for index in range(1, 11)
|
|
]
|
|
unsafe = [
|
|
"2026-10-15T00:01:00Z web person@example.com errorCode=should_drop",
|
|
"2026-10-15T00:01:01Z web note 1900-01-01 birth_date",
|
|
"2026-10-15T00:01:02Z web 对话内容 今天工作不顺",
|
|
]
|
|
fixture = tmp_path / "logs.txt"
|
|
write_text(fixture, "\n".join([*safe, *unsafe]) + "\n")
|
|
state = tmp_path / "state"
|
|
env = {
|
|
"JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
|
|
"JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
|
|
"JYOTISHA_LOG_FIXTURE": bash_path(fixture),
|
|
"JYOTISHA_NOW": "2026-10-15T01:00:00Z",
|
|
}
|
|
below = tmp_path / "below.txt"
|
|
write_text(below, "\n".join(safe[:9]) + "\n")
|
|
quiet = run_bash(DIGEST_SCRIPT, {**env, "JYOTISHA_LOG_FIXTURE": bash_path(below)}, ["--window"])
|
|
assert quiet.returncode == 0, quiet.stderr
|
|
assert alert_lines(state) == []
|
|
completed = run_bash(DIGEST_SCRIPT, env, ["--window"])
|
|
assert completed.returncode == 0, completed.stderr
|
|
lines = alert_lines(state)
|
|
assert len(lines) == 1
|
|
assert "errorCode=settlement_failed" in lines[0]
|
|
assert "count=10" in lines[0]
|
|
assert lines[0].count(",") == 4
|
|
assert uuid(6) not in lines[0]
|
|
assert "person@example.com" not in lines[0]
|
|
assert "1900-01-01" not in lines[0]
|
|
assert "今天工作不顺" not in lines[0]
|
|
assert "should_drop" not in lines[0]
|
|
|
|
fivexx = tmp_path / "fivexx.txt"
|
|
write_text(fivexx, "\n".join(f"2026-10-15T00:02:{index:02d}Z api status=502" for index in range(20)) + "\n")
|
|
state_5xx = tmp_path / "state5"
|
|
completed = run_bash(
|
|
DIGEST_SCRIPT,
|
|
{**env, "JYOTISHA_MONITOR_STATE_DIR": bash_path(state_5xx), "JYOTISHA_LOG_FIXTURE": bash_path(fivexx)},
|
|
["--window"],
|
|
)
|
|
assert completed.returncode == 0, completed.stderr
|
|
joined = "\n".join(alert_lines(state_5xx))
|
|
assert "errorCode=http_5xx" in joined
|
|
assert "count=20" in joined
|
|
|
|
|
|
def test_daily_digest_summarizes_previous_shanghai_day(tmp_path: Path) -> None:
|
|
fixture = tmp_path / "daily.txt"
|
|
write_text(
|
|
fixture,
|
|
"\n".join(
|
|
[
|
|
'2026-10-14T02:00:00Z api [agent-observability] {"requestId":"00000000-0000-4000-8000-000000000001","errorCode":"settlement_failed"}',
|
|
'2026-10-15T01:30:00Z api [agent-observability] {"requestId":"00000000-0000-4000-8000-000000000002","errorCode":"settlement_failed"}',
|
|
]
|
|
)
|
|
+ "\n",
|
|
)
|
|
state = tmp_path / "state"
|
|
completed = run_bash(
|
|
DIGEST_SCRIPT,
|
|
{
|
|
"JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
|
|
"JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
|
|
"JYOTISHA_LOG_FIXTURE": bash_path(fixture),
|
|
"JYOTISHA_NOW": "2026-10-15T01:00:00Z",
|
|
},
|
|
["--daily"],
|
|
)
|
|
assert completed.returncode == 0, completed.stderr
|
|
lines = alert_lines(state)
|
|
assert len(lines) == 1
|
|
assert "count=1" in lines[0]
|
|
assert "00000000-0000-4000-8000-000000000001" in lines[0]
|
|
assert "00000000-0000-4000-8000-000000000002" not in lines[0]
|
|
|
|
|
|
def test_find_request_log_sorts_and_rejects_injection(tmp_path: Path) -> None:
|
|
log_dir = tmp_path / "logs"
|
|
log_dir.mkdir()
|
|
request_id = "11111111-2222-4333-8444-555555555555"
|
|
write_text(log_dir / "caddy.log", f"2026-10-15T03:00:00Z caddy {request_id}\n")
|
|
write_text(log_dir / "web.log", f"2026-10-15T01:00:00Z web {request_id}\nignore\n")
|
|
write_text(log_dir / "api.log", f"2026-10-15T02:00:00Z api {request_id}\n")
|
|
canary = tmp_path / "canary.txt"
|
|
write_text(canary, "keep\n")
|
|
found = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, [request_id, "--since", "24h"])
|
|
assert found.returncode == 0, found.stderr
|
|
assert found.stdout.splitlines() == [
|
|
f"2026-10-15T01:00:00Z web {request_id}",
|
|
f"2026-10-15T02:00:00Z api {request_id}",
|
|
f"2026-10-15T03:00:00Z caddy {request_id}",
|
|
]
|
|
rejected = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, ["; rm -rf /", "--since", "24h"])
|
|
assert rejected.returncode == 2
|
|
assert canary.read_text(encoding="utf-8") == "keep\n"
|
|
injected = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, ["$(rm canary.txt)", "--since", "1h"])
|
|
assert injected.returncode == 2
|
|
assert canary.exists()
|
|
bad_since = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, [request_id, "--since", "24h;rm"])
|
|
assert bad_since.returncode == 2
|
|
|
|
|
|
def service_logging_blocks(text: str) -> dict[str, str]:
|
|
blocks: dict[str, str] = {}
|
|
service: str | None = None
|
|
in_services = False
|
|
capturing = False
|
|
buf: list[str] = []
|
|
for line in text.splitlines():
|
|
if line.startswith("services:"):
|
|
in_services = True
|
|
continue
|
|
if in_services and line and not line.startswith(" "):
|
|
in_services = False
|
|
if not in_services:
|
|
continue
|
|
if line.startswith(" ") and not line.startswith(" ") and line.endswith(":"):
|
|
if capturing and service is not None:
|
|
blocks[service] = "\n".join(buf)
|
|
service = line.strip()[:-1]
|
|
capturing = False
|
|
buf = []
|
|
continue
|
|
if service and line == " logging:":
|
|
capturing = True
|
|
buf = [line]
|
|
continue
|
|
if capturing:
|
|
if line.startswith(" ") and not line.startswith(" "):
|
|
blocks[service] = "\n".join(buf)
|
|
capturing = False
|
|
buf = []
|
|
else:
|
|
buf.append(line)
|
|
if capturing and service is not None:
|
|
blocks[service] = "\n".join(buf)
|
|
return blocks
|
|
|
|
|
|
def test_compose_logging_is_present_and_identical() -> None:
|
|
expected = "\n".join(
|
|
[
|
|
" logging:",
|
|
" driver: json-file",
|
|
" options:",
|
|
' max-size: "20m"',
|
|
' max-file: "10"',
|
|
]
|
|
)
|
|
expected_names = {
|
|
SERVER_COMPOSE: {"api", "web", "caddy"},
|
|
POSTGRES_COMPOSE: {"postgres", "migrator", "migration-checker"},
|
|
}
|
|
for path, names in expected_names.items():
|
|
blocks = service_logging_blocks(path.read_text(encoding="utf-8"))
|
|
assert set(blocks) == names
|
|
for name in names:
|
|
assert blocks[name] == expected, name
|
|
|
|
|
|
def test_shell_scripts_pass_bash_n() -> None:
|
|
for script in (PRUNE_SCRIPT, HEALTH_SCRIPT, DIGEST_SCRIPT, FIND_SCRIPT, NOTIFY_SCRIPT, RECOVERY_SCRIPT):
|
|
completed = subprocess.run(
|
|
[BASH, "-n", bash_path(script)],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
assert completed.returncode == 0, script.name + completed.stderr
|