admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
33ca6c75ed8dcc81dac4a67e587612c14c1c2a62
codex-app-server/cloud/tests/test_chv_muse_p2.py
T
host 33ca6c75ed fix(chv-muse-r12fix-2): P6 worker 接进镜像（build.sh + BUILDINFO + lockfile） 
R12 真机：chv-gitea-worker / chv-sync-worker-client /
privsep.py / safety.py / chv-gitea-worker.service 5 文件全不在
镜像内，build.sh 从未有 P6 拷入行，全链路无法跑。按 gap 文档
§2 接入说明落地：worker 三件硬校验 + 拷入 /usr/local/bin/
与 /etc/systemd/system/（mv 去后缀 + chmod +x + systemctl
enable）；privsep/safety 叶子模块条件拷入
/usr/local/lib/chv-gitea/（与 P3 同款口径，缺失只警告）；
叶子树 __init__ 用 stub（绝不拷真实 cloud/chv/__init__.py，
round-11 层 1 教训），stub 需 errors.py 故一并拷入；runcmd
在 chv-sync boot 前加 systemctl start chv-gitea-worker。
BUILDINFO contents 列入 worker 三件（无条件）与 privsep/
safety（条件，extra_contents）；两处 lockfile 登记补齐 P6
五项，另补 P4 三件自 P4 起缺失的登记。
2026-11-08 04:25:33 +00:00

1243 lines
53 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""对齐 Muse P2 测试：客机 nspawn 单元 + 浏览器子代理独立。
只测本轨交付物（chv-guest/chv-cell.py、chv-broker.py、chv-browser.py、
build.sh 干跑、buildinfo.py 内容项）。全部用假 socket/假进程/假
CDP broker，绝不连真实网络、不起真实 VM、不改生产行为。
"""
import importlib.util
import json
import os
import shutil
import socket
import struct
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parents[2]
def _load_script(name):
"""把 chv-guest 下的独立脚本当模块加载（无包结构）。"""
path = ROOT / "chv-guest" / name
spec = importlib.util.spec_from_file_location(
"chv_guest_p2_" + name[:-3], str(path))
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
CELL = _load_script("chv-cell.py")
BROKER = _load_script("chv-broker.py")
BROWSER = _load_script("chv-browser.py")
GUARD = _load_script("chv_browser_guard.py")
BUILDINFO = _load_script("buildinfo.py")
def _load_host():
"""把仓库根的 host.py 当模块加载（只含定义，__main__ 才跑
CLI，导入无副作用），供 worktree 根选择逻辑单测。"""
path = ROOT / "host.py"
spec = importlib.util.spec_from_file_location(
"chv_host_py", str(path))
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
_HOST = None
def _host():
"""懒加载 host.py（缓存）。合入 round-12 时由模块级 eager
改懒加载：host.py 顶层 `import queue` 会把 stdlib queue
缓进 sys.modules，使同进程后加载的 test_queue_sandbox
的 `from queue import TaskQueue`（依赖 cloud/queue.py
遮蔽 stdlib）取到 stdlib 而 ImportError——全量 2476 项
里唯一因此失败的模块。host.py 只在 worktree 测试里用，
首次使用时再加载即可。"""
global _HOST
if _HOST is None:
_HOST = _load_host()
return _HOST
class CellArgvTest(unittest.TestCase):
def test_private_users_maps_cell_root_to_chv(self):
argv = CELL.build_nspawn_argv(["true"])
self.assertIn("--private-users=1000:65536", argv)
self.assertIn("--machine=chv-cell", argv)
self.assertIn("--directory=/", argv)
self.assertIn("--as-pid2", argv)
# round-11 真机修复：nspawn 255 拒裸 --no-new-privileges
#（当作带必填参数的选项）与裸 --directory=/（须
# --volatile/--ephemeral）。复测纠正：--ephemeral 在客内
# ext4 根盘上做整盘快照拷贝，2.4G 盘只剩 593M 即 ENOSPC，
# 改 --volatile=yes（tmpfs 根 + 只读 /usr，不拷盘）。
self.assertIn("--no-new-privileges=yes", argv)
self.assertNotIn("--no-new-privileges", argv)
self.assertIn("--volatile=yes", argv)
self.assertNotIn("--ephemeral", argv)
def test_muse_named_capabilities_dropped(self):
argv = CELL.build_nspawn_argv(["true"])
for cap in ("CAP_SYS_PTRACE", "CAP_NET_ADMIN", "CAP_SYS_ADMIN",
"CAP_NET_RAW", "CAP_SYS_MODULE"):
self.assertIn("--drop-capability=%s" % cap, argv)
# 绝不能出现“保留这些能力”的写法。
for token in argv:
self.assertFalse(token.startswith("--capability=CAP_SYS_PTRACE"))
self.assertFalse(token.startswith("--capability=CAP_NET_ADMIN"))
def test_io_uring_blocked_in_syscall_filter(self):
argv = CELL.build_nspawn_argv(["true"])
filt = [a for a in argv if a.startswith("--system-call-filter=")]
self.assertEqual(len(filt), 1)
self.assertIn("@system-service", filt[0])
for call in ("io_uring_setup", "io_uring_enter",
"io_uring_register"):
self.assertIn("~%s" % call, filt[0])
def test_secrets_and_browser_hidden_from_cell(self):
argv = CELL.build_nspawn_argv(
["true"],
playwright_package_dir="/usr/local/lib/python3.12/dist-packages/playwright")
hidden = [a.split("=", 1)[1] for a in argv
if a.startswith("--inaccessible=")]
# /usr 下的浏览器路径仍遮蔽（/usr 只读挂进单元，遮蔽有效）。
for path in ("/usr/local/share/ms-playwright",
"/usr/local/lib/python3.12/dist-packages/playwright"):
self.assertIn(path, hidden)
# round-11 复测纠正（第二轮）：/etc/chv 与
# /home/chv/.git-credentials 不再列出——--volatile=yes 下
# 容器根是基本空的 tmpfs，这两处在容器根里根本不存在，
# 凭证文件按构造即不可达；列出反而让 nspawn 解析失败
#（Failed to resolve）而起不来。
for path in ("/etc/chv", "/home/chv/.git-credentials"):
self.assertNotIn(path, hidden)
def test_writable_binds_and_tmpfs(self):
argv = CELL.build_nspawn_argv(["true"])
self.assertIn("--bind=/home/chv/workspace", argv)
self.assertIn("--bind=/home/chv/.codex", argv)
self.assertIn("--bind=/run/chv", argv)
self.assertIn("--tmpfs=/tmp", argv)
self.assertIn("--chdir=/home/chv/workspace", argv)
self.assertIn("--setenv=CHV_CELL=1", argv)
self.assertIn("--setenv=HOME=/home/chv", argv)
def test_veth_default_and_shared_opt_in(self):
self.assertIn("--network-veth",
CELL.build_nspawn_argv(["true"]))
self.assertNotIn("--network-veth",
CELL.build_nspawn_argv(["true"],
network="shared"))
with self.assertRaises(ValueError):
CELL.build_nspawn_argv(["true"], network="macvlan")
with self.assertRaises(ValueError):
CELL.build_nspawn_argv([])
def test_env_forward_allowlist(self):
env = {"SUB2API_API_KEY": "sk-test",
"SUB2API_BASE_URL": "https://example.invalid",
"GITEA_TOKEN": "must-never-enter-cell",
"XCM_API_KEY": "xk-test"}
with mock.patch.dict(os.environ, env, clear=False):
argv = CELL.build_nspawn_argv(["true"])
self.assertIn("--setenv=SUB2API_API_KEY=sk-test", argv)
self.assertIn("--setenv=XCM_API_KEY=xk-test", argv)
self.assertFalse(any("GITEA_TOKEN" in a for a in argv))
class CellNetworkTest(unittest.TestCase):
def test_host_network_commands_nat_via_uplink(self):
cmds = CELL.host_network_commands("enp0s5")
flat = [" ".join(c) for c in cmds]
self.assertIn("ip addr add 10.200.0.1/24 dev ve-chv-cell", flat)
self.assertIn("ip link set ve-chv-cell up", flat)
self.assertIn("sysctl -w net.ipv4.ip_forward=1", flat)
self.assertTrue(any("iptables -t nat -A POSTROUTING -s 10.200.0.0/24 "
"-o enp0s5 -j MASQUERADE" in line
for line in flat))
self.assertTrue(any("-A FORWARD -i ve-chv-cell -o enp0s5 -j ACCEPT"
in line for line in flat))
with self.assertRaises(ValueError):
CELL.host_network_commands("")
def test_host_network_cleanup_mirrors_with_delete(self):
cmds = CELL.host_network_cleanup_commands("enp0s5")
self.assertTrue(cmds)
for cmd in cmds:
self.assertIn("iptables", cmd)
self.assertIn("-D", cmd)
self.assertNotIn("-A", cmd)
def test_cell_network_commands_via_nsenter(self):
cmds = CELL.cell_network_commands(4321)
flat = [" ".join(c) for c in cmds]
self.assertIn("nsenter -t 4321 -n ip addr add 10.200.0.2/24 "
"dev host0", flat)
self.assertIn("nsenter -t 4321 -n ip link set host0 up", flat)
self.assertIn("nsenter -t 4321 -n ip route add default via "
"10.200.0.1", flat)
with self.assertRaises(ValueError):
CELL.cell_network_commands(0)
def test_parse_default_route_iface(self):
out = ("default via 192.168.10.1 dev enp0s5 proto dhcp metric 100\n"
"192.168.10.0/24 dev enp0s5 proto kernel scope link src "
"192.168.10.10\n")
self.assertEqual(CELL.parse_default_route_iface(out), "enp0s5")
self.assertIsNone(CELL.parse_default_route_iface(""))
self.assertIsNone(CELL.parse_default_route_iface(
"192.168.10.0/24 dev enp0s5 proto kernel scope link\n"))
def test_find_uplink_iface_uses_injected_runner(self):
class FakeProc:
returncode = 0
stdout = "default via 192.168.10.1 dev tap0 proto static\n"
def fake_run(cmd, **kwargs):
self.assertEqual(cmd, ["ip", "route", "show", "default"])
return FakeProc()
self.assertEqual(CELL.find_uplink_iface(run=fake_run), "tap0")
def test_find_cell_init_pid_from_fake_proc(self):
with tempfile.TemporaryDirectory() as tmp:
task_dir = os.path.join(tmp, "777", "task", "777")
os.makedirs(task_dir)
with open(os.path.join(task_dir, "children"), "w") as fh:
fh.write("778 779")
self.assertEqual(
CELL.find_cell_init_pid(777, proc_root=tmp), 778)
self.assertIsNone(
CELL.find_cell_init_pid(999, proc_root=tmp))
class BrokerTest(unittest.TestCase):
def _fake_proc(self, root, mapping):
"""mapping: pid -> ns 标识。建出 /proc/<pid>/ns/pid 软链。"""
for pid, ns in mapping.items():
d = os.path.join(root, str(pid), "ns")
os.makedirs(d)
os.symlink("pid:[%s]" % ns, os.path.join(d, "pid"))
def test_peer_allowed_same_pid_ns(self):
with tempfile.TemporaryDirectory() as tmp:
self._fake_proc(tmp, {100: "111", 200: "111"})
ok, _ = BROKER.peer_allowed(1000, 200, cell_pid=100,
proc_root=tmp)
self.assertTrue(ok)
def test_peer_denied_uid_outside_cell_range(self):
with tempfile.TemporaryDirectory() as tmp:
self._fake_proc(tmp, {100: "111", 200: "111"})
for uid in (0, 999, 66536, 100000):
ok, reason = BROKER.peer_allowed(uid, 200, cell_pid=100,
proc_root=tmp)
self.assertFalse(ok, uid)
self.assertIn("单元映射范围", reason)
def test_peer_denied_different_pid_ns(self):
with tempfile.TemporaryDirectory() as tmp:
self._fake_proc(tmp, {100: "111", 200: "222"})
ok, reason = BROKER.peer_allowed(1000, 200, cell_pid=100,
proc_root=tmp)
self.assertFalse(ok)
self.assertIn("pid ns", reason)
def test_peer_denied_when_cell_ns_unreadable(self):
with tempfile.TemporaryDirectory() as tmp:
self._fake_proc(tmp, {200: "111"})
ok, _ = BROKER.peer_allowed(1000, 200, cell_pid=100,
proc_root=tmp)
self.assertFalse(ok)
def test_peer_uid_only_when_cell_pid_missing(self):
ok, reason = BROKER.peer_allowed(1000, 200, cell_pid=None)
self.assertTrue(ok)
self.assertIn("cell.pid 缺失", reason)
def test_ping_and_unknown_op(self):
self.assertEqual(BROKER.handle_request({"op": "ping"}),
{"ok": True, "cell": True})
resp = BROKER.handle_request({"op": "rm"})
self.assertFalse(resp["ok"])
resp = BROKER.handle_request("not-a-dict")
self.assertFalse(resp["ok"])
def test_sync_rejects_bad_action_and_outside_repo(self):
resp = BROKER.handle_request({"op": "sync", "action": "reset"})
self.assertFalse(resp["ok"])
for bad in ("/etc/passwd", "/home/chv/workspace/../..",
"/home/chv", "/tmp/x"):
resp = BROKER.handle_request(
{"op": "sync", "action": "pull", "repo": bad})
self.assertFalse(resp["ok"], bad)
def test_sync_runs_chv_sync_as_chv_without_shell(self):
seen = {}
class FakeProc:
returncode = 0
stdout = "已同步\n"
stderr = ""
def fake_run(cmd, **kwargs):
seen["cmd"] = cmd
self.assertFalse(kwargs.get("shell", False))
return FakeProc()
with mock.patch.object(BROKER.subprocess, "run", fake_run):
resp = BROKER.handle_request(
{"op": "sync", "action": "status",
"repo": "/home/chv/workspace/demo"})
self.assertTrue(resp["ok"])
self.assertEqual(seen["cmd"][:6],
["runuser", "-u", "chv", "--", "env",
"HOME=/home/chv"])
self.assertIn("/usr/local/bin/chv-sync", seen["cmd"])
self.assertIn("status", seen["cmd"])
self.assertIn("/home/chv/workspace/demo", seen["cmd"])
def test_sync_failure_reported_not_swallowed(self):
class FakeProc:
returncode = 1
stdout = ""
stderr = "分叉\n"
with mock.patch.object(BROKER.subprocess, "run",
lambda *a, **k: FakeProc()):
resp = BROKER.handle_request({"op": "sync", "action": "pull"})
self.assertFalse(resp["ok"])
self.assertEqual(resp["returncode"], 1)
def test_browser_forward_and_down(self):
payload = {"ok": True, "title": "Example Domain"}
class FakeConn:
def __init__(self, *args, **kwargs):
self.sent = b""
def settimeout(self, timeout):
pass
def connect(self, path):
self.path = path
def sendall(self, data):
self.sent = data
def recv(self, n):
data, self._buf = self._buf, b""
return data
def close(self):
pass
fake = FakeConn()
fake._buf = (json.dumps(payload) + "\n").encode()
with mock.patch.object(BROKER.socket, "socket",
lambda *a, **k: fake):
resp = BROKER.handle_request(
{"op": "browser", "action": "open"},
browser_sock="/run/chv/browser.sock")
self.assertEqual(resp, payload)
self.assertEqual(json.loads(fake.sent.decode())["op"], "browser")
def boom(*args, **kwargs):
raise OSError("no socket")
with mock.patch.object(BROKER.socket, "socket", boom):
resp = BROKER.handle_request({"op": "browser"})
self.assertFalse(resp["ok"])
self.assertIn("浏览器子代理未就绪", resp["error"])
def test_peer_creds_unpack(self):
# struct ucred = pid/uid/gid 三个 int；broker 用它验对端。
raw = struct.pack("3i", 4321, 1000, 1000)
class FakeConn:
def getsockopt(self, level, opt, size):
self.seen = (level, opt, size)
return raw
fake = FakeConn()
self.assertEqual(BROKER._peer_creds(fake), (4321, 1000, 1000))
self.assertEqual(fake.seen[1], socket.SO_PEERCRED)
class BrowserTest(unittest.TestCase):
def test_validate_url(self):
self.assertTrue(GUARD.validate_url("https://example.com")[0])
self.assertTrue(GUARD.validate_url("http://example.com/x")[0])
self.assertTrue(
GUARD.validate_url("https://user@example.com/")[0])
for bad in ("ftp://example.com", "file:///etc/passwd",
"https://localhost:8080/", "http://127.0.0.1/",
"https://192.168.1.1/", "https://10.0.0.5/",
"https://172.17.0.2/", "", None):
ok, _ = GUARD.validate_url(bad)
self.assertFalse(ok, bad)
def test_truncate_aria_small_snapshot_untouched(self):
text = ("- generic [ref=e1]:\n"
" - heading \"Example Domain\" [level=1] [ref=e2]")
out, truncated = GUARD.truncate_aria(text)
self.assertFalse(truncated)
self.assertEqual(out, text)
def test_truncate_aria_node_cap(self):
text = "\n".join("- link \"l%d\" [ref=e%d]" % (i, i)
for i in range(300))
out, truncated = GUARD.truncate_aria(text)
self.assertTrue(truncated)
nodes = [ln for ln in out.splitlines()
if ln.lstrip().startswith("- ")]
self.assertLessEqual(len(nodes), 200)
def test_truncate_aria_char_cap(self):
text = "- paragraph \"" + "x" * 70000 + "\" [ref=e1]"
out, truncated = GUARD.truncate_aria(text)
self.assertTrue(truncated)
self.assertLessEqual(len(out), GUARD.MAX_A11Y_CHARS)
def test_forbidden_ops_never_execute(self):
agent = mock.Mock()
for op in ("evaluate", "content", "devtools", "js"):
resp = BROWSER.handle_request(agent, {"op": op})
self.assertFalse(resp["ok"])
self.assertIn("不暴露", resp["error"])
agent.open.assert_not_called()
resp = BROWSER.handle_request(agent, {"op": "bogus"})
self.assertFalse(resp["ok"])
def test_open_rejects_bad_url_without_calling_broker(self):
def exploding_broker(req):
raise AssertionError("坏 URL 不应调 broker")
agent = BROWSER.BrowserAgent(broker_call=exploding_broker)
resp = agent.open("https://127.0.0.1/secret")
self.assertFalse(resp["ok"])
def test_snapshot_and_screenshot_need_open_first(self):
with tempfile.TemporaryDirectory() as tmp:
agent = BROWSER.BrowserAgent(
broker_call=FakeCdpBroker(tmp))
self.assertFalse(agent.snapshot()["ok"])
self.assertFalse(agent.screenshot()["ok"])
self.assertTrue(agent.close()["ok"])
def test_open_returns_a11y_snapshot_with_fake_cdp_broker(self):
with tempfile.TemporaryDirectory() as tmp:
agent = BROWSER.BrowserAgent(
broker_call=FakeCdpBroker(tmp))
resp = agent.open("https://example.com")
self.assertTrue(resp["ok"])
self.assertEqual(resp["title"], "Example Domain")
self.assertEqual(resp["url"], "https://example.com/")
# round-12 移植 round-11 的 ARIA 字符串纠正
#（Page.aria_snapshot()）：a11y 是 ARIA 快照字符串，
# 不是 dict 树；快照走 chv-cdp-broker.py 的 snapshot。
self.assertIsInstance(resp["a11y"], str)
self.assertIn("Example Domain", resp["a11y"])
self.assertFalse(resp["truncated"])
snap = agent.snapshot()
self.assertTrue(snap["ok"])
shot = agent.screenshot()
self.assertTrue(shot["ok"])
self.assertTrue(shot["path"].endswith(".png"))
self.assertTrue(os.path.exists(shot["path"]))
self.assertTrue(agent.close()["ok"])
def test_browser_op_dispatches_on_action(self):
# round-11 复测回归：broker 原样转发 {"op": "browser", …}，
# chv-browser 白名单却不认 "browser"。内层 op 在 action 里；
# 无 action 但带 url 时默认 open（驱动只发 url）。
agent = mock.Mock()
agent.open.return_value = {"ok": True}
resp = BROWSER.handle_request(
agent, {"op": "browser", "action": "open",
"url": "https://example.com"})
self.assertTrue(resp["ok"])
agent.open.assert_called_once_with("https://example.com")
agent.reset_mock()
resp = BROWSER.handle_request(
agent, {"op": "browser", "url": "https://example.com"})
self.assertTrue(resp["ok"])
agent.open.assert_called_once_with("https://example.com")
# 内层 forbidden 同样拒绝。
agent.reset_mock()
resp = BROWSER.handle_request(
agent, {"op": "browser", "action": "evaluate"})
self.assertFalse(resp["ok"])
agent.open.assert_not_called()
# 无 action 也无 url：无法推断内层 op。
resp = BROWSER.handle_request(agent, {"op": "browser"})
self.assertFalse(resp["ok"])
class FakeCdpBroker:
"""假 CDP broker：收到 req dict 返回 resp dict；open 后才有页。
round-12：P4 起 BrowserAgent 不再直连 Playwright，Chromium
的所有权在单元外 CDP broker 手里；本测试用假 broker 走同一
协议，断言与 P2 原版一致。"""
def __init__(self, shots_dir):
self._shots_dir = shots_dir
self._open = False
self.calls = []
def __call__(self, req):
self.calls.append(req.get("op"))
op = req.get("op")
aria = ("- generic [ref=e1]:\n"
" - heading \"Example Domain\" [level=1] [ref=e2]")
if op == "open":
self._open = True
return {"ok": True, "title": "Example Domain",
"url": "https://example.com/", "a11y": aria}
if op == "snapshot":
if not self._open:
return {"ok": False, "error": "尚未 open"}
return {"ok": True, "title": "Example Domain",
"url": "https://example.com/", "a11y": aria}
if op == "screenshot":
if not self._open:
return {"ok": False, "error": "尚未 open"}
path = os.path.join(self._shots_dir, "shot.png")
with open(path, "wb") as fh:
fh.write(b"\x89PNG")
return {"ok": True, "path": path}
if op == "close":
self._open = False
return {"ok": True}
return {"ok": False, "error": "未知操作"}
class CellCliTest(unittest.TestCase):
def test_cli_argv_prints_json(self):
import contextlib
import io
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
rc = CELL.main(["argv", "--", "echo", "hi"])
self.assertEqual(rc, 0)
argv = json.loads(buf.getvalue())
self.assertIn("--private-users=1000:65536", argv)
self.assertEqual(argv[-2:], ["echo", "hi"])
def test_cli_run_requires_double_dash(self):
self.assertEqual(CELL.main(["run", "echo", "hi"]), 2)
self.assertEqual(CELL.main(["bogus"]), 2)
self.assertEqual(CELL.main([]), 2)
def test_cli_network_commands_prints_tables(self):
import contextlib
import io
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
rc = CELL.main(["network-commands", "--uplink", "enp0s5",
"--cell-pid", "99"])
self.assertEqual(rc, 0)
out = json.loads(buf.getvalue())
self.assertIn("host", out)
self.assertIn("cell", out)
self.assertIn("host_cleanup", out)
def test_cli_unknown_args_rejected(self):
self.assertEqual(BROKER.main(["--bogus"]), 2)
self.assertEqual(BROWSER.main(["--bogus"]), 2)
class BuildInfoTest(unittest.TestCase):
def _synth_out(self, tmp):
out = os.path.join(tmp, "out")
os.makedirs(out)
for name, content in (("disk.raw", b"D"), ("vmlinux", b"K")):
with open(os.path.join(out, name), "wb") as fh:
fh.write(content)
for name, lines in (
("guest-versions-dpkg.txt", ["git 9.9.9-9ubuntu9"]),
("guest-versions-pip.txt", ["playwright==9.9.9"]),
("guest-versions-gt.txt", ["gt version 1.0.0"]),
("guest-versions-chromium.txt",
["chromium_headless_shell-1248"])):
with open(os.path.join(out, name), "w") as fh:
fh.write("\n".join(lines) + "\n")
return out
def test_contents_include_p2_items(self):
with tempfile.TemporaryDirectory() as tmp:
out = self._synth_out(tmp)
info = BUILDINFO.build(
out, "chv-guest-24.04-0.164.0-alpha.9-abc1234", "24.04",
"0.164.0-alpha.9", "abc1234", "0" * 64, True, "1248")
for item in ("systemd-container", "chv-cell", "chv-broker",
"chv-browser"):
self.assertIn(item, info["contents"])
def test_extra_contents_appended_once(self):
with tempfile.TemporaryDirectory() as tmp:
out = self._synth_out(tmp)
info = BUILDINFO.build(
out, "chv-guest-24.04-0.164.0-alpha.9-abc1234", "24.04",
"0.164.0-alpha.9", "abc1234", "0" * 64, True, "1248",
["vsock_notify"])
self.assertIn("vsock_notify", info["contents"])
info2 = BUILDINFO.build(
out, "chv-guest-24.04-0.164.0-alpha.9-abc1234", "24.04",
"0.164.0-alpha.9", "abc1234", "0" * 64, True, "1248",
["vsock_notify", "chv-cell"])
self.assertEqual(info2["contents"].count("chv-cell"), 1)
self.assertIn("vsock_notify", info2["contents"])
def test_main_accepts_tenth_arg(self):
with tempfile.TemporaryDirectory() as tmp:
out = self._synth_out(tmp)
args = ["buildinfo.py", out,
"chv-guest-24.04-0.164.0-alpha.9-abc1234", "24.04",
"0.164.0-alpha.9", "abc1234", "0" * 64, "1", "1248",
"vsock_notify"]
self.assertEqual(BUILDINFO.main(args), 0)
with open(os.path.join(out, "BUILDINFO.json")) as fh:
info = json.load(fh)
self.assertIn("vsock_notify", info["contents"])
self.assertEqual(BUILDINFO.main(args[:9]), 0)
class BuildDryRunTest(unittest.TestCase):
def _fixture(self, tmp, with_vsock=False):
# 隔离的临时 git 仓：工作树干净、HEAD 与 REPO_COMMIT
# 一致、假 codex 二进制报对版本。with_vsock 时一并拷入
# vsock_notify.py（round-11 真机修复：服务须随文件建）。
repo = os.path.join(tmp, "repo")
guest = os.path.join(repo, "chv-guest")
bindir = os.path.join(tmp, "bin")
os.makedirs(guest)
os.makedirs(os.path.join(bindir, "codex-resources"))
for name in ("build.sh", "buildinfo.py", "check_lockfile.py",
"generate_lockfile.py", "chv-sync.sh",
"chv-cell.py", "chv-broker.py", "chv-browser.py",
"authd.py", "git-credential-chv-authd",
"chv-authd.service", "authd_init_stub.py",
# round-12 P4：build.sh 缺一即 fail-fast，夹具
# 须一并拷入（合入时补：R11 夹具本无此四件）。
"chv-cdp-broker.py", "chv_browser_guard.py",
"browser-blocklist.txt", "chv-memory.py",
# round-12 修复轮（项 2，P6）：worker 三件亦
# 是 build.sh 硬校验，夹具须一并拷入。
"chv-gitea-worker.py", "chv-sync-worker-client.py",
"chv-gitea-worker.service"):
shutil.copy2(ROOT / "chv-guest" / name,
os.path.join(guest, name))
shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py"))
# round-11：build.sh 要求 P3 云端叶子模块在仓里（fail-fast），
# 夹具须一并拷入，否则干跑在 P3 预检就失败。
for rel in ("cloud/chv/authd.py", "cloud/chv/sentinel.py",
"cloud/chv/errors.py",
"cloud/__init__.py", "cloud/chv/__init__.py"):
dest = os.path.join(repo, rel)
os.makedirs(os.path.dirname(dest), exist_ok=True)
shutil.copy2(ROOT / rel, dest)
if with_vsock:
rel = "cloud/chv/guest/vsock_notify.py"
dest = os.path.join(repo, rel)
os.makedirs(os.path.dirname(dest), exist_ok=True)
shutil.copy2(ROOT / rel, dest)
codex = os.path.join(bindir, "codex")
with open(codex, "w") as fh:
fh.write("#!/bin/sh\necho 'codex-cli 0.164.0-alpha.9'\n")
os.chmod(codex, 0o755)
host_bin = os.path.join(bindir, "codex-code-mode-host")
with open(host_bin, "w") as fh:
fh.write("#!/bin/sh\nexit 0\n")
os.chmod(host_bin, 0o755)
subprocess.run(["git", "init", "-q"], cwd=repo, check=True)
subprocess.run(["git", "add", "-A"], cwd=repo, check=True)
subprocess.run(
["git", "-c", "user.email=t@t", "-c", "user.name=t",
"commit", "-qm", "p2 dry-run fixture"],
cwd=repo, check=True)
head = subprocess.run(
["git", "rev-parse", "--short", "HEAD"], cwd=repo,
capture_output=True, text=True, check=True).stdout.strip()
env = dict(os.environ)
env.update({
"CODEX_BIN_DIR": bindir,
"UBUNTU_VERSION": "24.04",
"CODEX_VERSION": "0.164.0-alpha.9",
"REPO_COMMIT": head,
"UBUNTU_IMAGE_SHA256": "a" * 64,
})
return repo, guest, env
def _dry_run(self, repo, guest, env):
proc = subprocess.run(
["bash", os.path.join(guest, "build.sh"),
"--dry-run", "out"],
cwd=repo, capture_output=True, text=True, env=env)
self.assertEqual(proc.returncode, 0, proc.stderr)
return proc.stdout + proc.stderr
def test_dry_run_contains_p2_steps(self):
# 在隔离的临时 git 仓里跑 build.sh --dry-run。
with tempfile.TemporaryDirectory() as tmp:
repo, guest, env = self._fixture(tmp)
out = self._dry_run(repo, guest, env)
for needle in ("systemd-container", "chv-cell", "chv-broker",
"chv-browser", "chv-browser.service",
"chv-broker.service", "useradd --system",
"systemctl enable chv-broker chv-browser",
"chv-authd", "git-credential-chv-authd",
"systemctl enable chv-authd",
"systemctl start chv-authd",
# round-11 真机修复（缺陷 4）：browser.sock
# 移进 chv-browser 自有目录，build.sh 须
# 建目录 + tmpfiles 登记。
"/run/chv-browser",
# round-11 真机修复（层 1）：叶子树含
# errors.py + stub __init__，不拷真实的
# cloud/chv/__init__.py。
"errors.py", "authd_init_stub.py",
# round-12 修复轮（项 2，P6）：worker
# 三件进镜像，干跑输出须含拷入与 enable。
"chv-gitea-worker",
"chv-gitea-worker.service",
"systemctl enable chv-gitea-worker"):
self.assertIn(needle, out, needle)
# P1 文件不在本分支：干跑只警告、不失败。
self.assertIn("vsock_notify.py 不在仓库里", out)
# round-11 真机修复（层 5）：镜像不再设 store helper
#（authd helper 是唯一凭证 helper），git user 仍配。
self.assertNotIn("credential.helper store", out)
self.assertIn("git config --system user.name chv", out)
def test_dry_run_with_vsock_creates_service(self):
# round-11 真机修复（vsock 自动启动）：vsock_notify.py
# 在场时建 vsock-notify.service 并 enable，不再只警告。
with tempfile.TemporaryDirectory() as tmp:
repo, guest, env = self._fixture(tmp, with_vsock=True)
out = self._dry_run(repo, guest, env)
self.assertNotIn("vsock_notify.py 不在仓库里", out)
for needle in ("vsock-notify.service",
"systemctl enable vsock-notify",
"ExecStart=/usr/local/bin/vsock-notify"):
self.assertIn(needle, out, needle)
class CellGateTest(unittest.TestCase):
"""part 4：起步门包装。"""
def test_gate_wrapper_waits_then_execs(self):
wrapped = CELL.gate_wrapper_command(
["python3", "/app/host.py", "status"],
"/run/chv/.cell-gate-chv-cell-00")
self.assertEqual(wrapped[:4], [
"/usr/bin/sh", "-c",
"while [ ! -e \"$1\" ]; do sleep 0.05; done; "
"shift; exec \"$@\"",
"chv-cell-gate"])
self.assertEqual(wrapped[4],
"/run/chv/.cell-gate-chv-cell-00")
self.assertEqual(wrapped[5:],
["python3", "/app/host.py", "status"])
# 绝对路径：volatile 单元根里 /bin 未必有，/usr 必有。
self.assertTrue(wrapped[0].startswith("/usr/bin/"))
class CellInitPidPollTest(unittest.TestCase):
"""part 4：find_cell_init_pid 带超时轮询。"""
def _proc_root_with_child(self, tmp, pid=777, child=778):
task_dir = os.path.join(tmp, str(pid), "task", str(pid))
os.makedirs(task_dir)
with open(os.path.join(task_dir, "children"), "w") as fh:
fh.write(str(child))
def test_single_shot_default_unchanged(self):
with tempfile.TemporaryDirectory() as tmp:
self._proc_root_with_child(tmp)
self.assertEqual(
CELL.find_cell_init_pid(777, proc_root=tmp), 778)
self.assertIsNone(
CELL.find_cell_init_pid(999, proc_root=tmp))
def test_poll_finds_late_child(self):
import threading
with tempfile.TemporaryDirectory() as tmp:
def _late():
import time as _time
_time.sleep(0.3)
self._proc_root_with_child(tmp)
threading.Thread(target=_late, daemon=True).start()
self.assertEqual(
CELL.find_cell_init_pid(777, proc_root=tmp,
timeout=5.0), 778)
def test_poll_returns_none_on_timeout(self):
import time as _time
with tempfile.TemporaryDirectory() as tmp:
start = _time.time()
self.assertIsNone(
CELL.find_cell_init_pid(999, proc_root=tmp,
timeout=0.3))
self.assertLess(_time.time() - start, 5.0)
def test_poll_stops_when_nspawn_dead(self):
import time as _time
with tempfile.TemporaryDirectory() as tmp:
start = _time.time()
self.assertIsNone(
CELL.find_cell_init_pid(999, proc_root=tmp,
timeout=10.0,
proc_alive=lambda: False))
# nspawn 已死即立刻回，不空等 10 秒。
self.assertLess(_time.time() - start, 5.0)
class CellPart4ArgvTest(unittest.TestCase):
"""part 4：volatile 空根补 bind + 唯一身份进 argv。"""
def test_app_and_ca_bundle_bound_read_only(self):
argv = CELL.build_nspawn_argv(["true"])
self.assertIn("--bind-ro=/app", argv)
self.assertIn("--bind-ro=/etc/ssl/certs", argv)
# /app 是只读绑：单元内 python3 /app/host.py 只读，不写。
self.assertNotIn("--bind=/app", argv)
def test_extra_ro_binds(self):
argv = CELL.build_nspawn_argv(
["true"], extra_ro_binds=[
("/run/chv/resolv.conf.chv-cell-00", "/etc/resolv.conf"),
("/run/chv/hosts.chv-cell-00", "/etc/hosts")])
self.assertIn(
"--bind-ro=/run/chv/resolv.conf.chv-cell-00:/etc/resolv.conf",
argv)
self.assertIn(
"--bind-ro=/run/chv/hosts.chv-cell-00:/etc/hosts", argv)
def test_cell_net_machine_and_no_legacy_name(self):
net = CELL.CellNet(index=3, machine="chv-cell-03",
host_iface="ve-chv-cell-03",
host_ip="10.200.3.1", cell_ip="10.200.3.2",
subnet="10.200.3.0/24")
argv = CELL.build_nspawn_argv(["true"], cell_net=net)
self.assertIn("--machine=chv-cell-03", argv)
self.assertNotIn("--machine=chv-cell", argv)
# 旧常量身份仍是默认（argv 子命令与旧测试不改）。
self.assertIn("--machine=chv-cell",
CELL.build_nspawn_argv(["true"]))
def test_legacy_cell_net(self):
net = CELL.legacy_cell_net()
self.assertEqual(net.machine, "chv-cell")
self.assertEqual(net.host_iface, "ve-chv-cell")
self.assertEqual(net.subnet, "10.200.0.0/24")
class CellPart6WorktreeTest(unittest.TestCase):
"""part 6 修复（RETEST5 新发现）：worktree 根移出 /app。
part 5 的旧测试钉住的是 argv 顺序（/app/work 绑定排在
--bind-ro=/app 之后），RETEST5 真机证明该顺序本身不可行：
后挂的 bind 只有在挂载点已存在时才能盖住，挂载点不存在时
nspawn 在只读 /app 下连建都建不起来（全量 73 次同一错误，
CELL-RC=0 共 0 次）。本轮单测不再钉顺序，改钉语义：argv 里
不得出现任何指向只读 /app 之下的 bind（挂载点不存在时不发
致命 bind），worktree 根改经 CHV_WORKTREE_ROOT 环境变量传
进单元，落点在已可写绑进单元的 workspace 之下。"""
@staticmethod
def _bind_dests(argv):
"""argv 里全部 bind/bind-ro 的单元内目的路径。"""
dests = []
for token in argv:
for prefix in ("--bind=", "--bind-ro="):
if token.startswith(prefix):
spec = token[len(prefix):]
dests.append(spec.split(":", 1)[1]
if ":" in spec else spec)
return dests
def test_no_bind_targets_anything_under_readonly_app(self):
argv = CELL.build_nspawn_argv(["true"])
# RETEST5 的致命绑定本身必须消失。
self.assertNotIn(
"--bind=/home/chv/workspace/.chv-worktrees:/app/work",
argv)
# 更一般的语义：任何 bind 的目的路径都不得落在只读的
# /app 之下（/app 本身是 --bind-ro 的挂载点，允许；其
# 子路径挂载点在镜像里不存在，nspawn 建不了即 EROFS）。
for dest in self._bind_dests(argv):
self.assertFalse(
dest.startswith("/app/"),
"bind 目的路径在只读 /app 之下：%s" % dest)
def test_worktree_root_passed_via_setenv(self):
argv = CELL.build_nspawn_argv(["true"])
self.assertIn(
"--setenv=CHV_WORKTREE_ROOT="
"/home/chv/workspace/.chv-worktrees",
argv)
def test_worktree_dir_reachable_via_writable_workspace_bind(self):
# worktree 根在 workspace 之下，且 workspace 本身已以
# 可写 bind（非 --bind-ro）进单元——根无需额外 bind，
# 挂载点问题按构造即不存在。
self.assertTrue(
CELL.WORKTREE_DIR.startswith(CELL.WORKSPACE + "/"))
argv = CELL.build_nspawn_argv(["true"])
self.assertIn("--bind=%s" % CELL.WORKSPACE, argv)
self.assertNotIn("--bind-ro=%s" % CELL.WORKSPACE, argv)
# .chv-worktrees 只应出现在 --setenv 里，不应出现在
# 任何 bind/bind-ro 参数里。
for token in argv:
if ".chv-worktrees" in token:
self.assertTrue(
token.startswith("--setenv=CHV_WORKTREE_ROOT="),
"worktree 目录不应再经 bind 进单元：%s" % token)
def test_worktree_dir_matches_host_cell_fallback(self):
# chv-cell 的 WORKTREE_DIR 与 host.py 在 CHV_CELL=1 且
# 环境变量未设时的回落值须是同一路径，否则两边各说各话。
self.assertEqual(CELL.WORKTREE_DIR, _host().CELL_WORKTREE_ROOT)
class HostWorktreeBaseTest(unittest.TestCase):
"""part 6：host.py 的 worktree 根选择（worktree_base）与
make_worktree 落点——单元内永不在 /app 之下，单元外不变。"""
def test_env_override_wins_over_cell_fallback(self):
with mock.patch.dict(
os.environ,
{"CHV_WORKTREE_ROOT": "/tmp/wt-root",
"CHV_CELL": "1"},
clear=True):
self.assertEqual(
_host().worktree_base(), "/tmp/wt-root/host-worktrees")
def test_cell_without_env_falls_back_to_workspace(self):
with mock.patch.dict(os.environ, {"CHV_CELL": "1"},
clear=True):
base = _host().worktree_base()
self.assertEqual(
base,
"/home/chv/workspace/.chv-worktrees/host-worktrees")
self.assertFalse(base.startswith("/app/"))
def test_outside_cell_uses_here_work(self):
with mock.patch.dict(os.environ, {}, clear=True):
self.assertEqual(
_host().worktree_base(),
os.path.join(_host().HERE, "work", "host-worktrees"))
def _make_git_repo(self, tmp):
repo = os.path.join(tmp, "repo")
os.makedirs(repo)
for args in (["init", "-q", repo],
["-C", repo, "config", "user.email",
"test@example.invalid"],
["-C", repo, "config", "user.name", "test"]):
subprocess.run(["git"] + args, check=True,
capture_output=True)
with open(os.path.join(repo, "README.md"), "w") as fh:
fh.write("hi\n")
subprocess.run(["git", "-C", repo, "add", "."],
check=True, capture_output=True)
subprocess.run(
["git", "-C", repo, "commit", "-qm", "init"],
check=True, capture_output=True)
return repo
def test_make_worktree_uses_override_not_app(self):
with tempfile.TemporaryDirectory() as tmp:
repo = self._make_git_repo(tmp)
wt_root = os.path.join(tmp, "wt")
with mock.patch.dict(
os.environ,
{"CHV_WORKTREE_ROOT": wt_root, "CHV_CELL": ""},
clear=False):
path, branch = _host().make_worktree(repo, "t1")
self.assertEqual(
path, os.path.join(wt_root, "host-worktrees", "t1"))
self.assertEqual(branch, "host/t1")
self.assertFalse(path.startswith("/app/"))
# 真是 git worktree，不是空目录拷贝。
proc = subprocess.run(
["git", "-C", path, "rev-parse", "--show-toplevel"],
capture_output=True, text=True)
self.assertEqual(proc.returncode, 0)
self.assertEqual(proc.stdout.strip(), path)
def test_dispatch_from_branch_uses_same_base(self):
# 第二个 HERE/work 调用方（依赖派发）须与 make_worktree
# 走同一根，否则 goal/fork 的依赖链仍撞只读 /app。
with tempfile.TemporaryDirectory() as tmp:
repo = self._make_git_repo(tmp)
wt_root = os.path.join(tmp, "wt")
with mock.patch.dict(
os.environ,
{"CHV_WORKTREE_ROOT": wt_root, "CHV_CELL": ""},
clear=False):
dep_path, _ = _host().make_worktree(repo, "dep")
created = {}
class _FakeServer:
def create_thread(self, path, **kwargs):
created["path"] = path
return "th-1"
state = {"tasks": {}}
task, err = _host().dispatch_from_branch(
_FakeServer(), state, "child", "prompt", repo,
"host/dep", dep_path, "m", "low",
"danger-full-access", "ask")
self.assertEqual(err, "")
self.assertEqual(
task["cwd"],
os.path.join(wt_root, "host-worktrees", "child"))
self.assertFalse(task["cwd"].startswith("/app/"))
class CellFilesTest(unittest.TestCase):
"""part 4：单元专属 resolv.conf/hosts 生成。"""
def _write(self, path, content):
with open(path, "w") as fh:
fh.write(content)
return path
def test_resolv_filters_stub_and_falls_back(self):
with tempfile.TemporaryDirectory() as tmp:
stub = self._write(os.path.join(tmp, "resolv.conf"),
"nameserver 127.0.0.53\n"
"options edns0\n")
real = self._write(os.path.join(tmp, "resolved.conf"),
"nameserver 8.8.8.8\n"
"nameserver 1.1.1.1\n")
files = CELL.prepare_cell_files(
"chv-cell-00", run_dir=tmp, resolv_conf=stub,
resolved_conf=real,
hosts_file=os.path.join(tmp, "no-hosts"))
with open(files["resolv"]) as fh:
content = fh.read()
self.assertEqual(content,
"nameserver 8.8.8.8\nnameserver 1.1.1.1\n")
self.assertNotIn("127.0.0.53", content)
with open(files["hosts"]) as fh:
self.assertIn("127.0.0.1 localhost", fh.read())
self.assertEqual(oct(os.stat(files["resolv"]).st_mode
& 0o777), "0o644")
self.assertEqual(files["gate"],
os.path.join(tmp, ".cell-gate-chv-cell-00"))
# gate 文件本身不在这里创建（run_cell 配网后才开门）。
self.assertFalse(os.path.exists(files["gate"]))
def test_resolv_real_servers_used_directly(self):
with tempfile.TemporaryDirectory() as tmp:
resolv = self._write(os.path.join(tmp, "resolv.conf"),
"nameserver 8.8.4.4\n")
files = CELL.prepare_cell_files(
"chv-cell-01", run_dir=tmp, resolv_conf=resolv,
resolved_conf=os.path.join(tmp, "missing"),
hosts_file=os.path.join(tmp, "missing-hosts"))
with open(files["resolv"]) as fh:
self.assertEqual(fh.read(), "nameserver 8.8.4.4\n")
def test_no_dns_anywhere_is_loud(self):
with tempfile.TemporaryDirectory() as tmp:
with self.assertRaises(RuntimeError):
CELL.prepare_cell_files(
"chv-cell-00", run_dir=tmp,
resolv_conf=os.path.join(tmp, "missing"),
resolved_conf=os.path.join(tmp, "missing2"),
hosts_file=os.path.join(tmp, "missing-hosts"))
class CellPart4NetworkTest(unittest.TestCase):
"""part 4：配网命令表走唯一身份。"""
def _net(self, index=5):
return CELL.CellNet(
index=index, machine="chv-cell-%02x" % index,
host_iface="ve-chv-cell-%02x" % index,
host_ip="10.200.%d.1" % index,
cell_ip="10.200.%d.2" % index,
subnet="10.200.%d.0/24" % index)
def test_host_commands_use_cell_net(self):
net = self._net(5)
flat = [" ".join(c)
for c in CELL.host_network_commands("enp0s5", net)]
self.assertIn("ip addr add 10.200.5.1/24 dev ve-chv-cell-05",
flat)
self.assertTrue(any(
"iptables -t nat -A POSTROUTING -s 10.200.5.0/24 "
"-o enp0s5 -j MASQUERADE" in line for line in flat))
# 旧常量仍是默认。
legacy = [" ".join(c)
for c in CELL.host_network_commands("enp0s5")]
self.assertIn("ip addr add 10.200.0.1/24 dev ve-chv-cell",
legacy)
def test_cell_commands_use_cell_net(self):
net = self._net(7)
flat = [" ".join(c)
for c in CELL.cell_network_commands(4321, net)]
self.assertIn("nsenter -t 4321 -n ip addr add 10.200.7.2/24 "
"dev host0", flat)
self.assertIn("nsenter -t 4321 -n ip route add default via "
"10.200.7.1", flat)
class CellAllocTest(unittest.TestCase):
"""part 4：锁下分配唯一 cell 身份。"""
class _FakeProc:
def __init__(self, returncode=0, stdout=""):
self.returncode = returncode
self.stdout = stdout
def _fake_run(self, link_out="", addr_out="", machinectl_out="",
machinectl_rc=0):
def run(cmd, **kwargs):
if cmd[:3] == ["ip", "-o", "link"]:
return self._FakeProc(0, link_out)
if cmd[:3] == ["ip", "-o", "addr"]:
return self._FakeProc(0, addr_out)
if cmd[0] == "machinectl":
return self._FakeProc(machinectl_rc, machinectl_out)
raise AssertionError("意外命令：%r" % (cmd,))
return run
def test_allocates_lowest_free_index(self):
link = ("2: ve-chv-cell-00@if2: <UP> mtu 1500\n"
"3: ve-chv-cell-01@if3: <UP> mtu 1500\n")
addr = ("2: ve-chv-cell-00 inet 10.200.0.1/24 scope global\n"
"3: ve-chv-cell-01 inet 10.200.1.1/24 scope global\n")
with tempfile.TemporaryDirectory() as tmp:
lock = os.path.join(tmp, "lock")
net = CELL.allocate_cell_net(
run=self._fake_run(link, addr), lock_path=lock)
self.assertEqual(net.index, 2)
self.assertEqual(net.machine, "chv-cell-02")
self.assertEqual(net.host_iface, "ve-chv-cell-02")
self.assertEqual(net.host_ip, "10.200.2.1")
self.assertEqual(net.cell_ip, "10.200.2.2")
self.assertEqual(net.subnet, "10.200.2.0/24")
self.assertLessEqual(len(net.host_iface), 15)
def test_machinectl_registered_index_skipped(self):
# 接口已消失、machine 仍在注销：避让该下标。
run = self._fake_run(
"", "", machinectl_out="chv-cell-00 1234 ...\n")
with tempfile.TemporaryDirectory() as tmp:
net = CELL.allocate_cell_net(
run=run, lock_path=os.path.join(tmp, "lock"))
self.assertEqual(net.index, 1)
def test_machinectl_missing_proceeds(self):
def run(cmd, **kwargs):
if cmd[0] == "machinectl":
raise OSError("无此命令")
return self._FakeProc(0, "")
with tempfile.TemporaryDirectory() as tmp:
net = CELL.allocate_cell_net(
run=run, lock_path=os.path.join(tmp, "lock"))
self.assertEqual(net.index, 0)
def test_ip_failure_is_loud(self):
def run(cmd, **kwargs):
return self._FakeProc(1, "")
with tempfile.TemporaryDirectory() as tmp:
with self.assertRaises(RuntimeError):
CELL.allocate_cell_net(
run=run, lock_path=os.path.join(tmp, "lock"))
class BrokerMultiTest(unittest.TestCase):
"""part 4：broker 对多 cell 的 pid ns 对照。"""
def _fake_proc(self, root, mapping):
for pid, ns in mapping.items():
d = os.path.join(root, str(pid), "ns")
os.makedirs(d)
os.symlink("pid:[%s]" % ns, os.path.join(d, "pid"))
def test_peer_allowed_if_matches_any_cell(self):
with tempfile.TemporaryDirectory() as tmp:
self._fake_proc(tmp, {100: "111", 101: "222", 200: "222"})
ok, _ = BROKER.peer_allowed_multi(
1000, 200, [100, 101], proc_root=tmp)
self.assertTrue(ok)
def test_peer_denied_if_matches_none(self):
with tempfile.TemporaryDirectory() as tmp:
self._fake_proc(tmp, {100: "111", 101: "222", 200: "333"})
ok, reason = BROKER.peer_allowed_multi(
1000, 200, [100, 101], proc_root=tmp)
self.assertFalse(ok)
self.assertIn("pid ns", reason)
def test_stale_pid_file_skipped_not_fatal(self):
with tempfile.TemporaryDirectory() as tmp:
# 100 已退出（无 /proc 项），101 活着且同 ns。
self._fake_proc(tmp, {101: "222", 200: "222"})
ok, _ = BROKER.peer_allowed_multi(
1000, 200, [100, 101], proc_root=tmp)
self.assertTrue(ok)
def test_all_ns_unreadable_rejects(self):
with tempfile.TemporaryDirectory() as tmp:
self._fake_proc(tmp, {200: "222"})
ok, _ = BROKER.peer_allowed_multi(
1000, 200, [100, 101], proc_root=tmp)
self.assertFalse(ok)
def test_no_pid_files_uid_only(self):
ok, reason = BROKER.peer_allowed_multi(1000, 200, [])
self.assertTrue(ok)
self.assertIn("未做 pid ns 对照", reason)
# part 5：reason 须保留“cell.pid 缺失”字样——P2-9 驱动
# grep 该串认降级日志；part 4 切到 multi 版时改写成
# “无 cell pid 文件”，该串再也没进过 journal，
# RETEST4 的 degraded_logged 即因此为 False。
self.assertIn("cell.pid 缺失", reason)
def test_degraded_warning_carries_probe_string(self):
# serve() 打的警告行就是 _degraded_warning 的返回值：
# reason 原样进日志，journalctl grep 'cell.pid 缺失'
# 才能命中；pid/uid 也在行里（排障用）。
_, reason = BROKER.peer_allowed_multi(1000, 200, [])
line = BROKER._degraded_warning(reason, 123, 1000)
self.assertIn("cell.pid 缺失", line)
self.assertIn("pid=123", line)
self.assertIn("uid=1000", line)
def test_read_cell_pids_glob_and_legacy(self):
with tempfile.TemporaryDirectory() as tmp:
with open(os.path.join(tmp, "cell-chv-cell-00.pid"),
"w") as fh:
fh.write("100")
with open(os.path.join(tmp, "cell-chv-cell-01.pid"),
"w") as fh:
fh.write("101")
with open(os.path.join(tmp, "cell.pid"), "w") as fh:
fh.write("101")
with open(os.path.join(tmp, "cell-chv-cell-02.pid"),
"w") as fh:
fh.write("not-a-pid")
self.assertEqual(BROKER._read_cell_pids(tmp), [100, 101])
if __name__ == "__main__":
unittest.main()
Reference in new issue View Git Blame Copy Permalink