weichen
7719e807b8
Why: U7 acceptance bootstrapped u7-admin into the shared dev runtime's singleton rdm_admins ledger, locking the operator out of RDM admin pages (rdm.role.list 400); recovery required forensics. KTD4 keeps base isAdmin to audit-view only, so an operator-side trusted channel is needed. Evidence: python3 -m unittest tests.test_svc_rdm_admin -v (25 OK, SELF); live smoke on runtime/dev: show/status/typo-guard outputs verified. Impact: new svc.sh <env> rdm-admin [show|transfer --to x] command, help topic, status ledger line with stale-owner warning; no behavior change for existing commands. Refs: task card .task/feature-rdm-admin-lockout-recovery.md R1/R2
168 lines
7.3 KiB
Python
168 lines
7.3 KiB
Python
"""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:<old>）；幂等；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()
|