Files
hermes-agent/tests/hermes_cli/test_install_cua_driver.py
T
ethernet 30da5d0a89 test: run os-specific tests on their real host, not a faked one
many tests patched sys.platform or a module's _IS_WINDOWS flag, then
ran on linux ci. the patch selects the branch under test, but the host
does not have the behavior the branch exists for. the test proves the
patch, not the platform. some gated assertions never ran on any host.

this commit adds three markers: linux_only, macos_only, windows_only.
a conftest hook skips a marked test on the other hosts, with a clear
reason. no test fakes a host now. two documented fakes remain
(android/termux, freebsd) because no ci runner exists for them.

each fake site got one of four treatments:
- gate it: the real host supplies the platform; mocks cover real
  dependencies only, never host identity
- patch the module's own probe when the subject is the probe's consumer
- assert against the real host when the fake stood in for any non-x host
- delete the patch when it set the value the host already has

bare skipif(sys.platform != ...) guards became markers too. the lane
model skips these on linux and never imports them on windows, so they
ran on no host. platform parametrize tables are now one marked test
per os.

running on real hosts found real errors: a chrome-sandbox failure in
test_gui_command that main hides, and two windows failures fixed here.
the agents.md testing section now documents the policy.
2026-08-09 22:09:49 -04:00

1096 lines
45 KiB
Python

"""Tests for ``install_cua_driver`` upgrade semantics.
The cua-driver upstream installer always pulls the latest release tag, so
re-running it is the canonical upgrade path. ``install_cua_driver(upgrade=True)``
must:
* Be supported-platform-only — no-op silently elsewhere so ``hermes update``
can call it unconditionally without warning unsupported-platform users.
* Re-run the installer even when the binary is already on PATH (this is the
fix for the "we only pulled cua-driver once on enable" complaint).
* Preserve original ``upgrade=False`` behaviour for the toolset-enable flow:
skip if installed, install otherwise, warn on unsupported platforms.
The pre-install arch probe that used to live alongside this function was
deleted (see top-of-file comment in tools_config.py) — the upstream
installer has CUA_DRIVER_RS_BAKED_VERSION baked in by CD and errors
cleanly on missing-arch assets, and the upgrade path uses
``cua_driver_update_check()`` (which shells `cua-driver check-update
--json` against the already-installed binary).
"""
from __future__ import annotations
import sys
from types import SimpleNamespace
from unittest.mock import patch
import pytest
class TestInstallCuaDriverUpgrade:
# ``install_cua_driver`` supports macOS, Windows AND Linux. For everything
# below except the two unsupported-platform cases, the Linux host takes a
# byte-identical path to macOS — same ``fetch_tool`` ("curl"), same
# ``_cua_install_target_writable()`` verdict, same branch — so the old
# ``patch("platform.system", return_value="Darwin")`` bought nothing but a
# fake host. Dropped, and the names no longer claim macOS.
def test_upgrade_on_unsupported_platform_is_silent_noop(self):
"""The one branch no CI runner can reach for real.
``platform.system`` is still faked here, deliberately and narrowly: we
run Linux/macOS/Windows lanes, and every one of them is a *supported*
platform, so the refusal path is unreachable on all three. The fake is
sound because the function returns before touching any OS facility —
no subprocess, no path handling, no import — so there is nothing
underneath the branch for a real host to falsify.
"""
from hermes_cli import tools_config
with patch.object(tools_config, "_print_warning") as warn, \
patch("platform.system", return_value="FreeBSD"):
assert tools_config.install_cua_driver(upgrade=True) is False
warn.assert_not_called()
def test_non_upgrade_on_unsupported_platform_warns(self):
"""Same narrow exception as above — see that test's docstring."""
from hermes_cli import tools_config
with patch.object(tools_config, "_print_warning") as warn, \
patch("platform.system", return_value="FreeBSD"):
assert tools_config.install_cua_driver(upgrade=False) is False
warn.assert_called()
def test_upgrade_with_binary_present_runs_installer(self):
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/usr/local/bin/" + n
if n in {"cua-driver", "curl"} else None), \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner, \
patch("subprocess.run"):
assert tools_config.install_cua_driver(upgrade=True) is True
runner.assert_called_once()
kwargs = runner.call_args.kwargs
assert kwargs.get("verbose") is False
def test_upgrade_without_binary_runs_installer(self):
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/usr/bin/curl" if n == "curl" else None), \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner:
assert tools_config.install_cua_driver(upgrade=True) is True
runner.assert_called_once()
@pytest.mark.linux_only
def test_quiet_refresh_prints_single_contextual_progress_line(self):
"""``linux_only``: reaches Popen through the POSIX download-then-exec
branch, which this lane takes for real."""
from unittest.mock import MagicMock
from hermes_cli import tools_config
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 0
fake_proc.communicate.return_value = ("", None)
with patch(
"subprocess.run",
return_value=MagicMock(returncode=0, stderr=""),
), \
patch("subprocess.Popen", return_value=fake_proc), \
patch.object(
tools_config.shutil,
"which",
return_value="/usr/local/bin/cua-driver",
), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_info") as info:
assert tools_config._run_cua_driver_installer(
label="Refreshing",
verbose=False,
) is True
info.assert_called_once_with(
"→ Refreshing cua-driver (Computer Use)..."
)
@pytest.mark.linux_only
def test_quiet_refresh_can_suppress_progress_line(self):
"""``linux_only``: same POSIX Popen path as the test above."""
from unittest.mock import MagicMock
from hermes_cli import tools_config
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 0
fake_proc.communicate.return_value = ("", None)
with patch(
"subprocess.run",
return_value=MagicMock(returncode=0, stderr=""),
), \
patch("subprocess.Popen", return_value=fake_proc), \
patch.object(
tools_config.shutil,
"which",
return_value="/usr/local/bin/cua-driver",
), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_info") as info:
assert tools_config._run_cua_driver_installer(
label="Refreshing",
verbose=False,
show_progress=False,
) is True
info.assert_not_called()
def test_upgrade_can_suppress_installer_progress(self):
from hermes_cli import tools_config
with patch.object(
tools_config.shutil,
"which",
side_effect=lambda name: (
f"/usr/local/bin/{name}"
if name in {"cua-driver", "curl"}
else None
),
), \
patch.object(
tools_config,
"_run_cua_driver_installer",
return_value=True,
) as runner, \
patch("subprocess.run"):
assert tools_config.install_cua_driver(
upgrade=True,
show_installer_progress=False,
) is True
assert runner.call_args.kwargs["show_progress"] is False
def test_upgrade_non_writable_install_target_skips_refresh(self):
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/usr/local/bin/" + n
if n in {"cua-driver", "curl"} else None), \
patch.object(tools_config, "_cua_install_target_writable",
return_value=False), \
patch.object(tools_config, "_run_cua_driver_installer") as runner, \
patch.object(tools_config, "_print_info") as info:
assert tools_config.install_cua_driver(upgrade=True) is True
runner.assert_not_called()
assert any(
"/Applications is not writable" in call.args[0]
for call in info.call_args_list
)
def test_fresh_install_non_writable_install_target_skips_install(self):
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/usr/bin/curl" if n == "curl" else None), \
patch.object(tools_config, "_cua_install_target_writable",
return_value=False), \
patch.object(tools_config, "_run_cua_driver_installer") as runner, \
patch.object(tools_config, "_print_info") as info:
assert tools_config.install_cua_driver(upgrade=False) is False
runner.assert_not_called()
assert any(
"/Applications is not writable" in call.args[0]
for call in info.call_args_list
)
@pytest.mark.macos_only
def test_install_target_writability_is_probed_for_real_on_macos(self):
"""The ``_cua_install_target_writable`` seam the two tests above patch.
``macos_only``: ``/Applications`` is the only install target Hermes
checks, and the probe short-circuits to True on every other platform —
so this is the one host where the real filesystem answer means
anything.
"""
import os
from hermes_cli import tools_config
writable = tools_config._cua_install_target_writable()
if os.path.isdir("/Applications"):
assert writable is os.access("/Applications", os.W_OK)
else:
assert writable is True
def test_non_upgrade_with_binary_skips_install(self):
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/usr/local/bin/" + n
if n in {"cua-driver", "curl"} else None), \
patch.object(tools_config, "_run_cua_driver_installer") as runner, \
patch("subprocess.run"):
assert tools_config.install_cua_driver(upgrade=False) is True
runner.assert_not_called()
def test_non_upgrade_without_binary_runs_installer(self):
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/usr/bin/curl" if n == "curl" else None), \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner:
assert tools_config.install_cua_driver(upgrade=False) is True
runner.assert_called_once()
class TestRequireConfirmedUpdate:
"""`hermes update` passes require_confirmed_update=True: the full
upstream installer (multi-minute, output captured, plus install.ps1's
600s lock window on Windows) may only run when the driver's native
``check-update`` verb positively confirms a newer release. An
indeterminate check (old driver, offline, GitHub rate-limited, probe
timeout) keeps the installed version and returns fast.
Explicit `hermes computer-use install --upgrade` keeps the old
fall-through (require_confirmed_update=False): a force-refresh should
still reinstall when the check can't answer.
"""
def _install(self, check_state, require_confirmed):
"""Drive ``install_cua_driver`` on the host, whatever it is.
The old signature took a ``system`` string and faked
``platform.system`` with it, so callers picked "Windows"/"Darwin"
arbitrarily. Nothing in the confirmed-update gate is
platform-dependent — it's ``check-update`` state plus a flag — so the
fake only decided which lie the test told itself. ``which`` answers
for every fetch tool so the host's own branch resolves cleanly.
"""
from unittest.mock import MagicMock
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/x/" + n
if n in {"cua-driver", "curl", "powershell"} else None), \
patch.object(tools_config, "_resolved_cua_driver_cmd",
return_value="/x/cua-driver"), \
patch.object(tools_config, "_cua_install_target_writable",
return_value=True), \
patch("tools.computer_use.cua_backend.cua_driver_update_check",
return_value=check_state), \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner, \
patch("subprocess.run",
return_value=MagicMock(stdout="cua-driver 0.5.0", returncode=0)), \
patch.object(tools_config, "_print_success"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info") as info:
ok = tools_config.install_cua_driver(
upgrade=True, require_confirmed_update=require_confirmed
)
return ok, runner, info
def test_indeterminate_check_keeps_installed_version(self):
ok, runner, info = self._install(None, require_confirmed=True)
assert ok is True
runner.assert_not_called()
assert any(
"keeping the installed version" in call.args[0]
for call in info.call_args_list
)
def test_indeterminate_check_points_at_force_path(self):
ok, runner, info = self._install(None, require_confirmed=True)
assert ok is True
runner.assert_not_called()
assert any(
"computer-use install --upgrade" in call.args[0]
for call in info.call_args_list
)
def test_confirmed_update_still_runs_installer(self):
state = {"current_version": "0.5.0", "latest_version": "0.6.0",
"update_available": True}
ok, runner, _ = self._install(state, require_confirmed=True)
assert ok is True
runner.assert_called_once()
def test_up_to_date_short_circuits(self):
state = {"current_version": "0.6.0", "latest_version": "0.6.0",
"update_available": False}
ok, runner, _ = self._install(state, require_confirmed=True)
assert ok is True
runner.assert_not_called()
def test_explicit_upgrade_still_falls_through_on_indeterminate(self):
# `hermes computer-use install --upgrade` (default flag): the old
# behaviour — indeterminate check re-runs the installer.
ok, runner, _ = self._install(None, require_confirmed=False)
assert ok is True
runner.assert_called_once()
class TestUpdateCheckTimeoutDefaults:
"""cua_driver_update_check: platform-sensitive default timeout.
8s is fine on POSIX but too tight for Windows first-spawn (Defender /
SmartScreen scanning), and a false timeout is what used to trigger the
full reinstall fall-through during `hermes update`.
"""
def _captured_timeout(self):
from unittest.mock import MagicMock
from tools.computer_use import cua_backend
captured = {}
def fake_run(cmd, **kw):
captured["timeout"] = kw.get("timeout")
m = MagicMock()
m.stdout = '{"update_available": false, "current_version": "1.0"}'
return m
with patch("tools.computer_use.cua_backend.resolve_cua_driver_cmd",
return_value="/x/cua-driver"), \
patch("tools.computer_use.cua_backend.subprocess.run",
side_effect=fake_run):
cua_backend.cua_driver_update_check()
return captured.get("timeout")
@pytest.mark.windows_only
def test_windows_default_is_generous(self):
"""``windows_only``: the 25s default exists because a real Windows
first-spawn is delayed by Defender/SmartScreen scanning — a faked
platform asserted the constant, never the host it is chosen for.
"""
assert self._captured_timeout() == 25.0
def test_posix_default_unchanged(self):
# Unmarked: the POSIX default is what this (Linux) host already picks,
# so no platform faking is involved.
assert self._captured_timeout() == 8.0
def test_explicit_timeout_wins(self):
from unittest.mock import MagicMock
from tools.computer_use import cua_backend
captured = {}
def fake_run(cmd, **kw):
captured["timeout"] = kw.get("timeout")
m = MagicMock()
m.stdout = "{}"
return m
with patch("tools.computer_use.cua_backend.resolve_cua_driver_cmd",
return_value="/x/cua-driver"), \
patch("tools.computer_use.cua_backend.subprocess.run",
side_effect=fake_run):
cua_backend.cua_driver_update_check(timeout=3.0)
assert captured.get("timeout") == 3.0
class TestArchProbeRemoval:
"""Regression tests for the deletion of `_check_cua_driver_asset_for_arch`.
The old probe queried ``/releases/latest`` on trycua/cua and inspected
asset names. That was wrong in two ways:
1. cua-driver-rs releases are marked **prerelease** on every cut, so
``/releases/latest`` returns the Python ``cua-agent`` / ``cua-computer``
package instead — a release with zero binary assets. The probe then
reported "no asset for $arch" on Linux x86_64, Windows, macOS Intel,
Linux arm64 — every non-Apple-Silicon host.
2. Even with the right endpoint, it duplicated tag-resolution the upstream
installer already does correctly via ``CUA_DRIVER_RS_BAKED_VERSION``
(auto-baked by CD on every release).
The fix: stop probing. Trust the upstream installer for fresh installs
(it has the baked version + correct API fallback) and the
``cua-driver check-update --json`` MCP-binary native command for the
upgrade path.
"""
def test_probe_function_is_gone(self):
from hermes_cli import tools_config
assert not hasattr(tools_config, "_check_cua_driver_asset_for_arch")
assert not hasattr(tools_config, "_latest_cua_driver_rs_release")
def test_fresh_install_does_not_call_github_api(self):
"""Pre-install no longer probes the GitHub API — the upstream
``install.sh`` resolves the tag from its baked CUA_DRIVER_RS_BAKED_VERSION
line. install.sh errors cleanly when the arch has no asset, so the
probe was duplicate gatekeeping.
"""
from hermes_cli import tools_config
# No platform fake: "does Python hit the GitHub API?" is host-agnostic,
# and ``which`` is stubbed so the host's own fetch tool resolves.
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/usr/bin/" + n
if n in ("curl", "powershell") else None), \
patch("urllib.request.urlopen") as urlopen, \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner:
assert tools_config.install_cua_driver(upgrade=False) is True
runner.assert_called_once()
urlopen.assert_not_called()
def test_upgrade_with_binary_does_not_call_github_api_directly(self):
"""The upgrade path no longer hits GitHub from Python — it delegates
to the upstream ``install.sh`` (which has the baked release tag and
the proper API fallback). When cua-driver is already installed,
``cua_driver_update_check()`` (added in a separate change) further
short-circuits the network re-install via the binary's native
``check-update --json`` verb.
"""
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/usr/local/bin/" + n
if n in ("cua-driver", "curl", "powershell") else None), \
patch("urllib.request.urlopen") as urlopen, \
patch("subprocess.run"), \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner:
assert tools_config.install_cua_driver(upgrade=True) is True
runner.assert_called_once()
# Probe deleted — no direct GitHub API call from Python.
urlopen.assert_not_called()
@pytest.mark.skipif(
sys.platform == "win32",
reason="POSIX installer uses the .install.lock.d directory protocol",
)
class TestPosixStaleInstallLockClear:
"""_clear_stale_cua_install_lock: pre-clears the upstream installer's
concurrent-install lock only when the holder is provably dead (or the
lock is old and pid-less). Issue #58762."""
def _make_lock(self, tmp_path, pid=None):
import os
home = tmp_path / ".cua-driver"
lock = home / "packages" / ".install.lock.d"
lock.mkdir(parents=True)
if pid is not None:
(lock / "info").write_text(f"pid={pid}\n")
os.environ["CUA_DRIVER_RS_HOME"] = str(home)
return lock
def teardown_method(self):
import os
os.environ.pop("CUA_DRIVER_RS_HOME", None)
def test_dead_holder_lock_is_cleared(self, tmp_path):
from hermes_cli import tools_config
dead_pid = 4194000 # above default pid_max on most systems
lock = self._make_lock(tmp_path, pid=dead_pid)
with patch.object(tools_config, "_print_info"):
tools_config._clear_stale_cua_install_lock()
assert not lock.exists()
def test_live_holder_lock_is_kept(self, tmp_path):
import os
from hermes_cli import tools_config
lock = self._make_lock(tmp_path, pid=os.getpid())
tools_config._clear_stale_cua_install_lock()
assert lock.exists()
def test_pidless_fresh_lock_is_kept(self, tmp_path):
from hermes_cli import tools_config
lock = self._make_lock(tmp_path, pid=None)
tools_config._clear_stale_cua_install_lock()
assert lock.exists()
def test_pidless_old_lock_is_cleared(self, tmp_path):
import os
import time
from hermes_cli import tools_config
lock = self._make_lock(tmp_path, pid=None)
old = time.time() - (tools_config._CUA_LOCK_STALE_AFTER + 60)
os.utime(lock, (old, old))
with patch.object(tools_config, "_print_info"):
tools_config._clear_stale_cua_install_lock()
assert not lock.exists()
def test_no_lock_is_noop(self, tmp_path):
import os
os.environ["CUA_DRIVER_RS_HOME"] = str(tmp_path / ".cua-driver")
from hermes_cli import tools_config
tools_config._clear_stale_cua_install_lock() # must not raise
class TestWindowsStaleInstallLockClearDispatch:
@pytest.mark.windows_only
def test_windows_branch_uses_file_lock_probe(self):
"""``windows_only``: which lock protocol applies IS the host fact under
test — on Linux the faked platform asserted the dispatch and skipped
the ``.install.lock.d`` directory that really exists here.
"""
from hermes_cli import tools_config
with patch.object(
tools_config, "_clear_stale_windows_cua_install_lock"
) as clear_windows:
tools_config._clear_stale_cua_install_lock()
clear_windows.assert_called_once_with()
# ``windows_only`` rather than ``skipif(sys.platform != "win32")``: the
# dedicated Windows CI job selects ``-m windows_only``, so a bare skipif left
# these real-CreateFileW tests running on no host at all.
@pytest.mark.windows_only
class TestWindowsStaleInstallLockClear:
def _make_lock(self, tmp_path):
import os
home = tmp_path / ".cua-driver"
home.mkdir()
lock = home / "install.lock"
lock.write_text("pid=stale\n", encoding="utf-8")
os.environ["CUA_DRIVER_RS_HOME"] = str(home)
return lock
def teardown_method(self):
import os
os.environ.pop("CUA_DRIVER_RS_HOME", None)
def test_unlocked_lock_file_is_cleared(self, tmp_path):
from hermes_cli import tools_config
lock = self._make_lock(tmp_path)
with patch.object(tools_config, "_print_info"):
tools_config._clear_stale_cua_install_lock()
assert not lock.exists()
def test_lock_held_with_file_share_none_is_kept(self, tmp_path):
import ctypes
from ctypes import wintypes
from hermes_cli import tools_config
lock = self._make_lock(tmp_path)
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
create_file = kernel32.CreateFileW
create_file.argtypes = [
wintypes.LPCWSTR,
wintypes.DWORD,
wintypes.DWORD,
wintypes.LPVOID,
wintypes.DWORD,
wintypes.DWORD,
wintypes.HANDLE,
]
create_file.restype = wintypes.HANDLE
close_handle = kernel32.CloseHandle
close_handle.argtypes = [wintypes.HANDLE]
close_handle.restype = wintypes.BOOL
handle = create_file(
str(lock),
0x80000000 | 0x40000000, # GENERIC_READ | GENERIC_WRITE
0, # FileShare::None, matching install.ps1
None,
3, # OPEN_EXISTING
0x00000080, # FILE_ATTRIBUTE_NORMAL
None,
)
assert handle != wintypes.HANDLE(-1).value
try:
tools_config._clear_stale_cua_install_lock()
assert lock.exists()
finally:
assert close_handle(handle)
class TestInstallerTimeoutKillsProcessGroup:
"""On timeout the whole installer process group must be killed, so the
`curl | bash` grandchildren can't survive holding the install lock.
The POSIX cases drop the old ``platform.system`` → "Linux" fake: this lane
IS Linux, so the branch is selected for real. The Windows cases are
``windows_only`` — the psutil tree-kill only runs when ``is_windows``, and
on Linux the fake picked that branch on a host with no such process model.
"""
@pytest.mark.linux_only
def test_timeout_kills_process_group_and_returns_false(self):
import signal
import subprocess
from unittest.mock import MagicMock
from hermes_cli import tools_config
killed = {}
sigkill = getattr(signal, "SIGKILL", signal.SIGTERM)
fake_proc = MagicMock()
fake_proc.pid = 12345
# First communicate() raises TimeoutExpired, second (post-kill) returns.
fake_proc.communicate.side_effect = [
subprocess.TimeoutExpired(cmd="x", timeout=1),
("", None),
]
def fake_killpg(pgid, sig):
killed["pgid"] = pgid
killed["sig"] = sig
with patch("subprocess.run", return_value=MagicMock(returncode=0, stderr="")), \
patch("subprocess.Popen", return_value=fake_proc), \
patch.object(
tools_config.os, "getpgid", return_value=99999, create=True
), \
patch.object(
tools_config.os, "killpg", side_effect=fake_killpg, create=True
), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"):
ok = tools_config._run_cua_driver_installer(label="Refreshing", verbose=False)
assert ok is False
assert killed.get("pgid") == 99999
assert killed.get("sig") == sigkill
# Post-kill reap happened.
assert fake_proc.communicate.call_count == 2
def test_timeout_ceiling_exceeds_upstream_lock_window(self):
from hermes_cli import tools_config
# The upstream installer waits up to 600s before reclaiming a stale
# lock; our ceiling must give that window room to complete.
assert tools_config._CUA_INSTALLER_TIMEOUT > tools_config._CUA_LOCK_STALE_AFTER
@pytest.mark.linux_only
def test_installer_runs_in_new_session_on_posix(self):
from unittest.mock import MagicMock
from hermes_cli import tools_config
captured = {}
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 1
fake_proc.communicate.return_value = ("", None)
def fake_popen(*args, **kwargs):
captured.update(kwargs)
return fake_proc
with patch("subprocess.run", return_value=MagicMock(returncode=0, stderr="")), \
patch("subprocess.Popen", side_effect=fake_popen), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"):
tools_config._run_cua_driver_installer(label="Refreshing", verbose=False)
assert captured.get("start_new_session") is True
@pytest.mark.windows_only
def test_windows_timeout_kills_descendants_and_parent(self):
import subprocess
from unittest.mock import MagicMock
from hermes_cli import tools_config
child = MagicMock()
parent = MagicMock()
parent.children.return_value = [child]
fake_proc = MagicMock()
fake_proc.pid = 12345
fake_proc.communicate.side_effect = [
subprocess.TimeoutExpired(cmd="powershell", timeout=1),
("", None),
]
with patch("subprocess.Popen", return_value=fake_proc), \
patch("psutil.Process", return_value=parent), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"):
ok = tools_config._run_cua_driver_installer(
label="Refreshing", verbose=False
)
assert ok is False
parent.children.assert_called_once_with(recursive=True)
child.kill.assert_called_once_with()
parent.kill.assert_called_once_with()
fake_proc.kill.assert_not_called()
assert fake_proc.communicate.call_count == 2
@pytest.mark.windows_only
def test_windows_tree_enumeration_failure_falls_back_to_direct_kill(self):
import psutil
import subprocess
from unittest.mock import MagicMock
from hermes_cli import tools_config
parent = MagicMock()
parent.children.side_effect = psutil.AccessDenied(pid=12345)
fake_proc = MagicMock()
fake_proc.pid = 12345
fake_proc.communicate.side_effect = [
subprocess.TimeoutExpired(cmd="powershell", timeout=1),
("", None),
]
with patch("subprocess.Popen", return_value=fake_proc), \
patch("psutil.Process", return_value=parent), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"):
ok = tools_config._run_cua_driver_installer(
label="Refreshing", verbose=False
)
assert ok is False
fake_proc.kill.assert_called_once_with()
assert fake_proc.communicate.call_count == 2
@pytest.mark.linux_only
class TestInstallerNoShell:
"""The POSIX installer path must not use shell=True or command
substitution: the script is downloaded to a mkstemp file and exec'd
as a plain argv list (salvage of #34974's intent, without the fixed
/tmp path TOCTOU that PR introduced).
``linux_only``: the download-then-exec argv IS the POSIX branch, and this
lane already takes it — the old ``platform.system`` → "Linux" fake was
asserting a branch the host had already selected.
"""
def _run(self, download_rc=0):
from unittest.mock import MagicMock
from hermes_cli import tools_config
calls = []
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 0
fake_proc.communicate.return_value = ("", None)
def fake_run(cmd, **kw):
calls.append(("run", cmd, kw))
m = MagicMock()
m.returncode = download_rc
m.stderr = "curl: (6) could not resolve" if download_rc else ""
return m
def fake_popen(cmd, **kw):
calls.append(("popen", cmd, kw))
return fake_proc
with patch("subprocess.run", side_effect=fake_run), \
patch("subprocess.Popen", side_effect=fake_popen), \
patch.object(tools_config.shutil, "which", return_value="/usr/local/bin/cua-driver"), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"), \
patch.object(tools_config, "_print_success"):
ok = tools_config._run_cua_driver_installer(label="Refreshing", verbose=False)
return ok, calls
def test_posix_path_downloads_then_execs_argv_list(self):
ok, calls = self._run()
assert ok is True
run_calls = [c for c in calls if c[0] == "run"]
popen_calls = [c for c in calls if c[0] == "popen"]
assert len(run_calls) == 1 and len(popen_calls) == 1
# Download: plain argv curl, no shell.
dl_cmd = run_calls[0][1]
assert isinstance(dl_cmd, list) and dl_cmd[0] == "curl"
# Exec: argv list ["/bin/bash", <mkstemp path>], shell=False.
exec_cmd, exec_kw = popen_calls[0][1], popen_calls[0][2]
assert isinstance(exec_cmd, list) and exec_cmd[0] == "/bin/bash"
assert "cua-driver-install-" in exec_cmd[1]
assert exec_kw.get("shell") is False
def test_download_failure_returns_false_without_exec(self):
ok, calls = self._run(download_rc=6)
assert ok is False
assert not [c for c in calls if c[0] == "popen"]
def test_temp_script_removed_after_run(self):
import os
captured = {}
from unittest.mock import MagicMock
from hermes_cli import tools_config
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 0
fake_proc.communicate.return_value = ("", None)
def fake_run(cmd, **kw):
m = MagicMock(); m.returncode = 0; m.stderr = ""
return m
def fake_popen(cmd, **kw):
captured["script"] = cmd[1]
return fake_proc
with patch("subprocess.run", side_effect=fake_run), \
patch("subprocess.Popen", side_effect=fake_popen), \
patch.object(tools_config.shutil, "which", return_value="/usr/local/bin/cua-driver"), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"), \
patch.object(tools_config, "_print_success"):
tools_config._run_cua_driver_installer(label="Refreshing", verbose=False)
assert "script" in captured
assert not os.path.exists(captured["script"])
class TestConfirmedVersionPinning:
"""When check-update confirms a newer release, the installer run must be
pinned to that exact version via CUA_DRIVER_RS_VERSION.
The upstream installer scripts on `main` carry a baked version that
Release Please bumps in the release PR *before* the release assets are
published. An unpinned install inside that window 404s (observed
2026-07-29: baked 0.14.0 vs latest published release 0.13.1). Pinning to
check-update's `latest_version` — which comes from the Releases API and
therefore has published assets — sidesteps the race.
"""
def _install(self, check_state):
"""Host-agnostic: version pinning is string handling, not an OS branch.
The old ``platform.system`` → "Windows" fake was incidental — the
pin flows into ``CUA_DRIVER_RS_VERSION`` identically on every host
(both upstream installers honour it), and this test never reaches the
installer anyway because ``_run_cua_driver_installer`` is mocked.
"""
from unittest.mock import MagicMock
from hermes_cli import tools_config
with patch.object(tools_config.shutil, "which",
side_effect=lambda n: "/x/" + n
if n in {"cua-driver", "curl", "powershell"} else None), \
patch.object(tools_config, "_resolved_cua_driver_cmd",
return_value="/x/cua-driver"), \
patch.object(tools_config, "_cua_install_target_writable",
return_value=True), \
patch("tools.computer_use.cua_backend.cua_driver_update_check",
return_value=check_state), \
patch.object(tools_config, "_run_cua_driver_installer",
return_value=True) as runner, \
patch("subprocess.run",
return_value=MagicMock(stdout="cua-driver 0.5.0", returncode=0)), \
patch.object(tools_config, "_print_success"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"):
ok = tools_config.install_cua_driver(
upgrade=True, require_confirmed_update=True
)
return ok, runner
def test_confirmed_update_pins_latest_version(self):
state = {"current_version": "0.12.6", "latest_version": "0.13.1",
"update_available": True}
ok, runner = self._install(state)
assert ok is True
assert runner.call_args.kwargs.get("pin_version") == "0.13.1"
def test_v_prefixed_latest_version_is_normalized(self):
state = {"current_version": "0.12.6", "latest_version": "v0.13.1",
"update_available": True}
ok, runner = self._install(state)
assert ok is True
assert runner.call_args.kwargs.get("pin_version") == "0.13.1"
def test_malformed_latest_version_falls_back_unpinned(self):
state = {"current_version": "0.12.6", "latest_version": "not a version",
"update_available": True}
ok, runner = self._install(state)
assert ok is True
assert runner.call_args.kwargs.get("pin_version") is None
def test_missing_latest_version_falls_back_unpinned(self):
state = {"current_version": "0.12.6", "update_available": True}
ok, runner = self._install(state)
assert ok is True
assert runner.call_args.kwargs.get("pin_version") is None
@pytest.mark.linux_only
class TestRunInstallerPinEnv:
"""_run_cua_driver_installer(pin_version=...) exports CUA_DRIVER_RS_VERSION
into the installer child env; unpinned runs leave it untouched.
``linux_only``: the helper reaches Popen through the POSIX
download-then-exec branch, which this lane takes for real — no
``platform.system`` fake needed. The pin itself is host-agnostic
(``TestConfirmedVersionPinning`` covers the caller side unmarked).
"""
def _run(self, pin_version):
from unittest.mock import MagicMock
from hermes_cli import tools_config
captured = {}
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 1
fake_proc.communicate.return_value = ("", None)
def fake_popen(cmd, **kw):
captured["env"] = kw.get("env")
return fake_proc
def fake_run(cmd, **kw):
m = MagicMock(); m.returncode = 0; m.stderr = ""
return m
with patch("subprocess.run", side_effect=fake_run), \
patch("subprocess.Popen", side_effect=fake_popen), \
patch.object(tools_config, "_cua_driver_env",
return_value={"PATH": "/usr/bin"}), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"):
tools_config._run_cua_driver_installer(
label="Refreshing", verbose=False, pin_version=pin_version
)
return captured.get("env") or {}
def test_pin_version_exported_to_installer_env(self):
env = self._run("0.13.1")
assert env.get("CUA_DRIVER_RS_VERSION") == "0.13.1"
def test_no_pin_leaves_env_untouched(self):
env = self._run(None)
assert "CUA_DRIVER_RS_VERSION" not in env
class TestWindowsAutostartRepair:
@pytest.mark.windows_only
def test_existing_task_skips_elevated_powershell_repair(self):
"""``windows_only``: ``_repair_cua_driver_autostart_windows`` returns
True unconditionally off Windows, so only the fake made the schtasks
probe run at all.
"""
from hermes_cli import tools_config
calls = []
def fake_run(cmd, **kwargs):
calls.append((cmd, kwargs))
return SimpleNamespace(returncode=0)
with patch("subprocess.run", side_effect=fake_run), \
patch.object(tools_config.shutil, "which") as which:
ok = tools_config._repair_cua_driver_autostart_windows(
"cua-driver", verbose=False
)
assert ok is True
assert [cmd for cmd, _kwargs in calls] == [
["schtasks.exe", "/Query", "/TN", "cua-driver-serve"]
]
which.assert_not_called()
@pytest.mark.windows_only
def test_windows_installer_runs_autostart_repair_after_success(self):
"""``windows_only``: the PowerShell install argv and the autostart
repair hook are both inside the ``is_windows`` branch, so on Linux the
fake selected a branch whose `powershell` doesn't exist on PATH."""
from unittest.mock import MagicMock
from hermes_cli import tools_config
captured = {}
fake_proc = MagicMock()
fake_proc.pid = 1
fake_proc.returncode = 0
fake_proc.communicate.return_value = ("", None)
def fake_popen(cmd, **kwargs):
captured["cmd"] = cmd
captured["kwargs"] = kwargs
return fake_proc
def fake_which(name: str):
if name == "cua-driver":
return r"C:\Users\Ha Trung\AppData\Local\Programs\Cua\cua-driver\bin\cua-driver.exe"
return None
with patch.object(tools_config.shutil, "which", side_effect=fake_which), \
patch("subprocess.Popen", side_effect=fake_popen), \
patch.object(tools_config, "_clear_stale_cua_install_lock"), \
patch.object(tools_config, "_repair_cua_driver_autostart_windows", return_value=True) as repair, \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"), \
patch.object(tools_config, "_print_success"):
ok = tools_config._run_cua_driver_installer(label="Refreshing", verbose=False)
assert ok is True
assert captured["kwargs"].get("shell") is False
assert isinstance(captured["cmd"], list)
assert captured["cmd"][:4] == [
"powershell", "-NoProfile", "-ExecutionPolicy", "Bypass",
]
repair.assert_called_once_with("cua-driver", verbose=False)
@pytest.mark.windows_only
def test_autostart_repair_quotes_username_space_path_via_file_path(self):
"""``windows_only``: same early return off Windows — the elevated
PowerShell command string is only built on a real Windows host.
"""
from hermes_cli import tools_config
calls = []
driver = (
r"C:\Users\Ha Trung\AppData\Local\Programs\Cua"
r"\cua-driver\bin\cua-driver.exe"
)
def fake_which(name: str):
if name == "cua-driver":
return driver
if name == "powershell":
return r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe"
return None
def fake_run(cmd, **kwargs):
calls.append((cmd, kwargs))
if cmd[0] == "schtasks.exe":
return SimpleNamespace(returncode=1)
return SimpleNamespace(returncode=0, stdout="", stderr="")
with patch.object(tools_config.shutil, "which", side_effect=fake_which), \
patch("subprocess.run", side_effect=fake_run), \
patch.object(tools_config, "_print_warning"), \
patch.object(tools_config, "_print_info"):
ok = tools_config._repair_cua_driver_autostart_windows(
"cua-driver", verbose=False
)
assert ok is True
ps_calls = [cmd for cmd, _kwargs in calls if cmd[0].endswith("powershell.exe")]
assert len(ps_calls) == 1
ps_command = ps_calls[0][-1]
assert "-FilePath $exe" in ps_command
assert "-ArgumentList @('autostart','enable')" in ps_command
assert f"$exe = '{driver}'" in ps_command
assert f"& {driver}" not in ps_command