264 lines
11 KiB
Python
264 lines
11 KiB
Python
"""svc.sh agent 数据目录快照回归（2026-10-12）。
|
||
|
||
验证 backup/backups：活跃 WAL 写入者下 SQLite 一致性导出（.backup + integrity_check，
|
||
不裸拷 .db/-wal）、快照落在 runtime/ 之外的 runtime-backups/<name>/、MANIFEST 逐文件
|
||
sha256、--keep/--prune 保留策略只清理时间戳目录、--out 自定义目录不参与清理、
|
||
kind 守卫（postgres 引导 db backup）、缺 sqlite3 拒绝备份、rm --purge 连带删除快照。
|
||
全部使用临时目录 + 手工登记 svc.env（backup 不触碰 bin/进程，无需桩二进制）。
|
||
"""
|
||
import os
|
||
import shutil
|
||
import sqlite3
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from _port_alloc import free_test_port # noqa: E402
|
||
|
||
SOURCE = Path(__file__).resolve().parents[1]
|
||
SVC_SCRIPT = SOURCE / "scripts" / "svc.sh"
|
||
|
||
SQLITE3 = shutil.which("sqlite3")
|
||
|
||
# 活跃写入者：后台进程持有 WAL 连接（提交后不关闭），保证备份时 -wal 真实存在——
|
||
# 裸拷 .db 会丢这部分已提交内容，.backup 不会。
|
||
WRITER = r"""
|
||
import sqlite3, sys, time
|
||
db = sys.argv[1]
|
||
c = sqlite3.connect(db)
|
||
c.execute("pragma journal_mode=wal")
|
||
c.execute("create table if not exists t(x)")
|
||
c.execute("insert into t values (42)")
|
||
c.commit()
|
||
print("ready", flush=True)
|
||
time.sleep(float(sys.argv[2]))
|
||
"""
|
||
|
||
|
||
@unittest.skipUnless(SQLITE3, "agent 备份依赖 sqlite3 CLI（一致性导出不回退裸拷贝）")
|
||
class SvcAgentBackupTests(unittest.TestCase):
|
||
def setUp(self):
|
||
tmp = tempfile.mkdtemp(prefix="svc-backup-")
|
||
self.addCleanup(shutil.rmtree, tmp, True)
|
||
self.root = Path(tmp)
|
||
(self.root / "scripts").mkdir(parents=True)
|
||
shutil.copy2(SVC_SCRIPT, self.root / "scripts" / "svc.sh")
|
||
self.writer_proc = None
|
||
self.env = os.environ.copy()
|
||
self.env.update({
|
||
"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
|
||
"SVC_SRC_DIR": str(self.root / "srccache"),
|
||
"XDG_CONFIG_HOME": str(self.root / "xdg"),
|
||
})
|
||
for key in list(self.env):
|
||
if key.startswith("SVC_") and key != "SVC_SRC_DIR":
|
||
self.env.pop(key)
|
||
for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE"):
|
||
self.env.pop(key, None)
|
||
|
||
# ---- 夹具 ----
|
||
|
||
def svc(self, *args, check=True, env=None):
|
||
proc = subprocess.run(
|
||
["bash", str(self.root / "scripts" / "svc.sh"), *map(str, args)],
|
||
cwd=self.root, env=env if env is not None else self.env, text=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60)
|
||
if check:
|
||
self.assertEqual(proc.returncode, 0, proc.stdout)
|
||
return proc
|
||
|
||
def register_agent(self, name="t1", port=None):
|
||
if port is None:
|
||
port = free_test_port()
|
||
d = self.root / "runtime" / name
|
||
(d / "data").mkdir(parents=True, exist_ok=True)
|
||
(d / "svc.env").write_text(
|
||
"kind=agent\nenv=%s\npid=\naddr=127.0.0.1:%d\ntags=\n"
|
||
"source=dev@test123\nversion=stub\nstarted=\npidcheck=\n"
|
||
"exec=--addr 127.0.0.1:%d --data %s\n" % (name, port, port, d / "data"))
|
||
return d
|
||
|
||
def seed_db(self, name="t1", live_writer=True, hold_seconds=30):
|
||
"""建 WAL 库并提交一行；live_writer=True 时留后台连接持有 WAL。"""
|
||
db = self.root / "runtime" / name / "data" / "agent-system.db"
|
||
if live_writer:
|
||
self.writer_proc = subprocess.Popen(
|
||
["python3", "-c", WRITER, str(db), str(hold_seconds)],
|
||
stdout=subprocess.PIPE, text=True)
|
||
self.assertEqual(self.writer_proc.stdout.readline().strip(), "ready")
|
||
self.addCleanup(self.stop_writer)
|
||
else:
|
||
c = sqlite3.connect(db)
|
||
c.execute("pragma journal_mode=wal")
|
||
c.execute("create table t(x)")
|
||
c.execute("insert into t values (42)")
|
||
c.commit()
|
||
c.close()
|
||
return db
|
||
|
||
def stop_writer(self):
|
||
if self.writer_proc and self.writer_proc.poll() is None:
|
||
self.writer_proc.kill()
|
||
self.writer_proc.wait(timeout=10)
|
||
|
||
def snapshots(self, name="t1"):
|
||
root = self.root / "runtime-backups" / name
|
||
if not root.is_dir():
|
||
return []
|
||
return sorted(p.name for p in root.iterdir() if p.is_dir())
|
||
|
||
# ---- 用例 ----
|
||
|
||
def test_backup_live_writer_consistent_and_outside_runtime(self):
|
||
"""活跃 WAL 写入者：快照可读、行在、integrity ok；-wal/-shm 不入快照。"""
|
||
self.register_agent()
|
||
db = self.seed_db()
|
||
self.assertTrue(db.with_name("agent-system.db-wal").exists()) # 前置：WAL 真实存在
|
||
proc = self.svc("t1", "backup")
|
||
self.assertIn("runtime-backups/t1/", proc.stdout)
|
||
snaps = self.snapshots()
|
||
self.assertEqual(len(snaps), 1)
|
||
snap = self.root / "runtime-backups" / "t1" / snaps[0]
|
||
# 位置在 runtime/ 之外（runtime 整删不波及）
|
||
self.assertTrue(snap.exists())
|
||
self.assertFalse((self.root / "runtime" / "t1" / "backups").exists())
|
||
# 内容：登记 + 数据库一致性副本
|
||
self.assertTrue((snap / "svc.env").exists())
|
||
bak = snap / "data" / "agent-system.db"
|
||
self.assertTrue(bak.exists())
|
||
self.assertFalse((snap / "data" / "agent-system.db-wal").exists())
|
||
self.assertFalse((snap / "data" / "agent-system.db-shm").exists())
|
||
out = subprocess.run([SQLITE3, str(bak), "select x from t; pragma integrity_check;"],
|
||
capture_output=True, text=True, timeout=30)
|
||
self.assertEqual(out.stdout.split(), ["42", "ok"], out.stderr)
|
||
|
||
def test_backup_manifest_sha_matches(self):
|
||
self.register_agent()
|
||
self.seed_db(live_writer=False)
|
||
self.svc("t1", "backup")
|
||
snap = self.root / "runtime-backups" / "t1" / self.snapshots()[0]
|
||
manifest = (snap / "MANIFEST").read_text()
|
||
self.assertIn("service=t1\n", manifest)
|
||
self.assertIn("source=dev@test123\n", manifest)
|
||
entries = [l.split() for l in manifest.splitlines() if " ./" in l]
|
||
self.assertTrue(entries)
|
||
for sha, rel in entries:
|
||
self.assertRegex(sha, r"^[0-9a-f]{64}$")
|
||
self.assertEqual(
|
||
subprocess.run(["sha256sum", str(snap / rel)],
|
||
capture_output=True, text=True).stdout.split()[0], sha)
|
||
|
||
def test_backup_non_sqlite_db_kept_raw_without_leftovers(self):
|
||
self.register_agent()
|
||
self.seed_db(live_writer=False)
|
||
junk = self.root / "runtime" / "t1" / "data" / "blob.db"
|
||
junk.write_bytes(b"plain-bytes-not-a-sqlite-file")
|
||
proc = self.svc("t1", "backup")
|
||
self.assertIn("blob.db", proc.stdout) # 告警原样保留
|
||
snap = self.root / "runtime-backups" / "t1" / self.snapshots()[0]
|
||
self.assertEqual((snap / "data" / "blob.db").read_bytes(), b"plain-bytes-not-a-sqlite-file")
|
||
self.assertFalse(list(snap.rglob("*.bk"))) # 失败尝试的半成品不残留
|
||
|
||
def test_backup_target_exists_refused(self):
|
||
self.register_agent()
|
||
self.seed_db(live_writer=False)
|
||
dest = self.root / "runtime-backups" / "t1" / "20260101T000000Z"
|
||
dest.mkdir(parents=True)
|
||
(dest / "marker").write_text("x")
|
||
proc = self.svc("t1", "backup", "--out", dest, check=False)
|
||
self.assertEqual(proc.returncode, 1, proc.stdout)
|
||
self.assertIn("已存在", proc.stdout)
|
||
self.assertTrue((dest / "marker").exists()) # 不覆盖既有内容
|
||
|
||
def test_backup_custom_out_not_pruned(self):
|
||
self.register_agent()
|
||
self.seed_db(live_writer=False)
|
||
outdir = self.root / "custom-bk"
|
||
self.svc("t1", "backup", "--out", outdir)
|
||
self.assertTrue((outdir / "data" / "agent-system.db").exists())
|
||
# 默认 root 无快照；再走默认路径备份 --keep 0 也不应动自定义目录
|
||
self.svc("t1", "backup", "--keep", "0")
|
||
self.assertEqual(self.snapshots(), [])
|
||
self.assertTrue(outdir.exists())
|
||
|
||
def test_retention_and_listing(self):
|
||
self.register_agent()
|
||
self.seed_db(live_writer=False)
|
||
for _ in range(3):
|
||
self.svc("t1", "backup")
|
||
time.sleep(1.1) # 时间戳目录名唯一（UTC 秒级）
|
||
self.assertEqual(len(self.snapshots()), 3)
|
||
proc = self.svc("t1", "backups")
|
||
for name in self.snapshots():
|
||
self.assertIn(name, proc.stdout)
|
||
# backup --keep 2 → 剩 2 份最新
|
||
self.svc("t1", "backup", "--keep", "2")
|
||
self.assertEqual(len(self.snapshots()), 2)
|
||
# backups --prune --keep 1 → 剩最新 1 份
|
||
self.svc("t1", "backups", "--prune", "--keep", "1")
|
||
self.assertEqual(len(self.snapshots()), 1)
|
||
|
||
def test_backup_event_recorded(self):
|
||
self.register_agent()
|
||
self.seed_db(live_writer=False)
|
||
self.svc("t1", "backup")
|
||
hist = (self.root / "runtime" / "t1" / "history.jsonl").read_text()
|
||
self.assertIn('"event":"backup"', hist)
|
||
self.assertIn('"result":"ok"', hist)
|
||
self.assertIn("runtime-backups/t1/", hist)
|
||
|
||
def test_backup_rejects_postgres_with_db_hint(self):
|
||
d = self.root / "runtime" / "pgx"
|
||
d.mkdir(parents=True)
|
||
(d / "svc.env").write_text("kind=postgres\nenv=pgx\naddr=127.0.0.1:8890\n")
|
||
proc = self.svc("pgx", "backup", check=False)
|
||
self.assertEqual(proc.returncode, 2, proc.stdout)
|
||
self.assertIn("db backup", proc.stdout)
|
||
|
||
def test_backup_requires_registration(self):
|
||
proc = self.svc("ghost", "backup", check=False)
|
||
self.assertEqual(proc.returncode, 2, proc.stdout)
|
||
|
||
def test_backup_without_sqlite3_refuses(self):
|
||
"""PATH 无 sqlite3 → 拒绝（不回退裸拷贝）。重建与真实 PATH 等价的工具目录、
|
||
唯独不链 sqlite3——只测 command -v 失败这一条分支，其余行为不变形。"""
|
||
self.register_agent()
|
||
self.seed_db(live_writer=False)
|
||
bindir = self.root / "bin-nosqlite3"
|
||
bindir.mkdir()
|
||
seen = set()
|
||
for d in self.env.get("PATH", os.environ["PATH"]).split(os.pathsep):
|
||
if not d or not os.path.isdir(d):
|
||
continue
|
||
for f in os.listdir(d):
|
||
if f == "sqlite3" or f in seen or (bindir / f).exists():
|
||
continue
|
||
seen.add(f)
|
||
try:
|
||
os.symlink(os.path.join(d, f), bindir / f)
|
||
except OSError:
|
||
pass
|
||
self.assertFalse((bindir / "sqlite3").exists())
|
||
env = dict(self.env)
|
||
env["PATH"] = str(bindir)
|
||
proc = self.svc("t1", "backup", check=False, env=env)
|
||
self.assertEqual(proc.returncode, 1, proc.stdout)
|
||
self.assertIn("sqlite3", proc.stdout)
|
||
self.assertEqual(self.snapshots(), []) # 无半成品快照
|
||
|
||
def test_rm_purge_removes_snapshots(self):
|
||
self.register_agent()
|
||
self.seed_db(live_writer=False)
|
||
self.svc("t1", "backup")
|
||
self.assertEqual(len(self.snapshots()), 1)
|
||
self.svc("rm", "t1", "--purge")
|
||
self.assertFalse((self.root / "runtime-backups" / "t1").exists())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|