"""svc.sh 实例身份回归（U1 设计 §7 T1–T6，manifest 治理）。 覆盖：连续裸启停身份稳定（T1）、只改端口锚点不动（T2）、数据根漂移拒绝（T3）、 迁移窗口/损坏阻塞不猜测（T4）、marker 缺失/外宾/manifest 丢失均为事件（T5）、 cwd 无关性（T6）、原子写崩溃窗口单元级（原文件不受损）。 桩二进制与夹具复用 tests/test_svc_recover.py。真实 PG 身份专项（DB marker SQL 层） NOT_RUN：需真实实例，SVC_TEST_PG=1 且具备环境时另行执行，本文件不模拟。 """ import importlib.util import json import os import shutil import subprocess import tempfile import unittest from pathlib import Path SOURCE = Path(__file__).resolve().parents[1] SVC_SCRIPT = SOURCE / "scripts" / "svc.sh" _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) class IdentityTests(tsr.SvcAutoRecoverTests): def setUp(self): super().setUp() shutil.copy2(SOURCE / "scripts" / "svc_manifest.py", self.root / "scripts" / "svc_manifest.py") # ---- 夹具 ---- def manifest(self, name): p = self.runtime(name) / "manifest.json" return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None def marker(self, name): m = self.manifest(name) if not m: return None p = Path(m["identity"]["data_root"]) / ".agentx" / "data-marker.json" return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None def running(self, name): pid = self.pid_of(name) if not pid: return False try: os.kill(pid, 0) return True except OSError: return False # ---- T1 连续裸启停：同一身份与数据根 ---- def test_T1_bare_restart_stable_identity(self): self.seed_agent("id1", 9431) self.start_ok("id1") m1 = self.manifest("id1") self.assertIsNotNone(m1, "start 应完成 manifest 迁移") iid = m1["instance_id"] self.assertTrue(iid.startswith("svc-id1-"), iid) mk = self.marker("id1") self.assertEqual(mk["instance_id"], iid) data_before = m1["identity"]["data_root"] exec_before = m1["kv"]["exec"] self.svc("id1", "stop") self.start_ok("id1") m2 = self.manifest("id1") self.assertEqual(m2["instance_id"], iid, "裸启停不得换身份") self.assertEqual(m2["identity"]["data_root"], data_before) self.assertEqual(m2["kv"]["exec"], exec_before, "裸启停不得漂移 exec（数据/插件根）") self.assertGreaterEqual(m2["revision"], m1["revision"], "整份提交 revision 单调") self.assertTrue(self.running("id1")) self.svc("id1", "stop") # ---- T2 只改端口：数据/DB 锚点不动 ---- def test_T2_port_change_keeps_anchors(self): self.seed_agent("id2", 9432) self.start_ok("id2") anchor = self.manifest("id2")["identity"] self.svc("id2", "stop") result = self.svc("id2", "start", "--port", "9433") self.assertIn("已启动", result.stdout) m2 = self.manifest("id2") self.assertEqual(m2["identity"]["data_root"], anchor["data_root"], "改端口不得动数据根") self.assertEqual(m2["identity"]["db_mode"], anchor["db_mode"]) self.assertEqual(m2["identity"]["db_ref"], anchor["db_ref"]) self.assertIn("9433", m2["kv"]["addr"]) self.svc("id2", "stop") # ---- T3 数据根漂移：拒绝启动，不新建空系统 ---- def test_T3_data_drift_rejected(self): self.seed_agent("id3", 9434) self.start_ok("id3") self.svc("id3", "stop") other = self.root / "other-data" other.mkdir() result = self.svc("id3", "start", "--data", str(other), check=False) self.assertNotEqual(result.returncode, 0, "数据根漂移必须拒绝") self.assertTrue(any(k in result.stdout for k in ("身份校验失败", "数据根漂移")), result.stdout) self.assertFalse(self.running("id3"), "拒绝后不得有服务进程") self.assertEqual(self.manifest("id3")["identity"]["data_root"], str(self.runtime("id3") / "data"), "锚点不得被污染") # ---- T4 迁移窗口与损坏：不猜测 ---- def test_T4a_leftover_new_file_ignored(self): self.seed_agent("id4", 9435) # 模拟 rename 前崩溃残留：.new 半成品必须被忽略/覆盖，不影响迁移结果 (self.runtime("id4") / "manifest.json.new").write_text("{ truncated") self.start_ok("id4") m = self.manifest("id4") self.assertIsNotNone(m) self.assertEqual(m["schema"], 1) self.svc("id4", "stop") def test_T4b_corrupt_manifest_blocks(self): self.seed_agent("id5", 9436) self.start_ok("id5") self.svc("id5", "stop") (self.runtime("id5") / "manifest.json").write_text("{ not json") result = self.svc("id5", "start", check=False) self.assertNotEqual(result.returncode, 0, "损坏 manifest 必须阻塞而非猜测重建") self.assertIn("损坏", result.stdout) self.assertFalse(self.running("id5")) # ---- T5 marker/manifest 事件：缺一即阻塞，不自动新建 ---- def test_T5a_missing_marker_blocks(self): self.seed_agent("id6", 9437) self.start_ok("id6") self.svc("id6", "stop") m = self.manifest("id6") mk = Path(m["identity"]["data_root"]) / ".agentx" / "data-marker.json" mk.unlink() result = self.svc("id6", "start", check=False) self.assertNotEqual(result.returncode, 0, "marker 缺失即事件") self.assertIn("marker", result.stdout) self.assertFalse(self.running("id6")) def test_T5b_foreign_marker_blocks(self): self.seed_agent("id7", 9438) self.start_ok("id7") self.svc("id7", "stop") m = self.manifest("id7") mk = Path(m["identity"]["data_root"]) / ".agentx" / "data-marker.json" data = json.loads(mk.read_text(encoding="utf-8")) data["instance_id"] = "svc-other-zzzzzzzzzzzz" mk.write_text(json.dumps(data), encoding="utf-8") result = self.svc("id7", "start", check=False) self.assertNotEqual(result.returncode, 0, "marker 身份不符即事件") self.assertFalse(self.running("id7")) def test_T5c_lost_manifest_blocks_not_rebuild(self): self.seed_agent("id8", 9439) self.start_ok("id8") m0 = self.manifest("id8") iid = m0["instance_id"] marker_path = Path(m0["identity"]["data_root"]) / ".agentx" / "data-marker.json" self.svc("id8", "stop") (self.runtime("id8") / "manifest.json").unlink() result = self.svc("id8", "start", check=False) # manifest 丢失后 marker 仍在：重建会撞归属冲突 → 中止（I4），不得 legacy 复活 self.assertNotEqual(result.returncode, 0, "丢 manifest 不得静默重建/legacy 复活") self.assertTrue(any(k in result.stdout for k in ("归属冲突", "身份治理冲突")), result.stdout) self.assertFalse(self.running("id8")) # marker 保持原身份，未被新随机身份覆盖 mk = json.loads(marker_path.read_text(encoding="utf-8")) self.assertEqual(mk["instance_id"], iid) # ---- T6 cwd 无关：从别处调用身份不变 ---- def test_T6_cwd_independent_identity(self): self.seed_agent("id9", 9440) self.start_ok("id9") iid = self.manifest("id9")["instance_id"] self.svc("id9", "stop") elsewhere = tempfile.mkdtemp(prefix="svc-elsewhere-") self.addCleanup(shutil.rmtree, elsewhere, True) proc = subprocess.run( ["bash", str(self.root / "scripts" / "svc.sh"), "id9", "start"], cwd=elsewhere, env=self.env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=120) self.assertEqual(proc.returncode, 0, proc.stdout) self.assertIn("已启动", proc.stdout) self.assertEqual(self.manifest("id9")["instance_id"], iid, "换 cwd 不得换身份") self.svc("id9", "stop") # ---- 真实 PG 专项：显式 NOT_RUN 标注（不模拟） ---- def test_Z_real_pg_not_run(self): # DB marker（agent_system_identity 表）与跨库身份断言属真实 PG 专项， # 由隔离真实实例执行；本桩套件明确 NOT_RUN，不以桩冒充（设计 §7）。 self.skipTest("真实 PG 身份专项 NOT_RUN：需真实实例（SVC_TEST_PG 环境另行执行）") class AtomicWriteTests(unittest.TestCase): """崩溃窗口单元级：os.replace 失败时原文件必须完好（I3 半写不可能）。""" def test_replace_failure_keeps_original(self): spec = importlib.util.spec_from_file_location( "sm", SOURCE / "scripts" / "svc_manifest.py") sm = importlib.util.module_from_spec(spec) spec.loader.exec_module(sm) with tempfile.TemporaryDirectory() as d: path = os.path.join(d, "m.json") sm.atomic_write(path, '{"v": 1}\n') real_replace = os.replace calls = {"n": 0} def boom(a, b): calls["n"] += 1 raise OSError("simulated crash before rename completes") os.replace = boom try: try: sm.atomic_write(path, '{"v": 2}\n') except OSError: pass finally: os.replace = real_replace self.assertEqual(calls["n"], 1) with open(path, encoding="utf-8") as f: self.assertEqual(f.read(), '{"v": 1}\n', "rename 前崩溃必须保留完整旧版") if __name__ == "__main__": unittest.main()