host
0203f32627
定位结论（已定位，不是时序、不是日志路径变化）：part 4 提交
7a22732 把 serve() 从 peer_allowed 切到 peer_allowed_multi 时，
降级分支的 reason 与警告行都被改写成了“无 cell pid 文件”，
“cell.pid 缺失”这个串自此再也没进过 journal——P2-9 驱动
grep 'cell.pid 缺失' 自然命中不了，RETEST4 的
degraded_logged=False 即由此来。RETEST3 同项为 True，是因为
当时跑的还是 part 4 之前的 broker（serve 直接打印含该串的
警告行）。降级 ping 本身一直 ok=True，降级逻辑没坏，坏的是
日志里的认领字符串。
修复：peer_allowed_multi 无 pid 文件时的 reason 改回与单 cell
版 peer_allowed 完全一致的“uid 在范围内（cell.pid 缺失，未做
pid ns 对照）”；serve() 的警告行改走新 helper
_degraded_warning(reason, pid, uid)，把 reason 原样打进日志
（不再手写第二份文案，避免两者再次悄悄漂移；今后改文案只改
reason 一处）。journalctl grep 'cell.pid 缺失' 恢复命中。
新测试 cloud/tests/test_chv_muse_p2.py::BrokerMultiTest（2 项）：
reason 含“cell.pid 缺失”、警告行含该串且带 pid/uid。
1123 lines
47 KiB
Python
1123 lines
47 KiB
Python
"""对齐 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")
|
||
|
||
|
||
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 CellPart5WorktreeTest(unittest.TestCase):
|
||
"""part 5 修复 2（RETEST4 新发现 2）：/app/work 以可写 bind
|
||
盖住只读 /app——默认 dispatch 的 make_worktree 才写得进。"""
|
||
|
||
def test_worktree_bind_is_writable_under_workspace(self):
|
||
argv = CELL.build_nspawn_argv(["true"])
|
||
self.assertIn(
|
||
"--bind=/home/chv/workspace/.chv-worktrees:/app/work",
|
||
argv)
|
||
# 必须是可写 bind，不是只读绑。
|
||
self.assertNotIn(
|
||
"--bind-ro=/home/chv/workspace/.chv-worktrees:/app/work",
|
||
argv)
|
||
|
||
def test_worktree_bind_stacks_after_readonly_app(self):
|
||
# nspawn 按顺序挂载：work bind 必须在 --bind-ro=/app 之后，
|
||
# 才能盖住只读子树下的 /app/work。
|
||
argv = CELL.build_nspawn_argv(["true"])
|
||
self.assertLess(
|
||
argv.index("--bind-ro=/app"),
|
||
argv.index(
|
||
"--bind=/home/chv/workspace/.chv-worktrees:/app/work"))
|
||
|
||
def test_worktree_bind_custom_dir(self):
|
||
argv = CELL.build_nspawn_argv(["true"],
|
||
worktrees_dir="/tmp/wt")
|
||
self.assertIn("--bind=/tmp/wt:/app/work", argv)
|
||
|
||
def test_worktree_dir_constant_under_workspace(self):
|
||
self.assertTrue(
|
||
CELL.WORKTREE_DIR.startswith(CELL.WORKSPACE + "/"))
|
||
|
||
|
||
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()
|