"""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 失败，
                         # 合成清单须覆盖才拦得住回归）。
                         "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",
                         "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()