weichenandCursor
451c95be70
Why: probing identity by running the subcommand still launched old RDM builds, and ensure on a missing database created an empty one. A batch update also inherited the caller's AGENT_SYSTEM_* variables. Evidence: python3 -m unittest tests.test_svc_identity tests.test_svc_recover passed (41 tests, 1 skipped). tests.test_svc_lifecycle_isolation caller-env and restart cases passed. Impact: capability probes use version --json only. Established instances refuse a missing identity row until adopt --confirm. update and recover run each service in a clean environment. serve reads a 0600 serve.env and does not inherit the caller environment. Backend and documents pins move to the D1 identity and recovery-plan commits. Refs: recovery plan r3.1 D1; Backend feature/d1-identity-open; RDM feature/d1-version-json Co-authored-by: Cursor <cursoragent@cursor.com>
492 lines
22 KiB
Python
492 lines
22 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_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()
|