"""svc.sh 账户 CLI 实例绑定（U1 规格 §6 T1–T5）。 覆盖：合法 manifest 下 user list 命中 data_root+open-existing（T1）、禁止覆盖 --data（T2）、删除 data 根后 OpenExisting 拒绝且不建空库（T3）、无 flock 拒绝 user/deploy（T4）。T5 身份回归由 test_svc_identity 同发现执行；此处仅冒烟。 桩与夹具复用 tests/test_svc_recover.py；manifest 辅助对齐 test_svc_identity。 """ import importlib.util import json import os import shutil 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) class UserBindingTests(tsr.SvcAutoRecoverTests): def setUp(self): super().setUp() shutil.copy2(SOURCE / "scripts" / "svc_manifest.py", self.root / "scripts" / "svc_manifest.py") self.trace.mkdir(exist_ok=True) 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 running(self, name): pid = self.pid_of(name) if not pid: return False try: os.kill(pid, 0) return True except OSError: return False def touch_db(self, name): data = Path(self.manifest(name)["identity"]["data_root"]) data.mkdir(parents=True, exist_ok=True) (data / "agent-system.db").write_bytes(b"sqlite-stub") def cli_trace(self): f = self.trace / "cli.txt" return f.read_text(encoding="utf-8") if f.exists() else "" def clear_cli_trace(self): f = self.trace / "cli.txt" if f.exists(): f.unlink() # ---- T1 合法 manifest：user list 绑定 data_root + --open-existing ---- def test_T1_user_list_binds_manifest_data(self): self.seed_agent("ub1", 9441) self.start_ok("ub1") self.svc("ub1", "stop") root = self.manifest("ub1")["identity"]["data_root"] self.touch_db("ub1") self.clear_cli_trace() result = self.svc("ub1", "user", "list") self.assertIn("op", result.stdout) trace = self.cli_trace() self.assertIn("user list", trace) self.assertIn("--open-existing", trace) self.assertIn("--data %s" % root, trace) self.assertNotIn("./data", trace) # ---- T2 禁止覆盖 --data；未绑定默认目录拒绝 ---- def test_T2_reject_override_data_flag(self): self.seed_agent("ub2", 9442) self.start_ok("ub2") self.svc("ub2", "stop") self.touch_db("ub2") other = self.root / "evil-data" other.mkdir() (other / "agent-system.db").write_bytes(b"x") result = self.svc("ub2", "user", "list", "--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.trace / "cli.txt").exists(), "拒绝后不得调用二进制") def test_T2b_reject_literal_default_data(self): d = self.runtime("ub2b") (d / "bin").mkdir(parents=True) shutil.copy2(self.stub, d / "bin" / "agent-system") (d / "svc.env").write_text("".join("%s=%s\n" % kv for kv in [ ("kind", "agent"), ("env", "ub2b"), ("pid", ""), ("addr", "127.0.0.1:9443"), ("tags", ""), ("source", "local@deadbee"), ("version", "stub"), ("started", ""), ("pidcheck", str(d / "bin" / "agent-system")), ("exec", "--addr 127.0.0.1:9443 --data ./data --plugins %s" % (d / "plugins")), ])) env = dict(self.env, SVC_UB2B_DATA="./data") result = self.svc("ub2b", "user", "list", check=False, env=env) self.assertNotEqual(result.returncode, 0) self.assertTrue(any(k in result.stdout for k in ("未绑定数据目录", "不回落")), result.stdout) # ---- T3 删除 data 根：OpenExisting 失败，不建空库 ---- def test_T3_missing_data_root_rejected(self): self.seed_agent("ub3", 9444) self.start_ok("ub3") self.svc("ub3", "stop") root = Path(self.manifest("ub3")["identity"]["data_root"]) self.touch_db("ub3") shutil.rmtree(root) self.assertFalse(root.exists()) result = self.svc("ub3", "user", "create", "--owner", "x", "--name", "x", check=False) self.assertNotEqual(result.returncode, 0) self.assertTrue(any(k in result.stdout for k in ("既有库不存在", "OpenExisting")), result.stdout) self.assertFalse(root.exists(), "拒绝后不得重建数据根") self.assertFalse((root / "agent-system.db").exists()) # ---- T4 隐藏 flock：user/deploy 拒绝；已运行服务仍在 ---- def test_T4_no_flock_rejects_user_and_deploy(self): self.seed_agent("ub4", 9445) self.start_ok("ub4") self.assertTrue(self.running("ub4")) self.touch_db("ub4") # 构造不含 flock 的 PATH：链入常用外部命令，刻意排除 flock。 bin_dir = self.root / "nobin" bin_dir.mkdir() for name in ( "bash", "python3", "python", "dirname", "basename", "tr", "cksum", "cut", "mkdir", "cat", "sed", "grep", "head", "rm", "mv", "cp", "ls", "chmod", "uname", "date", "sleep", "kill", "ps", "awk", "sort", "uniq", "env", "printf", "realpath", "readlink", "stat", "tee", "wc", "find", "xargs", "touch", "ln", "mktemp", "id", "git", "make", "tar", "gzip", "which", "command", ): src = shutil.which(name) if src: dest = bin_dir / name if not dest.exists(): dest.symlink_to(src) self.assertIsNone(shutil.which("flock", path=str(bin_dir))) env = dict(self.env, PATH=str(bin_dir)) r_user = self.svc("ub4", "user", "list", check=False, env=env) self.assertNotEqual(r_user.returncode, 0) self.assertIn("缺少 flock", r_user.stdout) r_dep = self.svc("ub4", "deploy", "--dry-run", check=False, env=env) self.assertNotEqual(r_dep.returncode, 0, r_dep.stdout) self.assertIn("缺少 flock", r_dep.stdout) self.assertTrue(self.running("ub4"), "拒绝后旧服务应仍在跑") self.svc("ub4", "stop") # ---- T5 recovery U1 身份冒烟（完整套件见 test_svc_identity）---- def test_T5_identity_regression_smoke(self): self.seed_agent("ub5", 9446) self.start_ok("ub5") iid = self.manifest("ub5")["instance_id"] self.svc("ub5", "stop") self.start_ok("ub5") self.assertEqual(self.manifest("ub5")["instance_id"], iid) self.svc("ub5", "stop") if __name__ == "__main__": unittest.main()