"""svc.sh U2 回归：不可变版本目录（releases/）+ current 原子切换 + deploy.journal + 失败回退（2026-10-12 统一部署方案 U2）。 覆盖：release 布局与 current 解析（pidcheck 经未解析路径跨版本稳定）、启动失败 回退上一完整版本并恢复期望态、构建失败旧服务保持运行、中断切换经 start/recover 入口回退、schema 前进阻断回退（不触碰业务库）、legacy 布局兼容。夹具复用 tests/test_svc_identity.py（IdentityTests：manifest 治理 + 源仓轨道 + noid 桩）。 """ import importlib.util import json import os import shlex import shutil import subprocess import unittest from pathlib import Path SOURCE = Path(__file__).resolve().parents[1] _spec = importlib.util.spec_from_file_location("tsr", SOURCE / "tests" / "test_svc_recover.py") tsr = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(tsr) _spec2 = importlib.util.spec_from_file_location("tsi", SOURCE / "tests" / "test_svc_identity.py") tsi = importlib.util.module_from_spec(_spec2) _spec2.loader.exec_module(tsi) class U2ReleaseTests(tsi.IdentityTests): """manifest 治理 + identity 桩 + dev 轨道源仓（make_src_repo 等）。""" def setUp(self): super().setUp() # --from local 需要根为 git 仓（local@HEAD 作源标签）；recover 夹具不初始化。 self.git_ok(self.root, "init", "--initial-branch=main", ".") self.git_ok(self.root, "add", ".") self.git_ok(self.root, "commit", "-m", "u2 fixture") # ---- 夹具 ---- def cur_release(self, name): c = self.runtime(name) / "current" return c.resolve().name if c.is_symlink() else None def active_bin(self, name): return self.runtime(name) / "current" / "bin" / "agent-system" def cmdline_of(self, pid): return Path("/proc/%d/cmdline" % pid).read_bytes().replace(b"\0", b" ").decode() def journal(self, name): f = self.runtime(name) / "deploy.journal" return json.loads(f.read_text()) if f.exists() else None def die_stub(self): """identity 可用、serve 即退的桩（模拟新制品启动失败）。""" p = self.root / "stub-agent-system-die" if not p.exists(): stub = tsr.STUB head, sep, tail = stub.partition(" serve)") assert sep, "STUB 缺 serve 分支锚点" _, sep2, tail2 = tail.partition(' wait "$child"\n ;;') assert sep2, "STUB 缺 serve 分支结尾锚点" p.write_text(head + ' serve)\n shift\n' + ' echo "die-stub: serve exits immediately" >&2\n' + " exit 1 ;;" + tail2) p.chmod(0o755) return p def set_makefile(self, stub): (self.root / "Makefile").write_text( "build:\n\tmkdir -p bin\n\tcp %s bin/agent-system\n" % stub) def bump_schema(self, name, version): """把库 schema_migrations 推进到 version（桩库无该表则建；模拟新制品 已迁移数据库）。""" import sqlite3 db = self.runtime(name) / "data" / "agent-system.db" con = sqlite3.connect(str(db)) con.execute( "CREATE TABLE IF NOT EXISTS schema_migrations" "(id INTEGER PRIMARY KEY, name TEXT, applied_at TEXT)") for i in (1, 2, 3, version): con.execute( "INSERT OR IGNORE INTO schema_migrations(id,name,applied_at)" " VALUES(?, 'u2-test', '2026-10-12T00:00:00Z')", (i,)) con.commit() con.close() # ---- T1：release 布局 + 启动失败回退 + 期望态恢复 ---- def test_U2_layout_switch_and_rollback_on_start_failure(self): # Both deployments intentionally share the same clock tick: a failed # candidate must never overwrite the version needed for rollback. clock_bin = self.root / "fixed-clock" clock_bin.mkdir() real_date = shutil.which("date") clock = clock_bin / "date" clock.write_text( '#!/bin/bash\n' 'if [[ "$*" == *"+%Y%m%dT%H%M%SZ"* ]]; then\n' ' echo 20261012T120000Z\n' 'else\n' f' exec {shlex.quote(real_date)} "$@"\n' 'fi\n') clock.chmod(0o755) self.env["PATH"] = str(clock_bin) + os.pathsep + self.env["PATH"] self.make_src_repo() # 预克隆 srccache（本用例走 --from local，仅保环境一致） self.seed_agent("u2a", 9461) self.start_ok("u2a") # legacy 运行中（期望态=running） self.assertTrue(self.running("u2a")) # v1 部署：legacy 领养 → current 切换 → 经 current 启动 self.set_makefile(self.stub) result = self.svc("u2a", "deploy", "--from", "local") self.assertIn("已切换 release", result.stdout) self.assertIn("领养 legacy", result.stdout) cur1 = self.cur_release("u2a") self.assertTrue(cur1 and cur1.startswith("local-"), cur1) self.assertTrue(self.active_bin("u2a").exists()) releases = sorted((self.runtime("u2a") / "releases").iterdir()) self.assertEqual(len(releases), 2, "legacy + v1 两个 release") pid = self.pid_of("u2a") self.assertIn("current/bin/agent-system", self.cmdline_of(pid), "进程 cmdline 须走 current 未解析路径（pidcheck 跨版本稳定）") self.assertIn("current/bin", self.kv("u2a", "pidcheck") or "") v1_bin = self.active_bin("u2a").read_bytes() self.assertEqual(v1_bin, Path(self.stub).read_bytes(), "v1 制品 = 可用桩") self.assertIsNone(self.journal("u2a"), "成功后 journal 清理") # v2 部署：serve 即退 → 回退 v1 并恢复运行 self.set_makefile(self.die_stub()) result = self.svc("u2a", "deploy", "--from", "local", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("已回退上一版本", result.stdout) self.assertTrue(self.running("u2a"), "回退须按原期望态恢复运行") pid2 = self.pid_of("u2a") self.assertNotEqual(pid2, pid) self.assertIn("current/bin/agent-system", self.cmdline_of(pid2)) self.assertEqual(self.active_bin("u2a").read_bytes(), v1_bin, "回退后制品=v1") hist = (self.runtime("u2a") / "history.jsonl").read_text() self.assertIn('"event":"deploy"', hist) self.assertIn("rollback", hist) self.assertIsNone(self.journal("u2a")) self.svc("u2a", "stop") # ---- T2：构建失败 → 旧服务保持运行、current 不动 ---- def test_U2_build_failure_keeps_old_running(self): self.seed_agent("u2b", 9462) self.set_makefile(self.stub) self.svc("u2b", "deploy", "--from", "local") cur1 = self.cur_release("u2b") pid = self.pid_of("u2b") (self.root / "Makefile").write_text("build:\n\texit 1\n") result = self.svc("u2b", "deploy", "--from", "local", check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("构建失败", result.stdout) self.assertTrue(self.running("u2b"), "构建失败旧服务必须保持运行") self.assertEqual(self.pid_of("u2b"), pid) self.assertEqual(self.cur_release("u2b"), cur1, "current 指针不动") self.svc("u2b", "stop") # ---- T3：中断切换（journal switched）→ start 入口回退并恢复 ---- def test_U2_journal_interrupt_resolved_on_start(self): self.seed_agent("u2c", 9463) self.set_makefile(self.stub) self.svc("u2c", "deploy", "--from", "local") v1 = self.cur_release("u2c") self.svc("u2c", "stop") # 伪造中断于「指针已切」：current → 假 release，journal switched fake = self.runtime("u2c") / "releases" / "fake-interrupt" (fake / "bin").mkdir(parents=True) shutil.copy2(self.die_stub(), fake / "bin" / "agent-system") (self.runtime("u2c") / "current").unlink() (self.runtime("u2c") / "current").symlink_to("releases/fake-interrupt") (self.runtime("u2c") / "deploy.journal").write_text(json.dumps({ "phase": "switched", "new": "fake-interrupt", "prev": v1, "desired": "running", "ts": "2026-10-12T00:00:00Z"})) result = self.svc("u2c", "start") self.assertIn("中断的切换", result.stdout) self.assertIn("已回退上一版本", result.stdout) self.assertTrue(self.running("u2c")) self.assertEqual(self.cur_release("u2c"), v1) self.assertIsNone(self.journal("u2c")) self.svc("u2c", "stop") # ---- T4：schema 前进 → 回退阻断，不触碰业务库 ---- def test_U2_schema_advance_blocks_rollback(self): self.seed_agent("u2d", 9464) self.set_makefile(self.stub) self.svc("u2d", "deploy", "--from", "local") v1 = self.cur_release("u2d") self.svc("u2d", "stop") # v1 提交时点 schema=3；库被「新制品」推进到 5 rf = self.runtime("u2d") / "releases" / v1 / "release.json" data = json.loads(rf.read_text()) data["schema_at_commit"] = "3" rf.write_text(json.dumps(data)) self.bump_schema("u2d", 5) fake = self.runtime("u2d") / "releases" / "fake-advance" (fake / "bin").mkdir(parents=True) shutil.copy2(self.die_stub(), fake / "bin" / "agent-system") (self.runtime("u2d") / "current").unlink() (self.runtime("u2d") / "current").symlink_to("releases/fake-advance") (self.runtime("u2d") / "deploy.journal").write_text(json.dumps({ "phase": "switched", "new": "fake-advance", "prev": v1, "desired": "running", "ts": "2026-10-12T00:00:00Z"})) result = self.svc("u2d", "start", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("禁止自动回退", result.stdout) self.assertFalse(self.running("u2d"), "阻断后保持停止") j = self.journal("u2d") self.assertEqual(j["phase"], "blocked") # 业务库不被回写：schema 仍是 5 import sqlite3 con = sqlite3.connect(str(self.runtime("u2d") / "data" / "agent-system.db")) self.assertEqual( con.execute("select max(id) from schema_migrations").fetchone()[0], 5) con.close() # blocked 下普通动作被拒；deploy 是解除路径（此处仅验证拒绝语义） result = self.svc("u2d", "restart", check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("切换被阻断", result.stdout) # ---- T5：legacy 布局未迁移 → 裸 start/restart 仍走原路径（兼容） ---- def test_U2_fault_after_switch_rolls_back_on_next_start(self): self.seed_agent("u2f", 9466) self.set_makefile(self.stub) self.svc("u2f", "deploy", "--from", "local") v1 = self.cur_release("u2f") self.svc("u2f", "stop") env = dict(self.env) env["SVC_TEST_FAULTS"] = "1" env["SVC_FAULT_AT"] = "switched" result = self.svc("u2f", "deploy", "--from", "local", check=False, env=env) self.assertEqual(result.returncode, 86, result.stdout) self.assertIn("故障注入于 switched", result.stdout) result = self.svc("u2f", "start") self.assertIn("已回退上一版本", result.stdout) self.assertEqual(self.cur_release("u2f"), v1) self.assertTrue(self.running("u2f")) self.svc("u2f", "stop") def test_U2_health_timeout_rolls_back(self): self.seed_agent("u2g", 9467) self.set_makefile(self.stub) self.svc("u2g", "deploy", "--from", "local") v1 = self.cur_release("u2g") good = self.active_bin("u2g").read_bytes() self.set_makefile(self.hang_stub()) result = self.svc("u2g", "deploy", "--from", "local", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("健康检查未通过", result.stdout) self.assertIn("已回退上一版本", result.stdout) self.assertEqual(self.cur_release("u2g"), v1) self.assertEqual(self.active_bin("u2g").read_bytes(), good) self.assertTrue(self.running("u2g")) self.svc("u2g", "stop") def hang_stub(self): """identity 可用、serve 挂起且不监听，用来触发健康检查超时。""" p = self.root / "stub-agent-system-hang" if not p.exists(): stub = tsr.STUB head, sep, tail = stub.partition(" serve)") assert sep, "STUB 缺 serve 分支锚点" _, sep2, tail2 = tail.partition(' wait "$child"\n ;;') assert sep2, "STUB 缺 serve 分支结尾锚点" p.write_text(head + " serve)\n shift\n sleep 60\n exit 0 ;;" + tail2) p.chmod(0o755) return p def test_U2_legacy_layout_bare_lifecycle_still_works(self): self.seed_agent("u2e", 9465) self.start_ok("u2e") self.assertFalse((self.runtime("u2e") / "current").exists()) self.assertTrue((self.runtime("u2e") / "bin" / "agent-system").exists()) self.svc("u2e", "stop") self.start_ok("u2e") self.assertTrue(self.running("u2e")) self.svc("u2e", "stop") if __name__ == "__main__": unittest.main()