Agent-System-Dev/tests/test_svc_rdm_admin.py

"""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()