host
64995fafab
- migrations: MIGRATION_TABLES now equals POSTGRES_SCHEMA tables
(uploads, task_attachments, task_followups, environments added);
SEQUENCE_TABLES matches IDENTITY tables; task_followups
sender_task_id column ensured before insert; UNMIGRATED_TABLES
and migration_coverage() report tables with no PG schema instead
of silently skipping them; CLI prints the coverage report.
- deploy: DEFAULT_TEST_SUITES gains deploy/tests (discover -s only,
never package path); regression tests pin the default list.
- monitor: MONITOR_SINGLE_HOST=1 makes the server check single-host;
default stays dual to avoid silent downgrade; server_mode is in
the JSON. Optional fail-only alert outlet (webhook or mail,
default off, alert body is pass/fail JSON only, no secrets).
- backup/rollback: README and restore-drill document what is
recoverable (DB, code) vs not (worktrees, traces, uploads, host
configs); rollback.sh shares the server-unit list and dual-port
health success standard with host_deploy.sh (sync test guards it).
- public entry: Caddyfile uses native {$DOMAIN}/{$ACME_EMAIL}
variables plus a caddy.service template; README documents the
steps and the current BT-nginx production status.
297 lines
13 KiB
Python
297 lines
13 KiB
Python
"""P2-ops 回归：monitor 单机开关与告警出口、rollback 与
|
||
host_deploy 的 unit/健康清单同步。
|
||
|
||
全部用假 systemctl / curl / ssh / mail，不碰真实主机与真实服务。
|
||
告警测试用的 webhook 接收器只监听 127.0.0.1 临时端口。
|
||
"""
|
||
import http.server
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import tarfile
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
|
||
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
|
||
os.path.abspath(__file__))))
|
||
DEPLOY_DIR = os.path.join(REPO_ROOT, "deploy")
|
||
MONITOR_SCRIPT = os.path.join(DEPLOY_DIR, "monitor.sh")
|
||
ROLLBACK_SCRIPT = os.path.join(DEPLOY_DIR, "rollback.sh")
|
||
HOST_DEPLOY_SCRIPT = os.path.join(DEPLOY_DIR, "host_deploy.sh")
|
||
REAL_CURL = shutil.which("curl")
|
||
|
||
|
||
def _write_exe(path, body):
|
||
with open(path, "w") as f:
|
||
f.write(body)
|
||
os.chmod(path, 0o755)
|
||
|
||
|
||
class MonitorSim(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
base = self.tmp.name
|
||
self.app_dir = os.path.join(base, "app")
|
||
os.makedirs(self.app_dir)
|
||
self.backup_dir = os.path.join(base, "backups")
|
||
os.makedirs(self.backup_dir)
|
||
with open(os.path.join(self.backup_dir,
|
||
"cloud-backup-fresh.sql"), "w") as f:
|
||
f.write("-- fresh\n")
|
||
self.health_path = os.path.join(base, "health.json")
|
||
# 健康正文里埋一个哨兵串：告警正文绝不能把它带出去
|
||
# （告警只能发 LAST_JSON 的 pass/fail，不发 health 正文）。
|
||
with open(self.health_path, "w") as f:
|
||
json.dump({"ok": True, "backend": "postgres", "db_ok": True,
|
||
"worker_ok": True,
|
||
"note": "sentinel-health-secret"}, f)
|
||
fake_bin = os.path.join(base, "fakebin")
|
||
os.makedirs(fake_bin)
|
||
_write_exe(os.path.join(fake_bin, "systemctl"),
|
||
"#!/bin/sh\n"
|
||
"if [ \"$1\" = \"is-active\" ]; then\n"
|
||
" case \"$2\" in\n"
|
||
" codex-cloud-server) exit \"$FAKE_SERVER1\" ;;\n"
|
||
" codex-cloud-server-2) exit \"$FAKE_SERVER2\" ;;\n"
|
||
" *) exit 1 ;;\n"
|
||
" esac\n"
|
||
"fi\n"
|
||
"exit 0\n")
|
||
real_curl = REAL_CURL or "/usr/bin/curl"
|
||
self.curl_log = os.path.join(base, "curl.log")
|
||
open(self.curl_log, "w").close()
|
||
_write_exe(os.path.join(fake_bin, "curl"),
|
||
"#!/bin/sh\n"
|
||
"echo \"$@\" >> \"$FAKE_CURL_LOG\"\n"
|
||
"for a in \"$@\"; do\n"
|
||
" case \"$a\" in\n"
|
||
" *\"/health\") cat \"$FAKE_HEALTH_JSON\"; exit 0 ;;\n"
|
||
" esac\n"
|
||
"done\n"
|
||
"exec \"%s\" \"$@\"\n" % real_curl)
|
||
self.fake_bin = fake_bin
|
||
self.env = dict(os.environ)
|
||
self.env["PATH"] = fake_bin + os.pathsep + os.environ["PATH"]
|
||
self.env["MONITOR_HEALTH_URL"] = "http://127.0.0.1:8765/health"
|
||
self.env["MONITOR_APP_DIR"] = self.app_dir
|
||
self.env["MONITOR_DISK_PATH"] = base
|
||
self.env["MONITOR_BACKUP_DIR"] = self.backup_dir
|
||
self.env["MONITOR_ALERT_URL"] = ""
|
||
self.env["MONITOR_ALERT_MAILTO"] = ""
|
||
self.env["FAKE_HEALTH_JSON"] = self.health_path
|
||
self.env["FAKE_CURL_LOG"] = self.curl_log
|
||
self.env["FAKE_SERVER1"] = "0" # server-1 active
|
||
self.env["FAKE_SERVER2"] = "1" # server-2 inactive
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def run_monitor(self, extra_env=None):
|
||
env = dict(self.env)
|
||
if extra_env:
|
||
env.update(extra_env)
|
||
r = subprocess.run(["sh", MONITOR_SCRIPT],
|
||
capture_output=True, text=True, env=env)
|
||
return r, json.loads(r.stdout.strip())
|
||
|
||
def test_single_host_passes_server_check(self):
|
||
_, out = self.run_monitor({"MONITOR_SINGLE_HOST": "1"})
|
||
self.assertEqual(out["server_mode"], "single")
|
||
self.assertEqual(out["checks"]["server"], "pass")
|
||
|
||
def test_dual_host_fails_server_check_when_server2_down(self):
|
||
# 默认双机：server-2 不 active 即 fail，不能静默降级。
|
||
_, out = self.run_monitor()
|
||
self.assertEqual(out["server_mode"], "dual")
|
||
self.assertEqual(out["checks"]["server"], "fail")
|
||
|
||
def test_webhook_alert_on_fail_only(self):
|
||
if REAL_CURL is None:
|
||
self.skipTest("本机无 curl，跳过 webhook 告警测试")
|
||
received = []
|
||
|
||
class Handler(http.server.BaseHTTPRequestHandler):
|
||
def do_POST(self):
|
||
length = int(self.headers.get("Content-Length", 0))
|
||
received.append(self.rfile.read(length))
|
||
self.send_response(200)
|
||
self.end_headers()
|
||
|
||
def log_message(self, *a):
|
||
pass
|
||
|
||
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
try:
|
||
url = "http://127.0.0.1:%d/hook" % server.server_address[1]
|
||
r, out = self.run_monitor({"MONITOR_ALERT_URL": url})
|
||
self.assertNotEqual(r.returncode, 0) # 其它项 fail，整体 fail
|
||
self.assertEqual(len(received), 1)
|
||
body = json.loads(received[0].decode())
|
||
self.assertEqual(body["overall"], "fail")
|
||
self.assertNotIn("sentinel-health-secret",
|
||
received[0].decode())
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
|
||
def test_alert_default_off(self):
|
||
# 不配出口：即使整体 fail，monitor 也不应发出任何告警请求。
|
||
# 行为证据：假 curl 记录了全部调用，其中没有任何 POST
|
||
# （webhook 告警是唯一的 POST 调用；health 检查是 GET）。
|
||
r, out = self.run_monitor()
|
||
self.assertNotEqual(r.returncode, 0)
|
||
self.assertEqual(out["overall"], "fail")
|
||
with open(self.curl_log) as f:
|
||
calls = f.read()
|
||
self.assertNotIn("POST", calls)
|
||
self.assertIn("/health", calls)
|
||
|
||
def test_mail_alert_uses_mail_command(self):
|
||
mail_log = os.path.join(self.tmp.name, "mail.log")
|
||
_write_exe(os.path.join(self.fake_bin, "mail"),
|
||
"#!/bin/sh\n"
|
||
"echo \"ARGS: $@\" >> \"%s\"\n"
|
||
"cat >> \"%s\"\n" % (mail_log, mail_log))
|
||
r, _ = self.run_monitor(
|
||
{"MONITOR_ALERT_MAILTO": "ops@example.invalid"})
|
||
self.assertNotEqual(r.returncode, 0)
|
||
with open(mail_log) as f:
|
||
content = f.read()
|
||
self.assertIn("codex monitor FAIL", content)
|
||
self.assertIn('"overall": "fail"', content)
|
||
self.assertNotIn("sentinel-health-secret", content)
|
||
|
||
|
||
class RollbackSim(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
base = self.tmp.name
|
||
self.app = os.path.join(base, "codex-app-server")
|
||
os.makedirs(os.path.join(self.app, "work"))
|
||
with open(os.path.join(self.app, "work", "keep.txt"), "w") as f:
|
||
f.write("keep\n")
|
||
with open(os.path.join(self.app, "broken.py"), "w") as f:
|
||
f.write("x = 0\n")
|
||
self.backup_dir = os.path.join(base, "backups")
|
||
os.makedirs(self.backup_dir)
|
||
# 备份包：顶层目录名必须等于 APP_DIR 的 basename，
|
||
# rollback 解包到 dirname(APP_DIR) 后正好落回 APP_DIR。
|
||
src = os.path.join(base, "src", "codex-app-server")
|
||
os.makedirs(src)
|
||
with open(os.path.join(src, "restored.py"), "w") as f:
|
||
f.write("x = 1\n")
|
||
with open(os.path.join(src, "DEPLOYED_COMMIT"), "w") as f:
|
||
f.write("backupcommit\n")
|
||
tarball = os.path.join(self.backup_dir,
|
||
"codex-app-server-20260101.tar.gz")
|
||
with tarfile.open(tarball, "w:gz") as tf:
|
||
for root, dirs, files in os.walk(src):
|
||
dirs.sort()
|
||
for name in sorted(files):
|
||
full = os.path.join(root, name)
|
||
tf.add(full, arcname=os.path.relpath(full, os.path.join(
|
||
base, "src")))
|
||
self.health_dir = os.path.join(base, "health")
|
||
os.makedirs(self.health_dir)
|
||
fake_bin = os.path.join(base, "fakebin")
|
||
os.makedirs(fake_bin)
|
||
self.log_path = os.path.join(base, "systemctl.log")
|
||
open(self.log_path, "w").close()
|
||
_write_exe(os.path.join(fake_bin, "ssh"),
|
||
"#!/bin/sh\n"
|
||
"# 假 ssh：忽略目标参数，直接把 heredoc 脚本在本地跑。\n"
|
||
"exec sh -s\n")
|
||
_write_exe(os.path.join(fake_bin, "systemctl"),
|
||
"#!/bin/sh\n"
|
||
"echo \"$@\" >> \"$FAKE_SYSTEMCTL_LOG\"\n"
|
||
"if [ \"$1\" = \"cat\" ] && [ \"$2\" = \"codex-cloud-server-2\" ]; then\n"
|
||
" if [ \"$FAKE_SERVER2_PRESENT\" = \"0\" ]; then exit 1; fi\n"
|
||
" exit 0\n"
|
||
"fi\n"
|
||
"if [ \"$1\" = \"is-active\" ]; then exit 1; fi\n"
|
||
"exit 0\n")
|
||
_write_exe(os.path.join(fake_bin, "curl"),
|
||
"#!/bin/sh\n"
|
||
"for a in \"$@\"; do\n"
|
||
" case \"$a\" in\n"
|
||
" *\":8765/health\") cat \"$FAKE_HEALTH_DIR/8765.json\"; exit 0 ;;\n"
|
||
" *\":8766/health\") cat \"$FAKE_HEALTH_DIR/8766.json\"; exit 0 ;;\n"
|
||
" esac\n"
|
||
"done\n"
|
||
"exit 1\n")
|
||
self.env = dict(os.environ)
|
||
self.env["PATH"] = fake_bin + os.pathsep + os.environ["PATH"]
|
||
self.env["CODEX_ROLLBACK_APP_DIR"] = self.app
|
||
self.env["CODEX_ROLLBACK_BACKUP_DIR"] = self.backup_dir
|
||
self.env["CODEX_ROLLBACK_HEALTH_ATTEMPTS"] = "2"
|
||
self.env["CODEX_ROLLBACK_HEALTH_INTERVAL"] = "0.1"
|
||
self.env["FAKE_SYSTEMCTL_LOG"] = self.log_path
|
||
self.env["FAKE_HEALTH_DIR"] = self.health_dir
|
||
self.env["FAKE_SERVER2_PRESENT"] = "1"
|
||
self.env.pop("SSH_PROXY_CMD", None)
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def set_health(self, port, ok):
|
||
with open(os.path.join(self.health_dir, "%s.json" % port), "w") as f:
|
||
json.dump({"ok": ok, "backend": "postgres", "db_ok": ok}, f)
|
||
|
||
def run_rollback(self):
|
||
return subprocess.run(["sh", ROLLBACK_SCRIPT, "testhost"],
|
||
capture_output=True, text=True, env=self.env)
|
||
|
||
def systemctl_log(self):
|
||
with open(self.log_path) as f:
|
||
return f.read()
|
||
|
||
def test_success_requires_both_ports(self):
|
||
self.set_health(8765, True)
|
||
self.set_health(8766, True)
|
||
r = self.run_rollback()
|
||
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
|
||
self.assertIn("全部端口 health 通过", r.stdout)
|
||
log = self.systemctl_log()
|
||
self.assertIn(
|
||
"stop codex-cloud-worker codex-cloud-server codex-cloud-server-2",
|
||
log)
|
||
self.assertIn(
|
||
"start codex-cloud-server codex-cloud-server-2 codex-cloud-worker",
|
||
log)
|
||
self.assertTrue(os.path.exists(os.path.join(self.app, "restored.py")))
|
||
self.assertFalse(os.path.exists(os.path.join(self.app, "broken.py")))
|
||
with open(os.path.join(self.app, "work", "keep.txt")) as f:
|
||
self.assertEqual(f.read().strip(), "keep")
|
||
|
||
def test_server2_health_failure_fails_rollback(self):
|
||
# 8765 通、8766 不通：只通一端口不算回滚成功。
|
||
self.set_health(8765, True)
|
||
self.set_health(8766, False)
|
||
r = self.run_rollback()
|
||
self.assertNotEqual(r.returncode, 0, r.stdout + r.stderr)
|
||
self.assertIn("health 仍失败", r.stdout + r.stderr)
|
||
|
||
|
||
class TestUnitListSync(unittest.TestCase):
|
||
"""两脚本清单同步核对：任一脚本改了清单、另一脚本没跟进，
|
||
这里就变红，而不是等到回滚时才发现 server-2 没被停/验。"""
|
||
|
||
def test_both_scripts_share_server_units_and_ports(self):
|
||
with open(ROLLBACK_SCRIPT) as f:
|
||
rollback = f.read()
|
||
with open(HOST_DEPLOY_SCRIPT) as f:
|
||
host_deploy = f.read()
|
||
for script, name in ((rollback, "rollback.sh"),
|
||
(host_deploy, "host_deploy.sh")):
|
||
self.assertIn("codex-cloud-server-2", script, name)
|
||
self.assertIn("8766", script, name)
|
||
self.assertIn("共享的 unit 清单", script, name)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|