"""svc.sh rdm-admin：RDM 域管理员台账查看/受信移交（2026-10-13 8788 dev 锁死复现后）。 背景：U7 验收在共享 dev runtime 执行 -bootstrap-admin u7-admin，把 rdm.db rdm_admins 单行台账登记为测试账号，操作者（基座超管）依 KTD4 不自动获得 RDM 域管理权—— 系统管理「RDM 角色权限」全 tab 400。本套覆盖： - show：台账行输出；台账管理员不在基座 users 表时警示（残留占用可见性） - transfer：受信单行改写（added_by=svc:rdm-admin:）；幂等；typo 防线 （目标不在基座 users 表拒绝）；空台账拒绝（引导走 -bootstrap-admin） - 绑定合同：无 agent-system.db/rdm.db 拒绝；未知子命令 exit 2 - status：rdm.db 存在时输出 RDM 域管理员行与残留警示 夹具复用 tests/test_svc_recover.py（桩不需要运行——台账操作是 SQLite 直读直写）。 """ import importlib.util import shutil import sqlite3 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) def sqlite_exec(path, statements): con = sqlite3.connect(path) for stmt in statements: con.execute(stmt) con.commit() con.close() class RdmAdminTests(tsr.SvcAutoRecoverTests): def seed_dbs(self, name, ledger_owner="u7-admin", base_owners=("weichen", "u7-admin")): data = self.runtime(name) / "data" data.mkdir(parents=True, exist_ok=True) base = data / "agent-system.db" sqlite_exec(base, [ "CREATE TABLE users (id TEXT PRIMARY KEY, owner TEXT NOT NULL UNIQUE," " name TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL," " password_hash TEXT NOT NULL DEFAULT '', password_set_at TEXT NOT NULL DEFAULT ''," " must_change_password INTEGER NOT NULL DEFAULT 0, is_admin INTEGER NOT NULL DEFAULT 0)", ] + ["INSERT INTO users(id,owner,created_at) VALUES('usr_%d','%s','2026-10-12T00:00:00Z')" % (i, o) for i, o in enumerate(base_owners)]) rdm = data / "rdm.db" ddl = [ "CREATE TABLE rdm_admins (id INTEGER PRIMARY KEY CHECK(id=1), owner TEXT NOT NULL UNIQUE," " added_by TEXT NOT NULL DEFAULT '', added_at TEXT NOT NULL)", ] if ledger_owner is not None: ddl.append("INSERT INTO rdm_admins(id,owner,added_by,added_at)" " VALUES(1,'%s','bootstrap','2026-10-12T14:06:02Z')" % ledger_owner) sqlite_exec(rdm, ddl) return rdm def ledger_owner(self, name): con = sqlite3.connect("file:%s?mode=ro" % (self.runtime(name) / "data" / "rdm.db"), uri=True) row = con.execute("select owner, added_by from rdm_admins where id=1").fetchone() con.close() return row # ---- show ---- def test_show_prints_ledger_and_flags_unknown_owner(self): self.seed_agent("ra1", 9461) self.seed_dbs("ra1", ledger_owner="u7-admin", base_owners=("weichen",)) out = self.svc("ra1", "rdm-admin", "show").stdout self.assertIn("RDM 域管理员 u7-admin", out) self.assertIn("不在基座用户表", out) # u7-admin 不在 users → 残留警示 def test_show_known_owner_no_warning(self): self.seed_agent("ra2", 9462) self.seed_dbs("ra2", ledger_owner="weichen") out = self.svc("ra2", "rdm-admin").stdout # 缺省子命令 = show self.assertIn("RDM 域管理员 weichen", out) self.assertNotIn("不在基座用户表", out) def test_show_empty_ledger_hint(self): self.seed_agent("ra3", 9463) self.seed_dbs("ra3", ledger_owner=None) out = self.svc("ra3", "rdm-admin").stdout self.assertIn("台账为空", out) self.assertIn("-bootstrap-admin", out) # ---- transfer ---- def test_transfer_rewrites_single_row_with_provenance(self): self.seed_agent("ra4", 9464) self.seed_dbs("ra4", ledger_owner="u7-admin") out = self.svc("ra4", "rdm-admin", "transfer", "--to", "weichen").stdout self.assertIn("受信部署通道", out) self.assertIn("移交后 weichen", out) owner, added_by = self.ledger_owner("ra4") self.assertEqual(owner, "weichen") self.assertEqual(added_by, "svc:rdm-admin:u7-admin") # 前任可追溯 def test_transfer_idempotent_when_already_owner(self): self.seed_agent("ra5", 9465) self.seed_dbs("ra5", ledger_owner="weichen") out = self.svc("ra5", "rdm-admin", "transfer", "--to", "weichen").stdout self.assertIn("已是 weichen", out) def test_transfer_rejects_owner_missing_from_base_users(self): self.seed_agent("ra6", 9466) self.seed_dbs("ra6", ledger_owner="weichen") proc = self.svc("ra6", "rdm-admin", "transfer", "--to", "nobody", check=False) self.assertEqual(proc.returncode, 1, proc.stdout) self.assertIn("不在基座用户表", proc.stdout) self.assertEqual(self.ledger_owner("ra6")[0], "weichen") # 台账未被动 def test_transfer_rejects_empty_ledger(self): self.seed_agent("ra7", 9467) self.seed_dbs("ra7", ledger_owner=None) proc = self.svc("ra7", "rdm-admin", "transfer", "--to", "weichen", check=False) self.assertEqual(proc.returncode, 1, proc.stdout) self.assertIn("-bootstrap-admin", proc.stdout) # ---- 绑定合同与用法 ---- def test_refuses_when_rdm_db_missing(self): self.seed_agent("ra8", 9468) self.seed_dbs("ra8") # 建基座库后删 rdm.db (self.runtime("ra8") / "data" / "rdm.db").unlink() proc = self.svc("ra8", "rdm-admin", check=False) self.assertEqual(proc.returncode, 1, proc.stdout) self.assertIn("无 RDM 域库", proc.stdout) def test_refuses_when_base_db_missing(self): self.seed_agent("ra9", 9469) proc = self.svc("ra9", "rdm-admin", check=False) self.assertEqual(proc.returncode, 1, proc.stdout) self.assertIn("既有库不存在", proc.stdout) def test_unknown_subcommand_usage_exit_2(self): self.seed_agent("raa", 9470) proc = self.svc("raa", "rdm-admin", "bogus", check=False) self.assertEqual(proc.returncode, 2, proc.stdout) self.assertIn("用法", proc.stdout) # ---- status 巡检行 ---- def test_status_lists_rdm_admin_row_and_warning(self): self.seed_agent("rab", 9471) self.seed_dbs("rab", ledger_owner="u7-admin", base_owners=("weichen",)) out = self.svc("rab", "status").stdout self.assertIn("RDM 域管理员 u7-admin", out) self.assertIn("不在基座用户表", out) def test_status_silent_without_rdm_db(self): self.seed_agent("rac", 9472) self.seed_dbs("rac") (self.runtime("rac") / "data" / "rdm.db").unlink() out = self.svc("rac", "status").stdout self.assertNotIn("RDM 域管理员", out) # ---- help 可发现性 ---- def test_help_topic_lists_rdm_admin(self): out = self.svc("help", "rdm-admin").stdout self.assertIn("锁死恢复通道", out) self.assertIn("rdm.perm.transfer", out) if __name__ == "__main__": unittest.main()