agent-exec-transportandCursor
6023900eb6
svc.sh adopt always printed db_marker=established, even when the installed artifact lacked identity capability and the marker stayed pending. It now reads the marker back and warns when identity was not established. The deploy identity gate reported a pre-v27 database (identity table missing) as a foreign row or mismatch. It now explains that the schema predates the identity table and gives the backup, stop, migrate and redeploy steps. It still refuses before stopping the old service and never migrates on its own. Why: both messages misled the 2026-10-21 GenUI prod upgrade. Evidence: agent-exec unit dev-svc-identity-adopt-gate-messages (cursor, attempt 1, terminal=success); python3 -m unittest tests.test_svc_identity: 31 tests OK (1 skipped). Impact: svc.sh operator messages for adopt and the deploy identity gate; no change to success paths. Refs: Dev task fix/genui-prod-upgrade-followups. Co-authored-by: Cursor <cursoragent@cursor.com>
537 lines
25 KiB
Python
537 lines
25 KiB
Python
"""svc.sh 实例身份回归（U1 设计 §7 T1–T6 + I3 DB marker，manifest 治理）。
|
||
|
||
覆盖：连续裸启停身份稳定（T1）、只改端口锚点不动（T2）、数据根漂移拒绝（T3）、
|
||
迁移窗口/损坏阻塞不猜测（T4）、marker 缺失/外宾/manifest 丢失均为事件（T5）、
|
||
cwd 无关性（T6）、SQLite DB marker ensure/错 id 阻塞（I3 桩路径）、原子写崩溃窗口。
|
||
桩二进制与夹具复用 tests/test_svc_recover.py。真实 PG 身份专项（跨库）在
|
||
SVC_TEST_PG=1 且具备环境时执行；否则 test_Z_real_pg_not_run 显式 NOT_RUN。
|
||
"""
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import shutil
|
||
import sqlite3
|
||
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")
|
||
|
||
# ---- I3 DB marker（SQLite 桩路径；与真实 PG 跨库专项分离） ----
|
||
|
||
def db_identity_row(self, name):
|
||
m = self.manifest(name)
|
||
if not m:
|
||
return None
|
||
db = Path(m["identity"]["data_root"]) / "agent-system.db"
|
||
if not db.exists():
|
||
return None
|
||
con = sqlite3.connect(str(db))
|
||
try:
|
||
rows = list(con.execute(
|
||
"SELECT instance_id, dbref FROM agent_system_identity"))
|
||
finally:
|
||
con.close()
|
||
return rows
|
||
|
||
# ---- U1 批量隔离、候选门禁与启动语义（2026-10-12 rdm1/rdmcand 事故回归） ----
|
||
|
||
def noid_stub(self):
|
||
"""自包含「无 identity」桩：复用 recover STUB，把 identity 分支换成 2026-10-12
|
||
RDM 组合产物形态——identity 子命令落入服务路径、被身份门禁拒绝，报错文本
|
||
恰含「identity ensure」字样（旧探测 grep ensure 的误报源）。不 exec 转发：
|
||
serve 进程 cmdline 必须保留实例 bin 路径（pidcheck 契约与真实产物一致）。"""
|
||
p = self.root / "stub-agent-system-noid"
|
||
if not p.exists():
|
||
stub = tsr.STUB
|
||
head, sep, tail = stub.partition(" identity)")
|
||
assert sep, "STUB 缺 identity 分支锚点"
|
||
_, sep2, tail2 = tail.partition(" exit $? ;;")
|
||
assert sep2, "STUB 缺 identity 分支结尾锚点"
|
||
noid_case = (
|
||
" identity)\n"
|
||
" echo \"system identity: system identity missing: agent_system_identity 无行"
|
||
"（先 svc init / identity ensure，禁止静默建空身份）\" >&2\n"
|
||
" exit 1 ;;")
|
||
body = head + noid_case + tail2
|
||
body = body.replace(
|
||
'\'{"protocol":1,"product":"agent-system","version":"stub","identity":{"ops":["ensure","verify"],"open":"existing-no-migrate","dsn":"env"},"cli":{"unknownSubcommand":"reject"}}\'',
|
||
'"not-json"',
|
||
)
|
||
p.write_text(body)
|
||
p.chmod(0o755)
|
||
return p
|
||
|
||
def install_noid_bin(self, name):
|
||
"""把实例磁盘产物换成无 identity 形态（rdm1/rdmcand 事故现场的制品状态）。"""
|
||
shutil.copy2(self.noid_stub(), self.runtime(name) / "bin" / "agent-system")
|
||
|
||
def set_kv(self, name, key, value):
|
||
f = self.runtime(name) / "svc.env"
|
||
lines = f.read_text().splitlines()
|
||
out, hit = [], False
|
||
for l in lines:
|
||
if l.startswith(key + "="):
|
||
out.append("%s=%s" % (key, value)); hit = True
|
||
else:
|
||
out.append(l)
|
||
if not hit:
|
||
out.append("%s=%s" % (key, value))
|
||
f.write_text("\n".join(out) + "\n")
|
||
|
||
def serve_record(self):
|
||
f = self.trace / "serve.txt"
|
||
return f.read_text() if f.exists() else ""
|
||
|
||
def clear_serve_record(self):
|
||
(self.trace / "serve.txt").unlink(missing_ok=True)
|
||
|
||
def git_ok(self, cwd, *args):
|
||
proc = subprocess.run(["git", *args], cwd=cwd, env=self.env, text=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60)
|
||
self.assertEqual(proc.returncode, 0, proc.stdout)
|
||
|
||
def git_sha(self, cwd, rev):
|
||
proc = subprocess.run(["git", "rev-parse", "--short", rev], cwd=cwd,
|
||
env=self.env, text=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=30)
|
||
return proc.stdout.strip()
|
||
|
||
def _make_track_repo(self, artifact, target, with_web=False):
|
||
"""双提交源仓夹具（旧/新两版）：Makefile 产物 = artifact 复制为 target。"""
|
||
remote, work = self.root / "trackremote.git", self.root / "trackwork"
|
||
for p in (remote, work):
|
||
if p.exists():
|
||
shutil.rmtree(p)
|
||
self.git_ok(self.root, "init", "--bare", "--initial-branch=main", str(remote))
|
||
self.git_ok(self.root, "init", "--initial-branch=main", str(work))
|
||
(work / "Makefile").write_text(
|
||
"build:\n\tmkdir -p bin\n\tcp %s bin/%s\n" % (artifact, target))
|
||
if with_web:
|
||
(work / "web").mkdir()
|
||
(work / "web" / "index.html").write_text("<html>rdm</html>\n")
|
||
self.git_ok(work, "add", ".")
|
||
self.git_ok(work, "commit", "-m", "track v1")
|
||
self.git_ok(work, "remote", "add", "origin", str(remote))
|
||
self.git_ok(work, "push", "-u", "origin", "main")
|
||
(work / "v2.marker").write_text("v2\n")
|
||
self.git_ok(work, "add", ".")
|
||
self.git_ok(work, "commit", "-m", "track v2")
|
||
self.git_ok(work, "push")
|
||
return remote, self.git_sha(work, "HEAD~1"), self.git_sha(work, "HEAD")
|
||
|
||
def make_src_repo(self):
|
||
"""dev 轨道：产物 = 可用 identity 桩；预克隆进 SVC_SRC_DIR（srccache）。"""
|
||
remote, old, new = self._make_track_repo(self.stub, "agent-system")
|
||
cache = self.root / "srccache"
|
||
if cache.exists():
|
||
shutil.rmtree(cache)
|
||
self.git_ok(self.root, "clone", str(remote), str(cache))
|
||
return old, new
|
||
|
||
def make_rdm_repo(self, artifact):
|
||
"""rdm 轨道：产物 = 指定桩（可注入无 identity 形态）。"""
|
||
return self._make_track_repo(artifact, "agent-system-rdm", with_web=True)
|
||
|
||
def rdm_env(self, remote):
|
||
env = dict(self.env)
|
||
env.update({"SVC_RDM_SRC_DIR": str(self.root / "rdmcache"),
|
||
"SVC_RDM_ORIGIN": str(remote)})
|
||
return env
|
||
|
||
def test_U1_update_refuses_identityless_candidate_keeps_old_state(self):
|
||
"""manifest 治理实例 × 无 identity 候选（RDM 旧形态）：停服前拒绝——
|
||
旧服务继续运行、旧制品与登记不变（2026-10-12 rdm1/rdmcand 反例：停服→
|
||
覆盖→启动即拒，只剩混合状态）。"""
|
||
remote, old_sha, new_sha = self.make_rdm_repo(self.noid_stub())
|
||
self.seed_agent("rdx", 9451)
|
||
self.start_ok("rdx") # 已建立身份，有能力制品在运行
|
||
self.set_kv("rdx", "source", "rdm@%s" % old_sha)
|
||
pid = self.pid_of("rdx")
|
||
bin_before = (self.runtime("rdx") / "bin" / "agent-system").read_bytes()
|
||
|
||
result = self.svc("update", "rdx", check=False, env=self.rdm_env(remote))
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("候选产物不支持 identity 管理", result.stdout)
|
||
# 拒绝发生在停旧服务之前：进程、制品、登记全部保持
|
||
self.assertTrue(self.running("rdx"), "拒绝不得停掉可用旧服务")
|
||
self.assertEqual(self.pid_of("rdx"), pid)
|
||
self.assertEqual((self.runtime("rdx") / "bin" / "agent-system").read_bytes(),
|
||
bin_before, "旧制品不得被覆盖")
|
||
self.assertEqual(self.kv("rdx", "source"), "rdm@%s" % old_sha)
|
||
hist = (self.runtime("rdx") / "history.jsonl").read_text()
|
||
self.assertIn('"event":"update"', hist)
|
||
self.assertIn('"result":"fail"', hist)
|
||
self.svc("rdx", "stop")
|
||
|
||
def test_U1_update_ensures_missing_row_before_stop(self):
|
||
"""已建立身份后删行：update 在停服前拒绝，不自动补行，旧进程保持。
|
||
补行只走 adopt --confirm。"""
|
||
old_sha, new_sha = self.make_src_repo()
|
||
self.seed_agent("rowx", 9452)
|
||
self.start_ok("rowx")
|
||
self.set_kv("rowx", "source", "dev@%s" % old_sha)
|
||
db = Path(self.manifest("rowx")["identity"]["data_root"]) / "agent-system.db"
|
||
con = sqlite3.connect(str(db))
|
||
con.execute("DELETE FROM agent_system_identity")
|
||
con.commit()
|
||
con.close()
|
||
|
||
self.clear_serve_record()
|
||
result = self.svc("update", "rowx", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("候选无法验证已建立的 DB 身份", result.stdout)
|
||
self.assertTrue(self.running("rowx"), "缺行必须在停旧服务前拒绝")
|
||
self.assertEqual(self.db_identity_row("rowx"), [], "established 不得自动补行")
|
||
self.svc("rowx", "stop")
|
||
|
||
def test_U1_bare_start_missing_row_is_event_init_recovers(self):
|
||
"""普通启动只验证不补行：DB 行缺失 → 拒绝且不建空行（缺行即事件）；
|
||
显式 init 补齐后恢复。"""
|
||
self.seed_agent("rowm", 9453)
|
||
self.start_ok("rowm")
|
||
iid = self.manifest("rowm")["instance_id"]
|
||
self.svc("rowm", "stop")
|
||
db = Path(self.manifest("rowm")["identity"]["data_root"]) / "agent-system.db"
|
||
con = sqlite3.connect(str(db))
|
||
con.execute("DELETE FROM agent_system_identity")
|
||
con.commit()
|
||
con.close()
|
||
|
||
result = self.svc("rowm", "start", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("DB 身份校验失败", result.stdout)
|
||
self.assertFalse(self.running("rowm"))
|
||
self.assertEqual(self.db_identity_row("rowm"), [], "缺行不得被 start 静默补建")
|
||
|
||
refused = self.svc("rowm", "init", check=False)
|
||
self.assertNotEqual(refused.returncode, 0, refused.stdout)
|
||
self.assertIn("adopt", refused.stdout)
|
||
self.svc("rowm", "adopt", "--confirm", iid)
|
||
self.assertEqual(self.db_identity_row("rowm"), [(iid, "sqlite")])
|
||
self.start_ok("rowm")
|
||
self.svc("rowm", "stop")
|
||
|
||
def test_U1_capability_probe_no_false_positive_no_hang(self):
|
||
"""能力探测无副作用：含 ensure 字样的服务态门禁报错不算能力（旧实现
|
||
误报点）；identity 即挂死的产物探测有界返回无能力。"""
|
||
def probe(binpath):
|
||
# svc.sh 顶层启用 errexit：source 后裸调用失败函数会杀掉探针 shell，
|
||
# 用 if 条件上下文（errexit 豁免）取回真实返回码。
|
||
return subprocess.run(
|
||
["bash", "-c",
|
||
'source "%s" >/dev/null 2>&1; ENV_BIN="%s"; '
|
||
'if svc_bin_has_identity; then echo rc=0; else echo rc=$?; fi'
|
||
% (self.root / "scripts" / "svc.sh", binpath)],
|
||
cwd=self.root, env=self.env, text=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||
timeout=30).stdout.strip()
|
||
|
||
self.assertIn("rc=0", probe(self.stub), "identity 桩应判有能力")
|
||
self.assertIn("rc=1", probe(self.noid_stub()),
|
||
"报错文本含 ensure 不得误报为有能力")
|
||
hang = self.root / "stub-hang"
|
||
hang.write_text(
|
||
'#!/usr/bin/env bash\ncase "${1:-}" in version) sleep 30 ;; '
|
||
'*) exec "%s" "$@" ;; esac\n' % self.stub)
|
||
hang.chmod(0o755)
|
||
self.assertIn("rc=1", probe(hang), "探测挂死须有界返回无能力")
|
||
|
||
def test_U1_adopt_identityless_artifact_reports_pending(self):
|
||
"""已装制品无 identity 能力：adopt 的 ensure 静默跳过，marker 留在 pending——
|
||
输出必须如实报 pending，不得宣称 db_marker=established。"""
|
||
self.seed_agent("adx", 9454)
|
||
self.start_ok("adx")
|
||
iid = self.manifest("adx")["instance_id"]
|
||
self.svc("adx", "stop")
|
||
self.install_noid_bin("adx")
|
||
result = self.svc("adx", "adopt", "--confirm", iid)
|
||
self.assertNotIn("db_marker=established", result.stdout)
|
||
self.assertIn("db_marker=pending", result.stdout)
|
||
self.assertEqual(self.manifest("adx")["identity"]["db_marker"], "pending")
|
||
|
||
def test_U1_update_pre_v27_schema_refused_with_migrate_guidance(self):
|
||
"""候选有 identity 能力但库 schema 早于身份表：停服前拒绝，给出
|
||
备份→stop→migrate→再部署的指引，而非「外宾行」误导；不自动迁移。"""
|
||
cand = self.root / "stub-agent-system-noschema"
|
||
cand.write_text(
|
||
'#!/usr/bin/env bash\ncase "${1:-}" in identity) '
|
||
'echo "identity ${2:-ensure}: identity table missing" >&2; exit 1 ;; '
|
||
'*) exec "%s" "$@" ;; esac\n' % self.stub)
|
||
cand.chmod(0o755)
|
||
remote, old_sha, _ = self._make_track_repo(cand, "agent-system")
|
||
self.git_ok(self.root, "clone", str(remote), str(self.root / "srccache"))
|
||
self.seed_agent("scx", 9455)
|
||
self.start_ok("scx")
|
||
self.set_kv("scx", "source", "dev@%s" % old_sha)
|
||
mpath = self.runtime("scx") / "manifest.json"
|
||
m = json.loads(mpath.read_text(encoding="utf-8"))
|
||
m["identity"]["db_marker"] = "pending"
|
||
mpath.write_text(json.dumps(m, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
pid = self.pid_of("scx")
|
||
bin_before = (self.runtime("scx") / "bin" / "agent-system").read_bytes()
|
||
|
||
result = self.svc("update", "scx", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("migrate --data", result.stdout)
|
||
self.assertIn("AGENT_SYSTEM_DSN", result.stdout)
|
||
self.assertNotIn("外宾行", result.stdout)
|
||
self.assertTrue(self.running("scx"), "拒绝必须发生在停旧服务之前")
|
||
self.assertEqual(self.pid_of("scx"), pid)
|
||
self.assertEqual((self.runtime("scx") / "bin" / "agent-system").read_bytes(),
|
||
bin_before, "旧制品不得被覆盖")
|
||
self.svc("scx", "stop")
|
||
|
||
def test_I3_start_ensures_sqlite_identity(self):
|
||
self.seed_agent("idb1", 9441)
|
||
self.start_ok("idb1")
|
||
iid = self.manifest("idb1")["instance_id"]
|
||
rows = self.db_identity_row("idb1")
|
||
self.assertEqual(rows, [(iid, "sqlite")], rows)
|
||
self.svc("idb1", "stop")
|
||
self.start_ok("idb1") # ensure 幂等 + verify
|
||
self.assertEqual(self.db_identity_row("idb1"), [(iid, "sqlite")])
|
||
self.svc("idb1", "stop")
|
||
|
||
def test_I3_wrong_db_identity_blocks_start(self):
|
||
self.seed_agent("idb2", 9442)
|
||
self.start_ok("idb2")
|
||
iid = self.manifest("idb2")["instance_id"]
|
||
self.svc("idb2", "stop")
|
||
db = Path(self.manifest("idb2")["identity"]["data_root"]) / "agent-system.db"
|
||
con = sqlite3.connect(str(db))
|
||
con.execute("DELETE FROM agent_system_identity")
|
||
con.execute(
|
||
"INSERT INTO agent_system_identity(instance_id,dbref,schema_version,created_at)"
|
||
" VALUES(?,?,?,?)",
|
||
("svc-foreign-deadbeef", "sqlite", 26, "2026-01-01T00:00:00Z"))
|
||
con.commit()
|
||
con.close()
|
||
result = self.svc("idb2", "start", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(any(k in result.stdout for k in ("DB 身份", "ensure", "mismatch", "校验失败")),
|
||
result.stdout)
|
||
self.assertFalse(self.running("idb2"))
|
||
# 外宾行不得被 start 静默改写
|
||
self.assertEqual(self.db_identity_row("idb2"),
|
||
[("svc-foreign-deadbeef", "sqlite")])
|
||
self.assertNotEqual(iid, "svc-foreign-deadbeef")
|
||
|
||
# ---- 真实 PG 专项：跨库身份；缺环境显式 NOT_RUN ----
|
||
|
||
def test_Z_real_pg_not_run(self):
|
||
if os.environ.get("SVC_TEST_PG") == "1" and os.environ.get("AGENT_PG_TEST_DSN"):
|
||
self.skipTest("真 PG 专项应由隔离套件执行（本桩文件不连真实 PG）")
|
||
self.skipTest("真实 PG 身份专项 NOT_RUN：需真实实例（SVC_TEST_PG=1 + AGENT_PG_TEST_DSN）")
|
||
|
||
|
||
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()
|