admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(chv): P2-20/P2-19/P2-17/P2-18 残留收尾 + NO_PROXY 合并

Browse Source 
- P2-20：_guest_env_delivered 加专用锁，读写只走
  _delivered_secrets/_mark/_clear 三个入口，并发 hammer 测试钉死
- P2-19：_record_net_entries 失败发 net_registry_write_failed
  独立审计并返回 False；_boot_vm 见 False 即拒起（wake 随后清半醒 VM）
- P2-17：net.json 删失败发 vm_net_registry_delete_failed 审计；
  _write_dirty_marker 失败发 dirty_marker_write_failed 审计并返回
  False，stop 不抛的契约不变
- P2-18：新增 cloud/chv/atomic_file.py（mkstemp 随机名 + 文件
  fsync + rename + 目录 fsync），net.json/known_hosts/
  snapshots.json/pool registry/脏标记/spill 全部走它，固定 .tmp 清零
- H1 附带：NO_PROXY 实测确有覆盖 bug（env 重复赋值以后者为准，
  调用方 NO_PROXY 被 egress 的 localhost 值静默覆盖），改为合并
  单条发出；egress 已设时调用方代理变量（含小写）一律不发
- 测试：新增 cloud/tests/test_chv_residual.py（23 项），旧 closeup
  两处钉死旧语义的测试按新语义更新
This commit is contained in:
host committed 2026-11-06 18:41:50 +00:00
1 parent ba5df980bf
commit c39715659e
7 files changed
+850 -51
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cloud/tests/test_chv_residual.py
+593
View File
@@ -0,0 +1,593 @@
"""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"]
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",
"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()