181 lines
7.2 KiB
Python
181 lines
7.2 KiB
Python
"""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 sys
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from _port_alloc import free_test_port # noqa: E402
|
||
|
||
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):
|
||
port = free_test_port()
|
||
self.seed_agent("ub1", port)
|
||
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):
|
||
port = free_test_port()
|
||
self.seed_agent("ub2", port)
|
||
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):
|
||
port = free_test_port()
|
||
self.seed_agent("ub3", port)
|
||
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):
|
||
port = free_test_port()
|
||
self.seed_agent("ub4", port)
|
||
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):
|
||
port = free_test_port()
|
||
self.seed_agent("ub5", port)
|
||
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()
|