admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
581e850eb7641a7efaf27204bf9a732c12dbd019
codex-app-server/cloud/tests/test_chv_residual.py
T
host 2d12179b9a fix(chv-muse-r11fix): authd 层 1 叶子树补 errors.py + stub __init__、层 2 seed 补三键 
层 1：真实的 cloud/chv/__init__.py import executor/orphans，拷进
authd 最小叶子树（/usr/local/lib/chv-authd）会让 chv-authd 在
import 时 ModuleNotFoundError 而 crash loop——先是 errors.py
缺失，补上后又在 executor 上失败（探针 fixprobe3）。新增
chv-guest/authd_init_stub.py（仅导出 ChvFailure），build.sh 拷
errors.py + stub 作叶子树的 __init__.py，不再拷真实包 __init__；
P3_LIBS_PRESENT 条件加 errors.py；EXTRA_CONTENTS 加 errors.py 与
__init__.py；两 lockfile 脚本的 DECLARED_SOURCES 同款登记。
层 2：_seed_env_entries 把 guest-env 里的
GITEA_URL/GITEA_USER/SYNC_REPO 抄进 authd-seed（build_server
要求四键齐全，只 GITEA_TOKEN 即 ValueError）；没有不补、不编
默认值。测试：seed 三键落盘/缺键不补、干跑夹具含新文件、lockfile
合成清单含新两项。
2026-11-07 03:56:24 +00:00

608 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""CHV 残留收尾轨（feat-chv-closeup）新增测试。
覆盖 SECURITY-REAUDIT §5 第 5 点的逐项闭环：
P2-20（_guest_env_delivered 加锁）、P2-19（net 登记失败
审计 + 拒起）、P2-17（net.json 删失败审计、脏标记写失败
审计 + 返回信号）、P2-18（固定 tmp 名 + 无 fsync 改随机
名原子写 + 双 fsync）、H1 附带（NO_PROXY 合并顺序）、
M4（generate 只读 BUILDINFO.versions、Chromium 单独钉死、
build.sh /app 最小集）。
全部用标准库与 mock，不启动真实 VM、不碰 WSL/真机；
M4 的“真实 lockfile 产出”仍待 V-9 真机构建，本文件只
验“生成端以 BUILDINFO 为唯一来源 + 钉死校验逻辑”，
版本号全部是合成假值，绝不编造真实版本号。
"""
import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest import mock
try:
from cloud.chv.atomic_file import atomic_write_text
from cloud.chv.disk import _save_snapshots
from cloud.chv.executor import CloudHypervisorExecutor
from cloud.chv.hostkeys import dirty_marker_path
from cloud.chv.orphans import _write_registry
from cloud.chv.ssh_transport import (
remote_command,
write_known_hosts,
)
from cloud.sandbox import ChvFailure
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.chv.atomic_file import atomic_write_text
from cloud.chv.disk import _save_snapshots
from cloud.chv.executor import CloudHypervisorExecutor
from cloud.chv.hostkeys import dirty_marker_path
from cloud.chv.orphans import _write_registry
from cloud.chv.ssh_transport import (
remote_command,
write_known_hosts,
)
from cloud.sandbox import ChvFailure
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_" + name[:-3], str(path))
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class FakeResult:
def __init__(self, returncode=0, stdout="", stderr=""):
self.returncode = returncode
self.stdout = stdout
self.stderr = stderr
class FakeProc:
def __init__(self, alive=True):
self._alive = alive
def poll(self):
return None if self._alive else 1
def terminate(self):
self._alive = False
def wait(self, timeout=None):
return 0
def kill(self):
self._alive = False
def _write(path, content=b"x"):
with open(path, "wb") as fh:
fh.write(content)
os.chmod(path, 0o755)
return path
class ResidualCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.dir = self.tmp.name
self.binary = _write(os.path.join(self.dir, "cloud-hypervisor"))
self.ch_remote = _write(os.path.join(self.dir, "ch-remote"))
self.kernel = _write(os.path.join(self.dir, "vmlinux"))
self.rootfs = _write(os.path.join(self.dir, "disk.raw"),
b"R" * 100)
self.ssh_key = _write(os.path.join(self.dir, "id_chv"))
self.kvm = _write(os.path.join(self.dir, "kvm"))
self.cpuinfo = os.path.join(self.dir, "cpuinfo")
with open(self.cpuinfo, "w") as fh:
fh.write("flags : fpu vmx sse\n")
self.state_dir = os.path.join(self.dir, "state")
self.calls = []
self.api_calls = []
self.audit = []
def tearDown(self):
self.tmp.cleanup()
def make_exec(self, env_extra=None, **overrides):
env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
"CHV_HOST_KEY_MODE": "keyscan",
"CHV_SSH_FORWARD": "1"}
if env_extra:
env.update(env_extra)
kwargs = {
"binary": self.binary, "ch_remote": self.ch_remote,
"kernel": self.kernel, "rootfs": self.rootfs,
"ssh_key": self.ssh_key, "kvm_path": self.kvm,
"cpuinfo_path": self.cpuinfo, "state_dir": self.state_dir,
"runner": self._default_runner,
"launcher": self._default_launcher,
"api_client": self._default_api,
"sleeper": lambda s: None, "env": env}
kwargs.update(overrides)
ex = CloudHypervisorExecutor(**kwargs)
ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
(e, t.get("id") if isinstance(t, dict) else t, d, k))
return ex
def _default_runner(self, cmd, timeout=None):
self.calls.append(list(cmd))
return FakeResult(0, "", "")
def _default_api(self, socket_path, method, path, payload=None):
self.api_calls.append((method, path, payload))
return (200, {})
def _default_launcher(self, cmd):
return FakeProc(alive=True)
def _task(self, task_id="abc123"):
return {"id": task_id, "prompt": "做一个加法函数",
"project_id": "proj1"}
def _audit_events(self):
return [e for e, _t, _d, _k in self.audit]
# -- P2-20 ---------------------------------------------------------
def test_guest_env_lock_exists_and_helpers_roundtrip(self):
ex = self.make_exec()
self.assertIsInstance(ex._guest_env_lock, type(threading.Lock()))
ex._mark_guest_env_delivered("chv-a", {"SUB2API_API_KEY"})
self.assertEqual(ex._delivered_secrets("chv-a"),
{"SUB2API_API_KEY"})
ex._mark_guest_env_delivered("chv-a", set())
self.assertEqual(ex._delivered_secrets("chv-a"), set())
ex._mark_guest_env_delivered("chv-a", {"K"})
ex._clear_guest_env_delivered("chv-a")
self.assertEqual(ex._delivered_secrets("chv-a"), set())
def test_guest_env_concurrent_mark_read_clear(self):
ex = self.make_exec()
errors = []
def worker(n):
try:
for i in range(50):
name = f"chv-w{i % 4}"
ex._mark_guest_env_delivered(name, {f"K{n}"})
got = ex._delivered_secrets(name)
self.assertIsInstance(got, set)
ex._clear_guest_env_delivered(name)
except Exception as exc: # noqa: BLE001 — 汇总后断言
errors.append(exc)
threads = [threading.Thread(target=worker, args=(n,))
for n in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
self.assertEqual(errors, [])
# 结束后确定性收尾：最后一次 mark 必须可见，带密钥的
# build_ssh_command 不再报 secret_env_not_delivered。
ex._mark_guest_env_delivered("chv-final", {"SUB2API_API_KEY"})
cmd = ex.build_ssh_command(
self._task("final"), ["true"],
extra_env={"SUB2API_API_KEY": "x"})
self.assertEqual(cmd[0], "ssh")
self.assertNotIn("x", cmd)
def test_build_ssh_command_sees_delivered_after_mark(self):
ex = self.make_exec()
ex._mark_guest_env_delivered("chv-abc123", {"SUB2API_API_KEY"})
cmd = ex.build_ssh_command(
self._task(), ["true"],
extra_env={"SUB2API_API_KEY": "x"})
self.assertNotIn("x", cmd)
# -- P2-19 ---------------------------------------------------------
def test_record_net_entries_failure_returns_false_and_audits(self):
ex = self.make_exec()
blocker = os.path.join(self.state_dir, "chv-x")
os.makedirs(self.state_dir, exist_ok=True)
with open(blocker, "w") as fh:
fh.write("x")
with self.assertLogs("cloud.cloud_hypervisor",
level="WARNING") as logs:
ok = ex._record_net_entries("chv-x", [{"tap": "tap0"}])
self.assertFalse(ok)
self.assertTrue(any("net_registry_write_failed" in line
for line in logs.output))
self.assertIn("net_registry_write_failed",
self._audit_events())
def test_boot_vm_refuses_when_net_registry_fails(self):
ex = self.make_exec(env_extra={
"CHV_NET_JSON": '[{"tap": "tap0", "ip": "10.0.0.2"}]'})
os.makedirs(self.state_dir, exist_ok=True)
with open(os.path.join(self.state_dir, "chv-abc123"),
"w") as fh:
fh.write("x")
with self.assertRaises(ChvFailure) as ctx:
ex._boot_vm(self._task(), "chv-abc123", self.rootfs,
None, [{"tap": "tap0", "ip": "10.0.0.2"}])
self.assertEqual(ctx.exception.kind,
"net_registry_write_failed")
paths = [p for _m, p, _pl in self.api_calls]
self.assertIn("/vm.create", paths)
self.assertNotIn("/vm.boot", paths)
self.assertIn("net_registry_write_failed",
self._audit_events())
# -- P2-17 ---------------------------------------------------------
def test_cleanup_net_remove_failure_audited(self):
ex = self.make_exec()
vm_dir = os.path.join(self.state_dir, "chv-abc123")
os.makedirs(vm_dir, exist_ok=True)
os.makedirs(os.path.join(vm_dir, "net.json"), exist_ok=True)
with open(os.path.join(self.state_dir, "chv-abc123.sock"),
"w") as fh:
fh.write("")
self.assertTrue(ex.stop_vm(self._task()))
marker = dirty_marker_path(self.state_dir, "chv-abc123")
self.assertTrue(os.path.exists(marker))
self.assertIn("vm_net_registry_delete_failed",
self._audit_events())
def test_write_dirty_marker_failure_returns_false_and_audits(self):
blocker = os.path.join(self.dir, "blocker3")
with open(blocker, "w") as fh:
fh.write("x")
ex = self.make_exec(state_dir=blocker)
ok = ex._write_dirty_marker("chv-abc123", True) # 不抛
self.assertFalse(ok)
self.assertIn("dirty_marker_write_failed",
self._audit_events())
def test_write_dirty_marker_success_roundtrip(self):
ex = self.make_exec()
self.assertTrue(ex._write_dirty_marker("chv-abc123", True))
marker = dirty_marker_path(self.state_dir, "chv-abc123")
self.assertTrue(os.path.exists(marker))
self.assertTrue(ex._write_dirty_marker("chv-abc123", False))
self.assertFalse(os.path.exists(marker))
# -- P2-18 ---------------------------------------------------------
def test_atomic_write_no_fixed_tmp_and_fsync(self):
target = os.path.join(self.dir, "probe.json")
calls = {"n": 0}
real_fsync = os.fsync
def _counting(fd):
calls["n"] += 1
return real_fsync(fd)
with mock.patch.object(os, "fsync", side_effect=_counting):
atomic_write_text(target, '{"a": 1}')
with open(target) as fh:
self.assertEqual(fh.read(), '{"a": 1}')
self.assertEqual(oct(os.stat(target).st_mode & 0o777), "0o600")
leftovers = [p for p in os.listdir(self.dir)
if ".tmp" in p]
self.assertEqual(leftovers, [])
# 文件 fsync + 目录 fsync，至少两次。
self.assertGreaterEqual(calls["n"], 2)
def test_net_record_leaves_no_tmp_files(self):
ex = self.make_exec()
self.assertTrue(ex._record_net_entries(
"chv-abc123", [{"tap": "tap0"}]))
vm_dir = os.path.join(self.state_dir, "chv-abc123")
with open(os.path.join(vm_dir, "net.json")) as fh:
self.assertEqual(json.load(fh), [{"tap": "tap0"}])
self.assertEqual([p for p in os.listdir(vm_dir)
if ".tmp" in p], [])
def test_known_hosts_atomic_no_fixed_tmp(self):
path = os.path.join(self.state_dir, "chv-abc123",
"known_hosts")
write_known_hosts(path, ["127.0.0.1 ssh-ed25519 AAAA"])
with open(path) as fh:
self.assertEqual(fh.read(),
"127.0.0.1 ssh-ed25519 AAAA\n")
self.assertEqual(oct(os.stat(path).st_mode & 0o777), "0o600")
vm_dir = os.path.join(self.state_dir, "chv-abc123")
self.assertEqual([p for p in os.listdir(vm_dir)
if ".tmp" in p], [])
def test_snapshot_save_no_fixed_tmp(self):
_save_snapshots(self.state_dir, "chv-abc123",
[{"url": "file://x"}])
path = os.path.join(self.state_dir, "chv-abc123",
"snapshots.json")
with open(path) as fh:
self.assertEqual(json.load(fh), [{"url": "file://x"}])
vm_dir = os.path.join(self.state_dir, "chv-abc123")
self.assertEqual([p for p in os.listdir(vm_dir)
if ".tmp" in p], [])
def test_orphan_registry_no_fixed_tmp(self):
path = os.path.join(self.state_dir, "pool.json")
_write_registry(path, {"chv-a": {"vm": "chv-a"}})
with open(path) as fh:
self.assertEqual(json.load(fh),
{"chv-a": {"vm": "chv-a"}})
self.assertEqual([p for p in os.listdir(self.state_dir)
if ".tmp" in p], [])
# -- H1 附带：NO_PROXY ---------------------------------------------
def _env_assignments(self, tokens):
"""remote_command 返回的 env 赋值（KEY=VALUE 形式）。"""
out = {}
for tok in tokens:
if tok == "env":
continue
if "=" in tok and not tok.startswith("'"):
key, _, value = tok.partition("=")
if key.isidentifier() or key in (
"NO_PROXY", "no_proxy", "HTTPS_PROXY",
"HTTP_PROXY", "https_proxy", "http_proxy"):
out.setdefault(key, []).append(
value.strip("'"))
return out
def test_no_proxy_merged_with_egress(self):
tokens = remote_command(
["true"], egress_proxy="http://10.0.0.1:8888",
extra_env={"NO_PROXY": "internal.example.com,10.0.0.0/8"})
assigns = self._env_assignments(tokens)
self.assertEqual(len(assigns.get("NO_PROXY", [])), 1)
merged = assigns["NO_PROXY"][0]
for part in ("internal.example.com", "10.0.0.0/8",
"localhost", "127.0.0.1"):
self.assertIn(part, merged)
self.assertEqual(
assigns["HTTPS_PROXY"], ["http://10.0.0.1:8888"])
def test_no_proxy_lowercase_merged_and_not_duplicated(self):
tokens = remote_command(
["true"], egress_proxy="http://10.0.0.1:8888",
extra_env={"no_proxy": "gitea.tose.top",
"NO_PROXY": "localhost"})
assigns = self._env_assignments(tokens)
self.assertEqual(len(assigns.get("NO_PROXY", [])), 1)
self.assertIn("gitea.tose.top", assigns["NO_PROXY"][0])
self.assertNotIn("no_proxy", assigns)
# localhost 去重后只出现一次。
self.assertEqual(
assigns["NO_PROXY"][0].split(",").count("localhost"), 1)
def test_no_proxy_preserved_without_egress(self):
tokens = remote_command(
["true"], extra_env={"NO_PROXY": "a.example"})
assigns = self._env_assignments(tokens)
self.assertEqual(assigns.get("NO_PROXY"), ["a.example"])
def test_user_proxy_dropped_when_egress_set(self):
tokens = remote_command(
["true"], egress_proxy="http://10.0.0.1:8888",
extra_env={"HTTPS_PROXY": "http://evil:1",
"https_proxy": "http://evil:1",
"HTTP_PROXY": "http://evil:1"})
assigns = self._env_assignments(tokens)
self.assertEqual(
assigns["HTTPS_PROXY"], ["http://10.0.0.1:8888"])
self.assertEqual(
assigns["HTTP_PROXY"], ["http://10.0.0.1:8888"])
self.assertNotIn("https_proxy", assigns)
self.assertNotIn("http_proxy", assigns)
# -- M4 供应链 ----------------------------------------------------------
class LockfileTest(unittest.TestCase):
"""合成 BUILDINFO 干跑：版本全是假值（如 git 9.9.9），
只验“来源是 BUILDINFO、不是宿主现场”与钉死校验逻辑，
真实 lockfile 产出仍待 V-9 真机构建。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.dir = self.tmp.name
def tearDown(self):
self.tmp.cleanup()
def _synth_buildinfo(self, chromium="chromium_headless_shell-1248",
pinned_rev="1248"):
dpkg = ["git 9.9.9-9ubuntu9", "openssh-server 1:9.9",
"openssh-client 1:9.9", "cloud-init 9.9",
"python3 9.9", "python3-pip 9.9", "curl 9.9",
"ca-certificates 9.9", "xz-utils 9.9",
"fonts-wqy-microhei 9.9", "fontconfig 9.9",
"systemd-container 9.9"]
info = {
"tag": "chv-guest-24.04-0.164.0-alpha.9-abc1234",
"ubuntu": "24.04",
"codex_version": "0.164.0-alpha.9",
"repo_commit": "abc1234",
"gt_version": "1.0.0",
"built_at": "2026-11-07T00:00:00Z",
"contents": ["git", "openssh-server", "openssh-client",
"cloud-init", "python3", "python3-pip",
"curl", "ca-certificates", "xz-utils",
"fonts-wqy-microhei", "fontconfig", "codex",
"codex-code-mode-host", "codex-resources/bwrap",
"/app/host.py", "gt", "chv-sync",
# 与 717efc4 真实 BUILDINFO.contents 对齐：
# P2 三件套 + P3 authd 三件套 + 叶子模块 +
# vsock_notify（缺 DECLARED_SOURCES 登记时
# generate 记 unresolved、verify/check 失败，
# 合成清单须覆盖才拦得住回归）。round-11
# 真机修复（层 1）后叶子模块还含 errors.py
# 与 stub __init__.py。
"systemd-container", "chv-cell", "chv-broker",
"chv-browser", "chv-authd",
"git-credential-chv-authd", "chv-authd.service",
"cloud/chv/authd.py", "cloud/chv/sentinel.py",
"cloud/chv/errors.py", "cloud/chv/__init__.py",
"vsock_notify",
"playwright", "chromium-headless-shell"],
"versions": {"dpkg": dpkg,
"pip": ["playwright==9.9.9"],
"gt": ["gt version 1.0.0"],
"chromium": [chromium]},
"chromium_pinned_rev": pinned_rev,
"image_sha256": "0" * 64,
}
path = os.path.join(self.dir, "BUILDINFO.json")
with open(path, "w") as fh:
json.dump(info, fh)
return path
def test_generate_uses_buildinfo_not_host(self):
gen = _load_script("generate_lockfile.py")
buildinfo = self._synth_buildinfo()
lockfile = os.path.join(self.dir, "lockfile.json")
self.assertEqual(gen.generate(buildinfo, lockfile), 0)
with open(lockfile) as fh:
data = json.load(fh)
self.assertEqual(data["unresolved"], [])
# 假值进 lockfile：来源是 BUILDINFO，不是宿主 dpkg。
self.assertIn("git 9.9.9-9ubuntu9",
data["versions"]["dpkg"])
self.assertIn("playwright 9.9.9",
data["versions"]["pip"])
self.assertIn("chromium-headless-shell "
"chromium_headless_shell-1248",
data["versions"]["declared"])
self.assertEqual(
gen.verify(lockfile, buildinfo), 0)
def test_generate_verify_rejects_wrong_pinned_rev(self):
gen = _load_script("generate_lockfile.py")
buildinfo = self._synth_buildinfo(
chromium="chromium_headless_shell-9999",
pinned_rev="1248")
lockfile = os.path.join(self.dir, "lockfile.json")
self.assertEqual(gen.generate(buildinfo, lockfile), 0)
# 生成本身不设门（只记版本），--verify 才拦钉死不符。
self.assertEqual(gen.verify(lockfile, buildinfo), 1)
def test_check_lockfile_pinned_rev(self):
check = _load_script("check_lockfile.py")
ok = self._synth_buildinfo()
self.assertEqual(
check.main(["check_lockfile.py", ok]), 0)
bad = self._synth_buildinfo(
chromium="chromium_headless_shell-9999",
pinned_rev="1248")
self.assertEqual(
check.main(["check_lockfile.py", bad]), 1)
def test_buildinfo_records_and_enforces_pin(self):
buildinfo = _load_script("buildinfo.py")
out = os.path.join(self.dir, "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")
args = [out, "chv-guest-24.04-0.164.0-alpha.9-abc1234",
"24.04", "0.164.0-alpha.9", "abc1234", "0" * 64,
"1", "1248"]
info = buildinfo.build(*args[0:7], "1248")
self.assertEqual(info["chromium_pinned_rev"], "1248")
self.assertEqual(buildinfo.main(["buildinfo.py"] + args), 0)
wrong = list(args)
wrong[-1] = "9999"
self.assertEqual(
buildinfo.main(["buildinfo.py"] + wrong), 1)
# 未给钉死值：记 null，不强制。
info2 = buildinfo.build(*args[0:7])
self.assertIsNone(info2["chromium_pinned_rev"])
class BuildScriptMinimalAppTest(unittest.TestCase):
"""build.sh 干跑：/app 只剩 host.py（不用 cloud/ 也能
干跑通过），Chromium 钉死值透传进 buildinfo 回显。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.dir = self.tmp.name
self.bin_dir = os.path.join(self.dir, "bin")
os.makedirs(os.path.join(self.bin_dir, "codex-resources"))
for name in ("codex", "codex-code-mode-host"):
p = os.path.join(self.bin_dir, name)
with open(p, "w") as fh:
fh.write("#!/bin/sh\necho \"codex-cli 0.164.0-alpha.9\"\n")
os.chmod(p, 0o755)
self.repo = os.path.join(self.dir, "repo")
os.makedirs(self.repo)
# 故意不建 cloud/：最小集收紧后 build.sh 不再要求它。
with open(os.path.join(self.repo, "host.py"), "w") as fh:
fh.write("# host\n")
subprocess.run(["git", "init", "-q", self.repo],
check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo, "config",
"user.email", "t@t"], check=True,
capture_output=True)
subprocess.run(["git", "-C", self.repo, "config",
"user.name", "t"], check=True,
capture_output=True)
subprocess.run(["git", "-C", self.repo, "add", "-A"],
check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo, "commit", "-qm",
"init"], check=True, capture_output=True)
self.repo_commit = subprocess.run(
["git", "-C", self.repo, "rev-parse", "--short", "HEAD"],
check=True, capture_output=True, text=True).stdout.strip()
self.script = str(ROOT / "chv-guest" / "build.sh")
def tearDown(self):
self.tmp.cleanup()
def _run(self, args, env_extra=None):
env = dict(os.environ)
env.update({"CODEX_BIN_DIR": self.bin_dir,
"UBUNTU_VERSION": "24.04",
"CODEX_VERSION": "0.164.0-alpha.9",
"REPO_COMMIT": self.repo_commit,
"UBUNTU_IMAGE_SHA256": "0" * 64,
"REPO_DIR": self.repo})
if env_extra:
env.update(env_extra)
return subprocess.run(["bash", self.script] + args,
check=False,
capture_output=True, text=True, env=env)
def test_dry_run_without_cloud_dir_and_pin_echo(self):
out_dir = os.path.join(self.dir, "out")
r = self._run(["--dry-run", out_dir],
{"CHROMIUM_HEADLESS_SHELL_REV": "1248"})
self.assertEqual(r.returncode, 0, r.stderr)
self.assertNotIn("cloud:/app", r.stdout)
self.assertIn("host.py:/app", r.stdout)
self.assertIn("1248", r.stdout)
self.assertFalse(os.path.exists(
os.path.join(out_dir, "disk.raw")))
def test_build_sh_text_has_no_cloud_copy(self):
with open(self.script) as fh:
text = fh.read()
self.assertNotIn('"$REPO_DIR/cloud:/app/"', text)
self.assertIn('"$REPO_DIR/host.py:/app/"', text)
if __name__ == "__main__":
unittest.main()
Reference in new issue View Git Blame Copy Permalink