"""对齐 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//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: mtu 1500\n" "3: ve-chv-cell-01@if3: 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) 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()