admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
50cae2a0aff3ab87899a8e47bc5cb60e47d78684
codex-app-server/cloud/tests/test_chv_muse_p2.py
T
host 50cae2a0af part 6 修复：worktree 根移出 /app——RETEST5 挂载点回归 
RETEST5：part 5 的 --bind=<workspace>/.chv-worktrees:/app/work
（adc7bf6，排在 --bind-ro=/app 之后）让所有 cell 启动失败：
后挂的 bind 只有在挂载点已存在时才能盖住，挂载点不存在时
nspawn 在只读 /app 下连建都建不起来（Failed to create mount
point .../app/work: Read-only file system，全量 73 次，
CELL-RC=0 共 0 次），缺陷 1 的 15/15 回退为 0/15。

三选一取 (a) worktree 根移出 /app：
- chv-cell 不再发任何指向 /app 之下的 bind，删
  --bind=...:/app/work 与 worktrees_dir 参数；WORKTREE_DIR
  （/home/chv/workspace/.chv-worktrees）改经
  --setenv=CHV_WORKTREE_ROOT 传进单元——该路径在已可写绑进
  单元的 workspace 之下，挂载点问题按构造即不存在；backing
  仍在客机 workspace 下随磁盘持久化，隔离与落盘语义不变，
  只是单元内路径从 /app/work/host-worktrees 改为
  /home/chv/workspace/.chv-worktrees/host-worktrees。
- host.py 新增 worktree_base()：CHV_WORKTREE_ROOT 优先；
  CHV_CELL=1 但变量未设时回落到同一单元内路径（旧版 chv-cell
  / 单元内手动跑 host.py 仍不出 /app）；单元外沿用
  HERE/work/host-worktrees，行为不变。make_worktree 与
  dispatch_from_branch（第二个 HERE/work 调用方）同走该函数，
  goal/fork/依赖链一并修到。
- 不选 (b) 默认 --no-worktree：隔离丢了，且 goal/fork 同样
  调 make_worktree，照样撞 Errno 30；不选 (c) 镜像预建空
  目录 /app/work：修法耦合镜像构建，某次重建漏建即静默回归，
  且单测无法钉住镜像内容。
- 根因 1（39084ce）与根因 3（0203f32）代码未动。

单测：删 part 5 钉 argv 顺序的 4 项（该顺序本身已被真机证伪），
新增 CellPart6WorktreeTest 4 项 + HostWorktreeBaseTest 5 项：
argv 里无任何指向 /app 之下的 bind、CHV_WORKTREE_ROOT 经
setenv 进单元、worktree 根只经可写 workspace 绑定可达、
两处同值常量一致、worktree_base 三路选择、make_worktree
真 git worktree 落点、dispatch_from_branch 同根。

全量：cloud/ 下 discover -s tests，2067 项 OK（skipped=9）。
（注：仓库根跑 discover 会把 queue 解析成标准库、致
test_queue_sandbox 导入失败，须在 cloud/ 下跑。）
2026-11-07 14:23:48 +00:00

1268 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/假进程/假
playwright，绝不连真实网络、不起真实 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")
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 = _load_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(BROWSER.validate_url("https://example.com")[0])
self.assertTrue(BROWSER.validate_url("http://example.com/x")[0])
self.assertTrue(
BROWSER.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, _ = BROWSER.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 = BROWSER.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 = BROWSER.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 = BROWSER.truncate_aria(text)
self.assertTrue(truncated)
self.assertLessEqual(len(out), BROWSER.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_loading_playwright(self):
def exploding_factory():
raise AssertionError("坏 URL 不应加载 playwright")
agent = BROWSER.BrowserAgent(playwright_factory=exploding_factory)
resp = agent.open("https://127.0.0.1/secret")
self.assertFalse(resp["ok"])
def test_snapshot_and_screenshot_need_open_first(self):
agent = BROWSER.BrowserAgent(
playwright_factory=lambda: (_ for _ in ()).throw(
AssertionError("不应加载 playwright")))
self.assertFalse(agent.snapshot()["ok"])
self.assertFalse(agent.screenshot()["ok"])
self.assertTrue(agent.close()["ok"])
def test_open_returns_a11y_snapshot_with_fake_playwright(self):
page = FakePage()
with tempfile.TemporaryDirectory() as tmp:
agent = BROWSER.BrowserAgent(
playwright_factory=lambda: FakePlaywright(page),
shots_dir=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-11 复测纠正（第二轮）：a11y 是 ARIA 快照字符串
#（Playwright Page.aria_snapshot()），不是 dict 树。
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_load_playwright_returns_called_cm(self):
# round-11 复测回归：真实 _load_playwright 曾返回裸
# sync_playwright 函数（未调用），_ensure_browser 对它调
# __enter__ 即 AttributeError，open 永远失败；单测的假工厂
# 直接返回带 __enter__ 的对象，盖不住这个形状。现
# _load_playwright 必须返回调用后的上下文管理器。
import types
calls = []
class FakeCM:
def __enter__(self):
return self
def __exit__(self, *exc):
return False
cm = FakeCM()
def sync_playwright():
calls.append("called")
return cm
fake_api = types.ModuleType("playwright.sync_api")
fake_api.sync_playwright = sync_playwright
fake_pkg = types.ModuleType("playwright")
fake_pkg.sync_api = fake_api
sys.modules["playwright"] = fake_pkg
sys.modules["playwright.sync_api"] = fake_api
self.addCleanup(sys.modules.pop, "playwright.sync_api", None)
self.addCleanup(sys.modules.pop, "playwright", None)
self.assertIs(BROWSER._load_playwright(), cm)
self.assertEqual(calls, ["called"])
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 FakePage:
def __init__(self):
self.url = "https://example.com/"
self._closed = False
def goto(self, url, timeout=None):
self.url = url.rstrip("/") + "/"
def wait_for_load_state(self, state, timeout=None):
pass
def title(self):
return "Example Domain"
# 真实 Playwright 1.64.1 的快照入口是 Page.aria_snapshot()
#（返回字符串）；旧 fake 的 accessibility.snapshot() 是
# Puppeteer 的形状，真实库里不存在，round-11 第二轮复测
# 已证实会报 'Page' object has no attribute 'accessibility'。
def aria_snapshot(self):
return ("- generic [ref=e1]:\n"
" - heading \"Example Domain\" [level=1] [ref=e2]")
def screenshot(self, path=None):
with open(path, "wb") as fh:
fh.write(b"\x89PNG")
def close(self):
self._closed = True
class FakeBrowser:
def __init__(self, page):
self._page = page
def new_page(self):
return self._page
def close(self):
pass
class FakeEntered:
def __init__(self, page):
self.chromium = mock.Mock()
self.chromium.launch.return_value = FakeBrowser(page)
class FakePlaywright:
def __init__(self, page):
self._page = page
def __enter__(self):
return FakeEntered(self._page)
def __exit__(self, *exc):
return False
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"):
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"):
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