host
c61484aa68
- chv-cell：systemd-nspawn 单命令单元，--private-users=1000:65536 （单元 root → 客机 chv），丢 8 项能力（含 SYS_PTRACE/NET_ADMIN）， syscall 过滤扣 io_uring，遮蔽 /etc/chv 与 playwright， --network-veth（10.200.0.0/24 + NAT），白名单环境透传 - chv-broker：Unix socket + SO_PEERCRED（uid 范围 + pid ns 对照）， 白名单 ping/sync/browser；sync 只调 chv-sync（runuser -u chv） - chv-browser：单元外独立进程，只暴露 open/snapshot/screenshot/ close/ping，返回无障碍树快照（截断）；evaluate/content 拒绝 - build.sh：装 systemd-container/iproute2/iptables，拷入三脚本， 建 chv-browser 用户与两个 systemd 单元；vsock_notify 条件拷入 - buildinfo：内容项加 P2 四项，支持可选 extra_contents - 测试：cloud/tests/test_chv_muse_p2.py 40 项全过；broker/browser 真实 socket 回路冒烟通过（root 连 broker 被拒） - 差距表：docs/chv-muse-p2-gap.md；真机验证待测试机空闲（round 11） 本地做，不推、不合、不部署
617 lines
24 KiB
Python
617 lines
24 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)
|
||
self.assertIn("--no-new-privileges", 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=")]
|
||
for path in ("/etc/chv", "/home/chv/.git-credentials",
|
||
"/usr/local/share/ms-playwright",
|
||
"/usr/local/lib/python3.12/dist-packages/playwright"):
|
||
self.assertIn(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_a11y_small_tree_untouched(self):
|
||
tree = {"role": "WebArea", "name": "Example",
|
||
"children": [{"role": "heading", "name": "Hi"}]}
|
||
out, truncated = BROWSER.truncate_a11y(tree)
|
||
self.assertFalse(truncated)
|
||
self.assertEqual(out, tree)
|
||
|
||
def test_truncate_a11y_node_cap(self):
|
||
tree = {"role": "WebArea",
|
||
"children": [{"role": "link", "name": "l%d" % i}
|
||
for i in range(300)]}
|
||
out, truncated = BROWSER.truncate_a11y(tree)
|
||
self.assertTrue(truncated)
|
||
self.assertLessEqual(len(out["children"]), 200)
|
||
|
||
def test_truncate_a11y_char_cap(self):
|
||
tree = {"role": "WebArea", "name": "x" * 70000, "children": []}
|
||
out, truncated = BROWSER.truncate_a11y(tree)
|
||
self.assertTrue(truncated)
|
||
self.assertTrue(out.get("truncated_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/")
|
||
self.assertEqual(resp["a11y"]["role"], "WebArea")
|
||
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"])
|
||
|
||
|
||
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"
|
||
|
||
@property
|
||
def accessibility(self):
|
||
return self
|
||
|
||
def snapshot(self):
|
||
return {"role": "WebArea", "name": "Example Domain",
|
||
"children": [{"role": "heading", "name": "Example"}]}
|
||
|
||
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 test_dry_run_contains_p2_steps(self):
|
||
# 在隔离的临时 git 仓里跑 build.sh --dry-run：工作树干净、
|
||
# HEAD 与 REPO_COMMIT 一致、假 codex 二进制报对版本。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
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"):
|
||
shutil.copy2(ROOT / "chv-guest" / name,
|
||
os.path.join(guest, name))
|
||
shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py"))
|
||
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,
|
||
})
|
||
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)
|
||
out = proc.stdout + proc.stderr
|
||
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"):
|
||
self.assertIn(needle, out, needle)
|
||
# P1 文件不在本分支：干跑只警告、不失败。
|
||
self.assertIn("vsock_notify.py 不在仓库里", out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|