test: retarget merge-fallout tests at pm-clean's seams; drop tests of retired code

The merges from main kept or added about 45 test files written against the
legacy updater and installer seams that pm-clean replaced. Each failing test
was checked against the current code:

- Deleted, because every test targets code that pm-clean removed or that is
  only reachable from dead or frozen old-updater code:
  update_orphan/handoff_backend_reap, handoff_desktop_rebuild,
  interrupted_recovery (it also launched a real completion against the live
  checkout), update_stale_virtualenv, update_venv_health,
  verify_core_dependencies, lazy_refresh_venv_repair,
  certifi_repair (already dropped in 2f20ceb6f7; the merge resurrected it),
  banner_git_state (covered by test_source_check).
- Retargeted where production reads now:
  - version identity via version_info.get_code_identity
  - update receipts via a ContextVar scope
  - pm.extras / pm.installed_package / pm.ensure_import for matrix,
    computer_use, fal and tirith
  - install hints via pm.install_hint
  - the lock pins in pm/lock.json
  - probe_root for the critical-module probe
- Fixture repairs:
  - git-committed trees for source stamps
  - activate's real `--runtime-only` argument
  - a semver install stamp for requires_hermes gates
  - the Windows gateway restart after a verified update
  - snowflake-length ids that cannot collide with the runner uid
- Collection fix: one platforms() marker per test in gateway_proc_fallback.
This commit is contained in:
ethernet
2026-09-23 16:55:09 -04:00
parent d297db0c70
commit 427d4936c2
35 changed files with 354 additions and 1992 deletions

View File

@@ -1,6 +1,5 @@
"""Tests for agent/skill_utils.py."""
from unittest.mock import patch
import pytest
@@ -304,11 +303,10 @@ class TestParseFrontmatterBOM:
import sys
expected = sys.platform == "darwin"
with patch("agent.skill_utils.is_termux", return_value=False):
plain_fm, _ = parse_frontmatter(self.SKILL)
bom_fm, _ = parse_frontmatter("\ufeff" + self.SKILL)
assert skill_matches_platform(plain_fm) is expected
assert skill_matches_platform(bom_fm) is expected
plain_fm, _ = parse_frontmatter(self.SKILL)
bom_fm, _ = parse_frontmatter("\ufeff" + self.SKILL)
assert skill_matches_platform(plain_fm) is expected
assert skill_matches_platform(bom_fm) is expected

View File

@@ -20,8 +20,9 @@ shape.
from __future__ import annotations
import json
import sys
from io import StringIO
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
@@ -30,6 +31,11 @@ import pytest
# ── helpers ────────────────────────────────────────────────────────────────
def _pm_driver(binary: str):
"""PM reports *binary* as the installed cua-driver (the runtime's selection)."""
return patch("pm.installed_package", return_value=SimpleNamespace(binary=Path(binary)))
def _fake_proc_with_responses(*responses: dict) -> MagicMock:
"""Build a MagicMock subprocess.Popen handle that yields one JSON-RPC
response per `readline()` call, then returns "" (EOF)."""
@@ -106,7 +112,7 @@ class TestDoctorExitCodes:
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch("sys.stdout", new_callable=StringIO):
code = doctor.run_doctor()
@@ -119,7 +125,7 @@ class TestDoctorExitCodes:
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _degraded_report()}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch("sys.stdout", new_callable=StringIO):
code = doctor.run_doctor()
@@ -136,7 +142,7 @@ class TestDoctorExitCodes:
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": report}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch("sys.stdout", new_callable=StringIO):
code = doctor.run_doctor()
@@ -147,7 +153,7 @@ class TestDoctorExitCodes:
`WindowsApps` install): a diagnosis + exit 2, never a raw traceback."""
from tools.computer_use import doctor
with patch("shutil.which", return_value="/protected/cua-driver"), \
with _pm_driver("/protected/cua-driver"), \
patch("subprocess.Popen", side_effect=PermissionError(13, "Access is denied")):
code = doctor.run_doctor()
assert code == 2
@@ -168,7 +174,7 @@ class TestDoctorExitCodes:
proc.wait = MagicMock(return_value=0)
proc.kill = MagicMock()
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc):
code = doctor.run_doctor()
assert code == 2
@@ -190,7 +196,7 @@ class TestResponseShapeParsing:
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "error": {"code": -32601, "message": "method not found"}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc):
code = doctor.run_doctor()
assert code == 2
@@ -208,7 +214,7 @@ class TestArgPassthrough:
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch("sys.stdout", new_callable=StringIO):
doctor.run_doctor(include=["binary_version", "tcc_accessibility"])
@@ -232,7 +238,7 @@ class TestJsonOutput:
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch("sys.stdout", new_callable=StringIO) as out:
doctor.run_doctor(json_output=True)
@@ -250,56 +256,40 @@ class TestJsonOutput:
class TestDriverCmdResolution:
def test_explicit_driver_cmd_arg_wins(self):
@staticmethod
def _executable(path: Path) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("#!/bin/sh\nexit 0\n")
path.chmod(0o755)
return path
def _inspected_binary(self, **kwargs) -> Path:
from tools.computer_use import doctor
proc = _fake_proc_with_responses(
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
with patch("shutil.which", return_value="/fake/explicit-binary") as which_mock, \
patch("subprocess.Popen", return_value=proc), \
with _pm_driver("/pm/store/cua-driver"), \
patch("tools.computer_use.doctor._drive_health_report", return_value=_ok_report()) as health, \
patch("sys.stdout", new_callable=StringIO):
doctor.run_doctor(driver_cmd="/custom/path/cua-driver")
# shutil.which should have been called with the explicit arg, not
# the env-var / default resolver.
which_mock.assert_called_with("/custom/path/cua-driver")
assert doctor.run_doctor(**kwargs) == 0
return Path(health.call_args.args[0])
def test_env_var_used_when_no_arg_given(self, monkeypatch):
from tools.computer_use import doctor
def test_explicit_driver_cmd_arg_wins(self, tmp_path, monkeypatch):
explicit = self._executable(tmp_path / "custom" / "cua-driver")
monkeypatch.setenv("HERMES_CUA_DRIVER_CMD", str(self._executable(tmp_path / "env" / "cua-driver")))
monkeypatch.setenv("HERMES_CUA_DRIVER_CMD", "/env/path/cua-driver")
proc = _fake_proc_with_responses(
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
with patch("shutil.which", return_value="/env/path/cua-driver") as which_mock, \
patch("subprocess.Popen", return_value=proc), \
patch("sys.stdout", new_callable=StringIO), \
patch("hermes_cli.tools_config._cua_driver_cmd", side_effect=Exception("force env")):
# Force env-var resolution path inside run_doctor.
doctor.run_doctor()
which_mock.assert_called_with("/env/path/cua-driver")
assert self._inspected_binary(driver_cmd=str(explicit)) == explicit
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX user-local path regression")
def test_user_local_driver_is_found_when_path_omits_it(self, tmp_path, monkeypatch):
"""Doctor must inspect the same user-local driver as the runtime."""
from tools.computer_use import doctor
def test_env_var_used_when_no_arg_given(self, tmp_path, monkeypatch):
from_env = self._executable(tmp_path / "env" / "cua-driver")
monkeypatch.setenv("HERMES_CUA_DRIVER_CMD", str(from_env))
driver = tmp_path / ".local" / "bin" / "cua-driver"
driver.parent.mkdir(parents=True)
driver.write_text("#!/bin/sh\nexit 0\n")
driver.chmod(0o755)
assert self._inspected_binary() == from_env
def test_doctor_inspects_the_pm_selected_driver_not_path(self, tmp_path, monkeypatch):
"""Doctor must diagnose the driver the runtime invokes: PM's pin, not a PATH copy."""
monkeypatch.delenv("HERMES_CUA_DRIVER_CMD", raising=False)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("PATH", "/usr/bin:/bin:/usr/sbin:/sbin")
monkeypatch.setenv("PATH", str(self._executable(tmp_path / "bin" / "cua-driver").parent))
with patch("tools.computer_use.doctor._drive_health_report", return_value=_ok_report()) as health, \
patch("sys.stdout", new_callable=StringIO):
assert doctor.run_doctor() == 0
assert health.call_args.args[0] == str(driver)
assert self._inspected_binary() == Path("/pm/store/cua-driver")
# ── cua-driver 0.10 unclassified health_report fallback ────────────────────
@@ -375,7 +365,7 @@ class TestDoctorVersionIdentity:
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
# _ok_report claims 0.5.8; CLI says 0.12.6
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch.object(doctor, "_read_cli_version", return_value="cua-driver 0.12.6"), \
patch("sys.stdout", new_callable=StringIO) as out:
@@ -394,7 +384,7 @@ class TestDoctorVersionIdentity:
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
with _pm_driver("/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch.object(doctor, "_read_cli_version", return_value="cua-driver 0.5.8"), \
patch("sys.stdout", new_callable=StringIO) as out:
@@ -418,7 +408,7 @@ def test_failed_tcc_row_from_health_report_names_the_stale_row_reset_for_that_se
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": report}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), patch("subprocess.Popen", return_value=proc), \
with _pm_driver("/fake/cua-driver"), patch("subprocess.Popen", return_value=proc), \
patch("sys.stdout", new_callable=StringIO) as out:
doctor.run_doctor(json_output=True)
checks = {c["name"]: c for c in json.loads(out.getvalue())["checks"]}

View File

@@ -3,7 +3,7 @@
import asyncio
import threading
from datetime import datetime
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
import pytest

View File

@@ -1,7 +1,6 @@
"""Tests for plugins.platforms.feishu.adapter — Feishu scan-to-create registration."""
import json
import sys
from unittest.mock import patch, MagicMock
import pytest
@@ -202,7 +201,8 @@ class TestQrRegister:
def test_qr_fallback_tip_targets_active_interpreter(
self, mock_init, mock_begin, mock_poll, mock_render, capsys
):
"""#111695: the install tip must target the running venv (uv, no pip module)."""
"""#111695: the install tip goes through PM, never a bare pip that targets the wrong env."""
from pm import install_hint
from plugins.platforms.feishu.adapter import _qr_register_inner
mock_begin.return_value = {
@@ -217,7 +217,8 @@ class TestQrRegister:
output = capsys.readouterr().out
assert "https://example.com/qr" in output
assert sys.executable in output
assert install_hint("messaging") in output
assert "pip install" not in output
# -- Contract: expected errors → None, unexpected errors → propagate --

View File

@@ -672,7 +672,7 @@ class TestMatrixRequirements:
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False), \
patch("tools.lazy_deps.feature_missing", return_value=()):
patch("pm.extras.missing", return_value=()):
assert matrix_mod.check_matrix_requirements() is False
def test_check_requirements_e2ee_optional_no_deps_ok(self, monkeypatch):
@@ -684,8 +684,8 @@ class TestMatrixRequirements:
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False), \
patch("tools.lazy_deps.feature_missing", return_value=()), \
patch("tools.lazy_deps.ensure_and_bind", return_value=True):
patch("pm.extras.missing", return_value=()), \
patch("pm.extras.ensure_and_bind", return_value=True):
assert matrix_mod.check_matrix_requirements() is True
def test_check_requirements_encryption_false_no_e2ee_deps_ok(self, monkeypatch):
@@ -696,7 +696,7 @@ class TestMatrixRequirements:
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=False), \
patch("tools.lazy_deps.feature_missing", return_value=()):
patch("pm.extras.missing", return_value=()):
assert matrix_mod.check_matrix_requirements() is True
def test_check_requirements_encryption_true_with_e2ee_deps(self, monkeypatch):
@@ -707,7 +707,7 @@ class TestMatrixRequirements:
import plugins.platforms.matrix.adapter as matrix_mod
with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True), \
patch("tools.lazy_deps.feature_missing", return_value=()):
patch("pm.extras.missing", return_value=()):
assert matrix_mod.check_matrix_requirements() is True
def test_check_e2ee_deps_requires_asyncpg(self, monkeypatch):
@@ -764,17 +764,17 @@ class TestMatrixRequirements:
import plugins.platforms.matrix.adapter as matrix_mod
# Simulate "mautrix installed, asyncpg missing" → feature_missing
# Simulate "mautrix installed, asyncpg missing" → extras.missing
# returns a non-empty tuple → ensure_and_bind MUST be called.
called = {"ensure_and_bind": False}
def _fake_ensure_and_bind(feature, importer, target_globals, **kwargs):
def _fake_ensure_and_bind(extra, importer, target_globals):
called["ensure_and_bind"] = True
assert feature == "platform.matrix"
assert extra == "matrix"
return True # Pretend install succeeded.
with patch("tools.lazy_deps.feature_missing", return_value=("asyncpg==0.31.0",)), \
patch("tools.lazy_deps.ensure_and_bind", side_effect=_fake_ensure_and_bind):
with patch("pm.extras.missing", return_value=("asyncpg",)), \
patch("pm.extras.ensure_and_bind", side_effect=_fake_ensure_and_bind):
matrix_mod.check_matrix_requirements()
assert called["ensure_and_bind"], (

View File

@@ -91,12 +91,13 @@ class TestBuildSessionContextPrompt:
# Force the Discord IDs block on (it only emits when discord tools load).
with patch.object(_gs, "_discord_tools_loaded", return_value=True):
p1 = _prompt_for("1001")
p2 = _prompt_for("2002")
p3 = _prompt_for("3003")
# Snowflake-length ids: short ones like "1001" collide with the
# runner's uid-keyed scratch path that the prompt embeds.
ids = ("1286745390127748101", "1286745390127748202", "1286745390127748303")
p1, p2, p3 = (_prompt_for(i) for i in ids)
assert p1 == p2 == p3, "system prompt must be stable across message_id"
assert "1001" not in p1 and "2002" not in p2 and "3003" not in p3
assert not any(i in p1 for i in ids)

View File

@@ -1,160 +0,0 @@
from unittest.mock import MagicMock, patch
def test_check_via_local_git_ssh_fastpath_ahead_not_behind(tmp_path):
"""SSH fast path must not report an ahead (carried) HEAD as behind.
A carried local commit means tip SHAs differ, but the fresh upstream tip
is an ancestor of HEAD — that is "ahead", and reporting it as behind
nudges the user into `hermes update`, which can wipe the carried work.
"""
from hermes_cli import banner
repo_dir = tmp_path / "repo"
(repo_dir / ".git").mkdir(parents=True)
def fake_git_stdout(args, *, cwd, timeout=5, network=False):
if args == ["remote", "get-url", "origin"]:
return "git@github.com:NousResearch/hermes-agent.git"
if args == ["rev-parse", "HEAD"]:
return "b" * 40 # carried commit, differs from upstream tip
raise AssertionError(f"unexpected git call: {args}")
with (
patch.object(banner, "_git_stdout", side_effect=fake_git_stdout),
patch.object(banner, "_github_branch_tip", return_value="a" * 40),
# merge-base --is-ancestor exits 0: upstream tip IS an ancestor of HEAD
patch.object(banner.subprocess, "run", return_value=MagicMock(returncode=0)),
):
behind = banner._check_via_local_git(repo_dir)
assert behind == 0
def test_check_via_local_git_ssh_fastpath_genuinely_behind(tmp_path):
"""SSH fast path reports the exact count (compare API) when behind."""
from hermes_cli import banner
repo_dir = tmp_path / "repo"
(repo_dir / ".git").mkdir(parents=True)
def fake_git_stdout(args, *, cwd, timeout=5, network=False):
if args == ["remote", "get-url", "origin"]:
return "git@github.com:NousResearch/hermes-agent.git"
if args == ["rev-parse", "HEAD"]:
return "b" * 40
raise AssertionError(f"unexpected git call: {args}")
with (
patch.object(banner, "_git_stdout", side_effect=fake_git_stdout),
patch.object(banner, "_github_branch_tip", return_value="a" * 40),
# merge-base --is-ancestor exits 1: not an ancestor -> genuinely behind
patch.object(banner.subprocess, "run", return_value=MagicMock(returncode=1)),
patch.object(banner, "_github_compare_behind", return_value=3),
):
behind = banner._check_via_local_git(repo_dir)
assert behind == 3
def test_check_via_local_git_ssh_fastpath_offline_keeps_sentinel(tmp_path):
"""Behind + compare API unreachable = honest no-count sentinel, never 1."""
from hermes_cli import banner
repo_dir = tmp_path / "repo"
(repo_dir / ".git").mkdir(parents=True)
def fake_git_stdout(args, *, cwd, timeout=5, network=False):
if args == ["remote", "get-url", "origin"]:
return "git@github.com:NousResearch/hermes-agent.git"
if args == ["rev-parse", "HEAD"]:
return "b" * 40
raise AssertionError(f"unexpected git call: {args}")
with (
patch.object(banner, "_git_stdout", side_effect=fake_git_stdout),
patch.object(banner, "_github_branch_tip", return_value="a" * 40),
patch.object(banner.subprocess, "run", return_value=MagicMock(returncode=1)),
patch.object(banner, "_github_compare_behind", return_value=None),
):
behind = banner._check_via_local_git(repo_dir)
assert behind == banner.UPDATE_AVAILABLE_NO_COUNT
def test_check_via_local_git_insteadof_rewrite_routes_to_ssh_fastpath(tmp_path, monkeypatch):
"""#104591: the origin-URL probe must run under the fetch's config-isolated env.
A global ``url.<https>.insteadOf`` rewrite makes a plain ``git remote get-url origin``
report HTTPS for an SSH origin, so the SSH-avoiding fast path is skipped — while the
fetch itself drops global config (``GIT_CONFIG_GLOBAL=/dev/null``), dials the raw SSH
origin, and its host-key prompt opens /dev/tty and steals the CLI's keystrokes. With the
probe under the same isolated env both sides observe the raw SSH URL and the HTTPS
ls-remote fast path runs instead — no fetch, no ssh child.
"""
import os
import subprocess
from hermes_cli import banner
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
# Config-isolated setup so the developer's own global git config can't leak in.
setup_env = {**os.environ, "GIT_CONFIG_GLOBAL": os.devnull, "GIT_CONFIG_SYSTEM": os.devnull}
setup_cmds = [
["git", "init", "-q"],
# Pinned identity: with global/system config nulled, CI runners whose bare
# hostname makes git's auto-detected ident "user@host.(none)" reject the commit.
["git", "-c", "user.email=t@t", "-c", "user.name=t",
"commit", "--allow-empty", "-q", "-m", "init"],
["git", "remote", "add", "origin", "git@github.com:NousResearch/hermes-agent.git"],
["git", "rev-parse", "HEAD"],
]
head_sha = None
for argv in setup_cmds:
done = subprocess.run(
argv, cwd=repo_dir, env=setup_env, check=True, capture_output=True, text=True)
if argv[1] == "rev-parse":
head_sha = done.stdout.strip()
assert head_sha
# Global config (visible only without GIT_CONFIG_GLOBAL isolation) rewrites SSH to HTTPS.
home = tmp_path / "home"
home.mkdir()
(home / ".gitconfig").write_text(
'[url "https://github.com/"]\n\tinsteadOf = git@github.com:\n', encoding="utf-8")
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home)) # Git for Windows resolves global config here too
calls = []
real_run = banner.subprocess.run
def spy_run(args, **kwargs):
calls.append((list(args), kwargs))
if args[1] in {"ls-remote", "fetch"}:
raise AssertionError(f"a GitHub origin must be probed via the API, not git {args[1]}")
return real_run(args, **kwargs)
monkeypatch.setattr(banner.subprocess, "run", spy_run)
monkeypatch.setattr(banner, "_github_branch_tip", lambda slug, branch: head_sha)
behind = banner._check_via_local_git(repo_dir)
# Same upstream tip as HEAD: the SSH fast path concludes "not behind".
assert behind == 0
assert not any(args[1] == "fetch" for args, _ in calls), (
"insteadOf rewrite must not smuggle the check into the fetch branch")
probe = next(
(kwargs for args, kwargs in calls if args[1:3] == ["remote", "get-url"]), None)
assert probe is not None
assert probe["env"]["GIT_CONFIG_GLOBAL"] == os.devnull, (
"the origin-URL probe must observe the URL the isolated fetch will dial")

View File

@@ -1,183 +0,0 @@
"""Regression tests for issue #29866.
A brew Python upgrade (or an interrupted venv rebuild — see also the v0.19.0
report in the same issue) can leave ``certifi`` importable while its bundled
``cacert.pem`` is missing or a dangling symlink. Every TLS connection then
fails with an opaque ``Could not find a suitable TLS CA certificate bundle``
and the gateway is down on all platforms.
Behavior contracts pinned here:
1. The venv-repair import probes (early recovery + `hermes update`) must
classify certifi as BROKEN when the module imports but ``cacert.pem`` is
missing or corrupt — an attribute probe alone passes in that state.
2. ``hermes doctor`` must fail the certificate check in that state, and
``hermes doctor --fix`` must repair by force-reinstalling certifi and
re-verifying.
"""
import sys
import types
from pathlib import Path
import hermes_cli._early_recovery as er
import subprocess
from hermes_cli import doctor_platform
from hermes_cli import main_install_repair
def _fake_certifi(monkeypatch, bundle_path: Path):
"""Install a fake certifi module whose where() points at bundle_path."""
fake = types.ModuleType("certifi")
fake.contents = lambda: "" # satisfies the ('certifi', 'contents') probe
fake.where = lambda: str(bundle_path)
monkeypatch.setitem(sys.modules, "certifi", fake)
return fake
# =========================================================================
# 1. Import probes detect a missing/corrupt cacert.pem
# =========================================================================
class TestEarlyRecoveryCertifiBundleProbe:
def test_missing_bundle_flags_certifi_broken(self, monkeypatch, tmp_path):
_fake_certifi(monkeypatch, tmp_path / "nonexistent" / "cacert.pem")
broken = er._probe_broken_packages()
assert "certifi" in broken, (
"certifi imports but cacert.pem is missing — the probe must flag "
"it broken (#29866); the attribute check alone passes here"
)
def test_tiny_bundle_flags_certifi_broken(self, monkeypatch, tmp_path):
bundle = tmp_path / "cacert.pem"
bundle.write_text("truncated", encoding="utf-8")
_fake_certifi(monkeypatch, bundle)
broken = er._probe_broken_packages()
assert "certifi" in broken
def test_where_raising_flags_certifi_broken(self, monkeypatch):
fake = types.ModuleType("certifi")
fake.contents = lambda: ""
def _boom():
raise OSError("simulated broken installation")
fake.where = _boom
monkeypatch.setitem(sys.modules, "certifi", fake)
broken = er._probe_broken_packages()
assert "certifi" in broken
class TestUpdateProbeScriptChecksBundle:
"""The subprocess probe used by `hermes update`'s venv repair must apply
the same bundle-file check inside the target venv's interpreter."""
def _run_probe_script(self, monkeypatch, tmp_path, bundle_path):
"""Extract the generated probe script and run it in-process against a
fake certifi that points at bundle_path."""
captured = {}
def fake_run(cmd, **kwargs):
captured["script"] = cmd[-1]
class _R:
returncode = 0
stdout = ""
stderr = ""
return _R()
monkeypatch.setattr(main_install_repair.subprocess, "run", fake_run)
monkeypatch.setattr(
main_install_repair, "_resolve_install_target_python", lambda *a, **k: sys.executable
)
main_install_repair._detect_broken_lazy_refresh_imports(["pip"])
script = captured["script"]
# Execute the probe script with a fake certifi installed.
_fake_certifi(monkeypatch, bundle_path)
printed = []
namespace = {"__builtins__": __builtins__}
import builtins as _b
real_print = _b.print
monkeypatch.setattr(
_b, "print", lambda *a, **k: printed.append(" ".join(map(str, a)))
)
try:
exec(script, namespace)
finally:
monkeypatch.setattr(_b, "print", real_print)
return "\n".join(printed)
def test_probe_script_quiet_when_bundle_healthy(self, monkeypatch, tmp_path):
import certifi as real_certifi
out = self._run_probe_script(
monkeypatch, tmp_path, Path(real_certifi.where())
)
assert "certifi" not in out.splitlines()
# =========================================================================
# 2. hermes doctor: detection and --fix repair
# =========================================================================
class TestDoctorCertificates:
def test_broken_bundle_fails_without_fix(self, monkeypatch, capsys, tmp_path):
monkeypatch.setenv("SSL_CERT_FILE", str(tmp_path / "missing.pem"))
issues = []
doctor_platform.check_certificates(should_fix=False, issues=issues)
out = capsys.readouterr().out
assert "broken" in out.lower()
assert issues, "a broken bundle must be funneled into the action list"
assert any("doctor --fix" in i for i in issues)
def test_fix_reinstalls_certifi_and_reverifies(self, monkeypatch, capsys, tmp_path):
# First verification fails, post-reinstall verification succeeds.
calls = {"verify": 0, "pip": []}
def fake_verify():
calls["verify"] += 1
if calls["verify"] == 1:
from agent.errors import SSLConfigurationError
raise SSLConfigurationError("certifi points to a missing CA bundle")
def fake_run(cmd, **kwargs):
calls["pip"].append(cmd)
class _R:
returncode = 0
stdout = ""
stderr = ""
return _R()
monkeypatch.setattr(
"agent.ssl_guard.verify_ca_bundle", fake_verify
)
monkeypatch.setattr(subprocess, "run", fake_run)
issues = []
doctor_platform.check_certificates(should_fix=True, issues=issues)
out = capsys.readouterr().out
assert calls["pip"], "--fix must run a pip force-reinstall of certifi"
pip_cmd = calls["pip"][0]
assert "--force-reinstall" in pip_cmd and "certifi" in pip_cmd
assert calls["verify"] == 2, "must re-verify after the reinstall"
assert "repaired" in out.lower()
assert not issues
def test_healthy_bundle_never_touches_pip(self, monkeypatch, capsys):
def _fail_run(*a, **k):
raise AssertionError("healthy bundle must not trigger a reinstall")
monkeypatch.setattr(subprocess, "run", _fail_run)
doctor_platform.check_certificates(should_fix=True, issues=[])
out = capsys.readouterr().out
assert "valid" in out.lower()
# =========================================================================
# 3. Startup error message stays actionable
# =========================================================================

View File

@@ -13,8 +13,6 @@ import pytest
import hermes_cli.gateway as gateway_mod
pytestmark = pytest.mark.platforms("linux")
# ---------------------------------------------------------------------------
# Helpers
@@ -55,6 +53,7 @@ def _fake_proc_dir(entries: dict):
# ---------------------------------------------------------------------------
@pytest.mark.platforms("linux")
class TestProcFallback:
"""_scan_gateway_pids reads /proc when available, skips ps.
@@ -238,11 +237,11 @@ class TestGetServicePidsAllProfiles:
]
assert launchctl_calls == [["launchctl", "list"]]
@pytest.mark.platforms("linux")
def test_all_profiles_preserves_systemd_behavior(self):
"""systemd scope is unaffected by the all_profiles switch — it already
lists every hermes-gateway* unit unconditionally."""
with (
patch("hermes_cli.gateway.is_macos", return_value=False),
patch("hermes_cli.gateway.supports_systemd_services", return_value=True),
patch("subprocess.run") as mock_run,
):

View File

@@ -1,203 +0,0 @@
"""Tests for lazy-backend refresh venv repair (#57828 / #58004)."""
from __future__ import annotations
import textwrap
from unittest.mock import MagicMock
import hermes_cli.main as m
import hermes_cli.main_install_repair as hermes_cli_main_install_repair
from hermes_cli import main_install_repair
import pytest
def test_detect_returns_none_when_probe_subprocess_fails(tmp_path, monkeypatch):
python = tmp_path / "python"
python.write_text("", encoding="utf-8")
monkeypatch.setattr(
m, "_resolve_install_target_python", lambda *a, **k: python
)
monkeypatch.setattr(
hermes_cli_main_install_repair, "_resolve_install_target_python", lambda *a, **k: python
)
monkeypatch.setattr(
m.subprocess,
"run",
MagicMock(side_effect=OSError("exec failed")),
)
assert main_install_repair._detect_broken_lazy_refresh_imports(["uv", "pip"]) is None
def test_repair_runs_force_reinstall_with_pyproject_pins(
tmp_path, monkeypatch
):
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(
textwrap.dedent(
"""\
[project]
name = "fake"
version = "0.0.0"
dependencies = [
"pyyaml==6.0.3",
"click==8.2.1",
]
"""
)
)
monkeypatch.setattr(m, "PROJECT_ROOT", tmp_path)
calls: list[list[str]] = []
def fake_install(cmd, **kwargs):
calls.append(cmd)
detect_calls = {"count": 0}
def fake_detect(prefix, *, env=None):
detect_calls["count"] += 1
return []
monkeypatch.setattr(main_install_repair, "_run_package_only_install", fake_install)
monkeypatch.setattr(main_install_repair, "_detect_broken_lazy_refresh_imports", fake_detect)
ok = main_install_repair._repair_broken_lazy_refresh_imports(
["uv", "pip"],
["PyYAML", "click"],
env={"VIRTUAL_ENV": str(tmp_path)},
)
assert ok is True
assert calls == [
[
"uv",
"pip",
"install",
"--force-reinstall",
"pyyaml==6.0.3",
"click==8.2.1",
]
]
assert detect_calls["count"] == 1
def test_refresh_repairs_venv_after_lazy_failure(tmp_path, monkeypatch):
import tools.lazy_deps as lazy_deps_mod
monkeypatch.setattr(lazy_deps_mod, "active_features", lambda: ["platform.matrix"])
monkeypatch.setattr(
lazy_deps_mod,
"refresh_active_features",
lambda **kw: {"platform.matrix": "failed: pip install failed"},
)
repair_calls: list[list[str]] = []
def fake_repair(prefix, packages, *, env=None):
repair_calls.append(packages)
return True
monkeypatch.setattr(main_install_repair, "_detect_broken_lazy_refresh_imports", lambda *a, **k: ["PyYAML"])
monkeypatch.setattr(main_install_repair, "_repair_broken_lazy_refresh_imports", fake_repair)
ok = m._refresh_active_lazy_features(["uv", "pip"], env={"VIRTUAL_ENV": str(tmp_path)})
assert ok is True
assert repair_calls == [["PyYAML"]]
def test_refresh_uses_pre_rebuild_snapshot_when_provided(monkeypatch):
"""Replacement runtimes must not re-detect features after packages vanish."""
import tools.lazy_deps as lazy_deps_mod
monkeypatch.setattr(
lazy_deps_mod,
"active_features",
lambda: pytest.fail("post-rebuild detection must not run"),
)
restored = []
monkeypatch.setattr(
lazy_deps_mod,
"restore_features",
lambda features: restored.append(features) or {"platform.telegram": "restored"},
)
assert m._refresh_active_lazy_features(
["uv", "pip"], features=["platform.telegram"]
) is True
assert restored == [["platform.telegram"]]
def test_restore_active_tool_dependencies_uses_static_allowlist(monkeypatch):
calls = []
monkeypatch.setattr(
m,
"_run_package_only_install",
lambda cmd, *, env=None: calls.append((cmd, env)),
)
monkeypatch.setattr(
hermes_cli_main_install_repair,
"_run_package_only_install",
lambda cmd, *, env=None: calls.append((cmd, env)),
)
env = {"VIRTUAL_ENV": "/tmp/venv"}
m._restore_active_tool_dependencies(
["langfuse", "not-allowlisted"],
["uv", "pip"],
env=env,
)
assert calls == [(["uv", "pip", "install", "langfuse", "--quiet"], env)]
def test_venv_repair_restores_snapshots_with_an_isolated_uv_env(tmp_path, monkeypatch):
"""The unhealthy-venv repair reinstalls the pre-rebuild lazy/tool snapshots through an env
built by ``managed_python_env`` (#83914): a third-party ``UV_PYTHON_INSTALL_DIR`` /
``UV_PYTHON`` / ``VIRTUAL_ENV`` from unrelated software must not steer uv, and
``VIRTUAL_ENV`` points at this install's own venv."""
from hermes_cli import update_cmd
from hermes_constants import venv_python_path
project = tmp_path / "hermes"
venv_python = venv_python_path(project / "venv", windows=m._is_windows())
venv_python.parent.mkdir(parents=True)
venv_python.write_text("", encoding="utf-8")
foreign = tmp_path / "workbuddy"
monkeypatch.setenv("UV_PYTHON_INSTALL_DIR", str(foreign / "python"))
monkeypatch.setenv("UV_PYTHON", str(foreign / "python" / "python.exe"))
monkeypatch.setenv("VIRTUAL_ENV", str(foreign / "venv"))
calls = []
monkeypatch.setattr(m, "PROJECT_ROOT", project)
monkeypatch.setattr(m, "_abort_dependency_sync_if_self_locked", lambda *_a: None)
monkeypatch.setattr(
m, "_install_python_dependencies_with_optional_fallback",
lambda prefix, *, env, group: calls.append(("core", prefix, env, group)))
monkeypatch.setattr(
m, "_refresh_active_lazy_features",
lambda prefix, *, env, features: calls.append(("lazy", prefix, env, features)) or True)
monkeypatch.setattr(
m, "_restore_active_tool_dependencies",
lambda deps, prefix, *, env: calls.append(("tools", prefix, env, deps)))
monkeypatch.setattr(m, "_refresh_active_memory_provider_dependencies", lambda: None)
monkeypatch.setattr(m, "_reapply_plugin_python_dependencies", lambda: None)
monkeypatch.setattr(m, "_clear_update_incomplete_marker", lambda: None)
monkeypatch.setattr(update_cmd, "_write_update_incomplete_marker", lambda: None)
monkeypatch.setattr(update_cmd, "_venv_core_imports_healthy", lambda: (True, "ok"))
monkeypatch.setattr(
update_cmd, "_repair_node_deps_on_current_checkout", lambda *_a, **_k: True)
monkeypatch.setattr("hermes_cli.managed_uv.ensure_uv", lambda **_k: "uv")
assert update_cmd._repair_venv_on_current_checkout(
assume_yes=True, gateway_mode=False, pre_update_snapshot_id=None,
had_desktop_app_before_update=False,
active_lazy_features=["platform.telegram"],
active_tool_dependencies=["langfuse"],
_windows_gateway_resume=None,
) is True
assert [(c[0], c[1], c[3]) for c in calls] == [
("core", ["uv", "pip"], "all"),
("lazy", ["uv", "pip"], ["platform.telegram"]),
("tools", ["uv", "pip"], ["langfuse"]),
]
for _phase, _prefix, env, _extra in calls:
assert env["VIRTUAL_ENV"] == str(project / "venv")
assert str(foreign) not in env.get("UV_PYTHON_INSTALL_DIR", "")
assert env.get("UV_PYTHON") is None
assert env.get("UV_MANAGED_PYTHON") == "1"

View File

@@ -402,17 +402,36 @@ def test_completed_serve_token_is_acknowledged_without_preparation(tmp_path, enc
def test_bootstrap_lock_remains_live_without_application_dependencies(tmp_path):
"""A dependency-free (-I -S) takeover child still sees another live updater's claim.
The holder is a separate, unrelated process: a claim naming the caller's own pid is
adopted as a fresh attempt, so only a foreign live holder can prove liveness detection.
"""
root = Path(__file__).resolve().parents[2]
script = (
lock = tmp_path / "lock"
prelude = (
"import os, sys\nfrom pathlib import Path\n"
f"sys.path.insert(0, {str(root)!r})\n"
"from hermes_cli.update_lock import UpdateLock\n"
f"path = Path({str(tmp_path / 'lock')!r})\n"
"first = UpdateLock(path=path)\nassert first.acquire()\n"
"second = UpdateLock(path=path)\nassert not second.acquire(), 'live lock was stolen'\n"
"assert second.holder.pid == os.getpid()\n"
"first.release()\n"
f"lock = UpdateLock(path=Path({str(lock)!r}))\n"
)
result = subprocess.run([sys.executable, "-I", "-S", "-B", "-c", script],
capture_output=True, text=True, timeout=30)
holder = subprocess.Popen(
[sys.executable, "-I", "-S", "-B", "-c",
prelude + "assert lock.acquire()\nprint(os.getpid(), flush=True)\n"
"sys.stdin.readline()\nlock.release()\n"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, encoding="utf-8",
)
assert holder.stdout is not None
try:
holder_pid = int(holder.stdout.readline())
result = subprocess.run(
[sys.executable, "-I", "-S", "-B", "-c",
prelude + "assert not lock.acquire(), 'live lock was stolen'\n"
f"assert lock.holder.pid == {holder_pid}, lock.holder\n"],
stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding="utf-8", timeout=30,
)
finally:
holder.communicate("\n", timeout=30)
assert result.returncode == 0, result.stdout + result.stderr
assert holder.returncode == 0
assert not lock.exists()

View File

@@ -184,6 +184,11 @@ def test_failed_update_keeps_code_metadata_config_and_environment(installed, mon
old_head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=target, text=True).strip()
state["sha"] = _version(repo, "2.0.0", broken=failure == "dependencies",
minimum=">=999.0.0" if failure == "version" else "")
if failure == "version":
# A checkout without a vX.Y.Z tag (CI's depth-1 clone) runs an unparseable version,
# which the gate deliberately treats as permissive; pin a real one so it can refuse.
from hermes_cli import plugins_manifest
monkeypatch.setattr(plugins_manifest, "running_hermes_version", lambda: "1.0.0")
if failure == "manifest":
(repo / "plugin.yaml").write_text("name: [broken", encoding="utf-8")
state["sha"] = _commit(repo, "invalid manifest")

View File

@@ -1,22 +1,18 @@
"""Serve-kind runtime inventory + stop/relaunch rung (#63206, campaign #91277).
"""Serve-kind runtime inventory (#63206, campaign #91277).
A network-bound `hermes serve --host <ip>` powering a remote Desktop used to
be invisible to the update pipeline: not in the inventory, a dead-end at the
venv-holder guard, and never relaunched after `hermes update` killed it. The
fix threads the spawn ledger's structured launch identity (host/port/profile,
registered at serve startup) through inventory → guard rung → relaunch.
be invisible to the update pipeline. The spawn ledger's structured launch
identity (host/port/profile, registered at serve startup) now feeds the
update inventory and the dashboard process scan.
"""
from __future__ import annotations
import sys
from types import SimpleNamespace
from unittest.mock import patch # noqa: F401 - kept for parity with siblings
from unittest.mock import patch
import hermes_cli.update_cmd as update_cmd
import hermes_cli.update_inventory as update_inventory
from hermes_cli import main as cli_main
import hermes_cli.main_install_repair as main_install_repair
import hermes_cli.main_dashboard as main_dashboard
def _ledger_entry(**over):
@@ -103,74 +99,6 @@ def test_inventory_classifies_desktop_owned_serve(monkeypatch):
assert serves and serves[0].supervisor == "desktop"
assert serves[0].restart_via == "desktop"
# ---------------------------------------------------------------------------
# update_cmd: guard rung helpers
# ---------------------------------------------------------------------------
def test_ledger_manual_serve_holders_filters_correctly(monkeypatch):
manual = _ledger_entry(pid=100)
desktop_owned = _ledger_entry(pid=200, spawner_pid=999, spawner_create=1.0)
gateway = _ledger_entry(pid=300, purpose="gateway")
not_a_holder = _ledger_entry(pid=400)
fake_pi = SimpleNamespace(
ledger_entries=lambda **k: [manual, desktop_owned, gateway, not_a_holder],
spawner_is_dead=lambda e: False if e["pid"] == 200 else None,
)
monkeypatch.setitem(sys.modules, "hermes_cli.process_identity", fake_pi)
holders = [(100, "python.exe", "..."), (200, "python.exe", "..."), (300, "python.exe", "...")]
result = update_cmd._ledger_manual_serve_holders(holders)
pids = [e["pid"] for e in result]
assert pids == [100], (
"only the manual serve holder qualifies: desktop-owned keeps the "
"refusal, gateways belong to the pause machinery, non-holders skipped"
)
def test_serve_relaunch_commands_built_from_structured_identity(monkeypatch):
monkeypatch.setattr(cli_main, "_venv_scripts_dir", lambda: None)
monkeypatch.setattr(main_install_repair, "_venv_scripts_dir", lambda: None)
entries = [
_ledger_entry(), # default profile
_ledger_entry(pid=5000, profile="work", port=9200, host=""),
_ledger_entry(pid=6000, port=None), # no port → skipped
_ledger_entry(pid=7000, purpose="dashboard", host="0.0.0.0", port=9300),
]
cmds = update_cmd._serve_relaunch_commands(entries)
assert ["hermes", "serve", "--host", "100.94.65.93", "--port", "9119"] in cmds
assert ["hermes", "--profile", "work", "serve", "--port", "9200"] in cmds
assert ["hermes", "dashboard", "--host", "0.0.0.0", "--port", "9300"] in cmds
assert len(cmds) == 3 # the port-less entry is skipped
def test_relaunch_stopped_serves_is_idempotent(monkeypatch):
calls = []
monkeypatch.setattr(
cli_main, "_respawn_dashboard_processes", lambda cmds: calls.append(cmds) or []
)
monkeypatch.setattr(
main_dashboard, "_respawn_dashboard_processes", lambda cmds: calls.append(cmds) or []
)
monkeypatch.setattr(cli_main, "_venv_scripts_dir", lambda: None)
monkeypatch.setattr(main_install_repair, "_venv_scripts_dir", lambda: None)
token = {"pending": True, "entries": [_ledger_entry()]}
update_cmd._relaunch_stopped_serves(token)
update_cmd._relaunch_stopped_serves(token) # atexit double-fire
assert len(calls) == 1, "relaunch must fire exactly once"
assert token["pending"] is False
def test_relaunch_stopped_serves_untriggered_token_noop(monkeypatch):
calls = []
monkeypatch.setattr(
cli_main, "_respawn_dashboard_processes", lambda cmds: calls.append(cmds) or []
)
monkeypatch.setattr(
main_dashboard, "_respawn_dashboard_processes", lambda cmds: calls.append(cmds) or []
)
update_cmd._relaunch_stopped_serves({"pending": False, "entries": [_ledger_entry()]})
assert calls == []
# ---------------------------------------------------------------------------
# dashboard_procs: ledger augmentation of the scan (#81564 half)
# ---------------------------------------------------------------------------

View File

@@ -30,15 +30,14 @@ class TestQRCode:
def test_print_qr_code_tip_targets_active_interpreter(self, capsys):
# Regression for #111695: a bare `pip install` targets the wrong
# environment when Hermes runs in an isolated venv (which has no pip
# module at all). The fallback tip must name the interpreter that is
# actually running, via uv.
import sys
# module at all). The fallback tip must route through PM instead.
from pm import install_hint
with patch.dict("sys.modules", {"qrcode": None}):
print_qr_code("https://t.me/newbot/Bot/test_bot")
captured = capsys.readouterr()
assert f"uv pip install --python {sys.executable} qrcode" in captured.out
assert " pip install qrcode)" not in captured.out
assert install_hint("messaging") in captured.out
assert "pip install" not in captured.out
class TestCreatePairing:

View File

@@ -50,6 +50,11 @@ def completion(tmp_path, monkeypatch):
shutil.copytree(ROOT / "scripts/build", source / "scripts/build",
ignore=shutil.ignore_patterns("__pycache__"))
_put(source, "pyproject.toml", '[project]\nname="takeover-fixture"\nversion="2.0"\n')
# Successful completion stamps the selected checkout's own git identity.
git_env = {**os.environ, "GIT_AUTHOR_NAME": "fixture", "GIT_AUTHOR_EMAIL": "fixture@example.invalid",
"GIT_COMMITTER_NAME": "fixture", "GIT_COMMITTER_EMAIL": "fixture@example.invalid"}
for command in (["init", "-q"], ["add", "--all"], ["-c", "commit.gpgsign=false", "commit", "-qm", "selected"]):
subprocess.run(["git", *command], cwd=source, env=git_env, check=True, capture_output=True)
# The selected interpreter is dependency-free, not the pytest environment. A symlink, not a
# copy: a relocatable build (python-build-standalone) locates its stdlib beside the resolved
# executable, so a lone copied binary cannot even import ``encodings``.
@@ -384,6 +389,8 @@ def test_selected_child_builds_and_finalizes_under_parent_lock(completion):
assert receipt["steps"][0] == request["receipt"]["steps"][0]
assert receipt["pm_venv_rebuild"] == request["pm_receipt"]["venv_rebuild"]
assert json.loads(result.read_text()) == {"resume_handled": True, "receipt_handled": True}
head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=source, text=True).strip()
assert json.loads((source / "install-stamp.json").read_text())["commit"] == head
@pytest.mark.platforms("posix")

View File

@@ -1,139 +0,0 @@
"""Tests for the GUI-updater hand-off backend reap (_handoff_reapable_backend_pids).
Field incident (2026-08-20, Teknium's Windows box): a Desktop update hand-off
(`hermes update --yes --gateway --force`) left a *swarm* of per-profile `serve`
backends (mr-tester, probe-inherit, turqoise, clippy, maroon, …) holding
`cryptography\\_rust.pyd`. Some still had a live parent (the tearing-down
Electron process, or the venv launcher→worker two-hop chain mid-exit), so the
strict orphan-only reap (_orphaned_desktop_backend_pids) disqualified the whole
set and the update dead-ended — a 12-minute hang, then a force-close that
stranded bot sessions.
_handoff_reapable_backend_pids is the additional rung that ONLY runs in the
hand-off context (caller gates on args.gateway + the update-incomplete marker +
no live hermes.exe shim). There, any surviving Hermes `serve`/`dashboard`
backend from this venv is a leak — live parent or not — and safe to reap.
A non-backend holder still disqualifies the whole set.
Runs on any host via a fake psutil module (same approach as
test_update_orphan_backend_reap.py).
"""
from __future__ import annotations
import sys
import types
from unittest.mock import MagicMock, patch
from hermes_cli import main as cli_main
class _FakeNoSuchProcess(Exception):
pass
def _fake_psutil(procs: dict[int, MagicMock]):
def _process(pid: int):
if pid not in procs:
raise _FakeNoSuchProcess(pid)
return procs[pid]
return types.SimpleNamespace(Process=_process, NoSuchProcess=_FakeNoSuchProcess)
def _proc(pid: int, cmdline: list[str]):
proc = MagicMock()
proc.pid = pid
proc.cmdline.return_value = cmdline
return proc
def _serve_argv(profile: str = "mr-tester") -> list[str]:
return [
"C:\\hermes\\venv\\Scripts\\python.exe",
"-m",
"hermes_cli.main",
"--profile",
profile,
"serve",
"--host",
"127.0.0.1",
"--port",
"0",
]
def _holder(pid: int, cmdline: str):
return (pid, "python.exe", cmdline)
def test_live_parent_backend_reaped_in_handoff():
# The exact case the orphan-only path REFUSES: a serve backend that still
# has a live parent. In the hand-off context it must still be reaped.
backend = _proc(200, _serve_argv("mr-tester"))
fake = _fake_psutil({200: backend})
with patch.dict(sys.modules, {"psutil": fake}):
holders = [_holder(200, "python.exe -m hermes_cli.main --profile mr-tester serve")]
assert cli_main._handoff_reapable_backend_pids(holders) == [200]
def test_dashboard_backend_reaped():
backend = _proc(200, ["python.exe", "-m", "hermes_cli.main", "dashboard"])
fake = _fake_psutil({200: backend})
with patch.dict(sys.modules, {"psutil": fake}):
holders = [_holder(200, "python.exe -m hermes_cli.main dashboard")]
assert cli_main._handoff_reapable_backend_pids(holders) == [200]
def test_non_backend_holder_disqualifies_whole_set():
# An operator REPL / stray script during a hand-off is unexpected — refuse
# the whole set rather than reap something we can't justify.
backend = _proc(200, _serve_argv("mr-tester"))
repl = _proc(300, ["python.exe", "-m", "hermes_cli.main", "chat"])
fake = _fake_psutil({200: backend, 300: repl})
with patch.dict(sys.modules, {"psutil": fake}):
holders = [
_holder(200, "python.exe -m hermes_cli.main --profile mr-tester serve"),
_holder(300, "python.exe -m hermes_cli.main chat"),
]
assert cli_main._handoff_reapable_backend_pids(holders) is None
def test_exited_holder_skipped_not_fatal():
# A holder that vanished between scan and classification is skipped, and the
# remaining real backend still qualifies.
backend = _proc(200, _serve_argv("mr-tester"))
fake = _fake_psutil({200: backend}) # 300 absent → NoSuchProcess
with patch.dict(sys.modules, {"psutil": fake}):
holders = [
_holder(300, "python.exe -m hermes_cli.main --profile gone serve"),
_holder(200, "python.exe -m hermes_cli.main --profile mr-tester serve"),
]
assert cli_main._handoff_reapable_backend_pids(holders) == [200]
def test_no_holders_returns_none():
fake = _fake_psutil({})
with patch.dict(sys.modules, {"psutil": fake}):
assert cli_main._handoff_reapable_backend_pids([]) is None
def test_psutil_unavailable_returns_none():
# Can't re-read argv to classify → refuse (leave the decision to the
# caller's existing rungs / the dead-end).
import builtins
real_import = builtins.__import__
def _no_psutil(name, *a, **k):
if name == "psutil":
raise ImportError("no psutil")
return real_import(name, *a, **k)
with patch.dict(sys.modules, {}, clear=False):
sys.modules.pop("psutil", None)
with patch("builtins.__import__", _no_psutil):
holders = [_holder(200, "python.exe -m hermes_cli.main --profile x serve")]
assert cli_main._handoff_reapable_backend_pids(holders) is None

View File

@@ -1,90 +0,0 @@
"""The current-checkout repair path must rebuild the Desktop app (#97343).
A Windows git install runs `hermes update` from `hermes.exe`, which reexecs a
venv-Python child to finish the dependency sync. That child completes through
``_repair_node_deps_on_current_checkout`` / the hand-off repair branch, never
through the commits-pulled path that owns the Desktop rebuild — so a
successful-looking update left the packaged desktop app on the previous build.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
from hermes_cli import update_cmd
def test_current_checkout_repair_rebuilds_desktop_under_project_root():
"""The repair passes PROJECT_ROOT/apps/desktop and the pre-update flag."""
completion = MagicMock(return_value=True)
with (
patch.object(update_cmd, "_update_node_dependencies", return_value=[]),
patch.object(update_cmd, "_m") as m,
patch.object(update_cmd, "_check_and_apply_config_migration"),
patch.object(
update_cmd, "_rebuild_desktop_after_update", return_value=True
) as rebuild,
):
m.return_value.PROJECT_ROOT = update_cmd.Path("/fake/hermes")
complete = update_cmd._repair_node_deps_on_current_checkout(
completion, had_desktop_app_before_update=True
)
assert complete is True
rebuild.assert_called_once()
assert rebuild.call_args[0][0] == update_cmd.Path("/fake/hermes/apps/desktop")
assert rebuild.call_args[1]["had_desktop_app_before_update"] is True
completion.assert_called_once()
def test_failed_desktop_rebuild_withholds_success_completion():
"""A failed rebuild must not report success and must return False."""
completion = MagicMock(return_value=True)
with (
patch.object(update_cmd, "_update_node_dependencies", return_value=[]),
patch.object(update_cmd, "_m") as m,
patch.object(update_cmd, "_check_and_apply_config_migration"),
patch.object(update_cmd, "_rebuild_desktop_after_update", return_value=False),
):
m.return_value.PROJECT_ROOT = update_cmd.Path("/fake/hermes")
complete = update_cmd._repair_node_deps_on_current_checkout(completion)
assert complete is False
for call in completion.call_args_list:
assert not call[0][0].startswith("✓")
def test_handoff_venv_repair_finishes_node_and_web_phase(tmp_path):
"""A successful Python repair must not bypass the remaining update work."""
project_root = tmp_path / "hermes"
venv_python = project_root / "venv" / "Scripts" / "python.exe"
venv_python.parent.mkdir(parents=True)
venv_python.touch()
with (
patch.object(update_cmd, "venv_python_path", return_value=venv_python),
patch.object(update_cmd, "_pip_install_prefix", return_value=(["uv", "pip"], None)),
patch.object(update_cmd, "_venv_core_imports_healthy", return_value=(True, "ok")),
patch.object(update_cmd, "_write_update_incomplete_marker"),
patch.object(update_cmd, "_update_node_dependencies", return_value=[]) as update_node,
patch.object(update_cmd, "_check_and_apply_config_migration"),
patch.object(update_cmd, "_rebuild_desktop_after_update", return_value=True),
patch.object(update_cmd, "_print_verified_update_completion", return_value=True) as completion,
patch("hermes_cli.managed_uv.ensure_uv", return_value="uv"),
patch.object(update_cmd, "_m") as m,
):
m.return_value.PROJECT_ROOT = project_root
complete = update_cmd._repair_venv_on_current_checkout(
assume_yes=True,
gateway_mode=False,
pre_update_snapshot_id=None,
had_desktop_app_before_update=False,
active_lazy_features=(),
active_tool_dependencies=(),
_windows_gateway_resume=None,
)
assert complete is True
update_node.assert_called_once_with()
m.return_value._build_web_ui.assert_called_once_with(project_root / "web")
completion.assert_called_once()

View File

@@ -38,9 +38,9 @@ def _write_skewed_tree(root: Path, *, skewed: bool) -> None:
)
(root / "consumer.py").write_text("from provider.thing import SHARED_NAME\n")
def test_syntax_guard_passes_but_import_guard_catches_skew(monkeypatch, tmp_path):
def test_syntax_guard_passes_but_import_guard_catches_skew(monkeypatch, probe_root):
"""The regression: a skewed tree parses cleanly but cannot be imported."""
_write_skewed_tree(tmp_path, skewed=True)
_write_skewed_tree(probe_root, skewed=True)
# Both files are valid Python -- the syntax guard sees nothing wrong.
# NOTE: patch update_cmd's global, not hermes_main's. Both modules expose
@@ -51,44 +51,44 @@ def test_syntax_guard_passes_but_import_guard_catches_skew(monkeypatch, tmp_path
monkeypatch.setattr(
update_cmd, "_UPDATE_CRITICAL_FILES", ("consumer.py", "provider/thing.py")
)
syntax_ok, _, _ = update_cmd._validate_critical_files_syntax(tmp_path)
syntax_ok, _, _ = update_cmd._validate_critical_files_syntax(probe_root)
assert syntax_ok, "sanity: the skewed tree must parse cleanly"
# The import guard catches it.
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = update_cmd._validate_critical_modules_import(tmp_path)
ok, module, error = update_cmd._validate_critical_modules_import(probe_root)
assert ok is False
assert module == "consumer"
assert error is not None and "SHARED_NAME" in error
def test_import_guard_passes_on_consistent_tree(monkeypatch, tmp_path):
_write_skewed_tree(tmp_path, skewed=False)
def test_import_guard_passes_on_consistent_tree(monkeypatch, probe_root):
_write_skewed_tree(probe_root, skewed=False)
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
assert update_cmd._validate_critical_modules_import(tmp_path) == (True, None, None)
assert update_cmd._validate_critical_modules_import(probe_root) == (True, None, None)
def test_import_guard_ignores_non_import_errors(monkeypatch, tmp_path):
def test_import_guard_ignores_non_import_errors(monkeypatch, probe_root):
"""A module that raises at import time for config/env reasons is not
update breakage -- the guard must not roll back a good update."""
(tmp_path / "consumer.py").write_text(
(probe_root / "consumer.py").write_text(
"raise RuntimeError('no API key configured')\n"
)
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, _, _ = update_cmd._validate_critical_modules_import(tmp_path)
ok, _, _ = update_cmd._validate_critical_modules_import(probe_root)
assert ok is True
def test_import_guard_can_report_non_import_errors(monkeypatch, tmp_path):
def test_import_guard_can_report_non_import_errors(monkeypatch, probe_root):
"""Stash restore can compare runtime failures before and after apply."""
(tmp_path / "consumer.py").write_text("raise RuntimeError('broken config')\n")
(probe_root / "consumer.py").write_text("raise RuntimeError('broken config')\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = update_cmd._validate_critical_modules_import(
tmp_path, report_runtime_errors=True
probe_root, report_runtime_errors=True
)
assert ok is False
@@ -96,80 +96,80 @@ def test_import_guard_can_report_non_import_errors(monkeypatch, tmp_path):
assert error == "broken config"
def test_import_guard_can_report_missing_third_party_dependency(
monkeypatch, tmp_path
monkeypatch, probe_root
):
"""Stash comparison must see newly introduced missing dependencies."""
(tmp_path / "consumer.py").write_text("import totally_not_installed_pkg\n")
(probe_root / "consumer.py").write_text("import totally_not_installed_pkg\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = update_cmd._validate_critical_modules_import(
tmp_path, report_runtime_errors=True
probe_root, report_runtime_errors=True
)
assert ok is False
assert module == "consumer"
assert error is not None and "totally_not_installed_pkg" in error
def test_import_failure_comparison_preserves_exception_type(monkeypatch, tmp_path):
source = tmp_path / "consumer.py"
def test_import_failure_comparison_preserves_exception_type(monkeypatch, probe_root):
source = probe_root / "consumer.py"
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
source.write_text("raise RuntimeError('stopped')\n")
runtime_failure = update_cmd._critical_module_import_failures(
tmp_path, report_runtime_errors=True
probe_root, report_runtime_errors=True
)
source.write_text("raise SystemExit('stopped')\n")
terminating_failure = update_cmd._critical_module_import_failures(
tmp_path, report_runtime_errors=True
probe_root, report_runtime_errors=True
)
assert runtime_failure == {"consumer": ("RuntimeError", "stopped")}
assert terminating_failure == {"consumer": ("SystemExit", "stopped")}
def test_import_guard_reports_probe_termination_when_comparing_states(
monkeypatch, tmp_path
monkeypatch, probe_root
):
"""A terminating import is unsafe when validating a restored stash."""
(tmp_path / "consumer.py").write_text("import os\nos._exit(7)\n")
(probe_root / "consumer.py").write_text("import os\nos._exit(7)\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = update_cmd._validate_critical_modules_import(
tmp_path, report_runtime_errors=True
probe_root, report_runtime_errors=True
)
assert ok is False
assert module == "critical-module probe"
assert error and "7" in error
def test_import_guard_reports_probe_termination_by_default(monkeypatch, tmp_path):
def test_import_guard_reports_probe_termination_by_default(monkeypatch, probe_root):
"""A missing health marker must not classify a terminated probe as healthy."""
(tmp_path / "consumer.py").write_text("import os\nos._exit(9)\n")
(probe_root / "consumer.py").write_text("import os\nos._exit(9)\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = update_cmd._validate_critical_modules_import(tmp_path)
ok, module, error = update_cmd._validate_critical_modules_import(probe_root)
assert ok is False
assert module == "critical-module probe"
assert error and "9" in error
def test_import_guard_reports_system_exit_by_default(monkeypatch, tmp_path):
def test_import_guard_reports_system_exit_by_default(monkeypatch, probe_root):
"""Catchable terminating imports must not complete with a healthy marker."""
(tmp_path / "consumer.py").write_text("raise SystemExit('stopped')\n")
(probe_root / "consumer.py").write_text("raise SystemExit('stopped')\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = update_cmd._validate_critical_modules_import(tmp_path)
ok, module, error = update_cmd._validate_critical_modules_import(probe_root)
assert ok is False
assert module == "consumer"
assert error == "stopped"
def test_import_guard_does_not_accept_forged_static_marker(monkeypatch, tmp_path):
def test_import_guard_does_not_accept_forged_static_marker(monkeypatch, probe_root):
"""Imported stdout cannot impersonate the per-probe completion marker."""
(tmp_path / "consumer.py").write_text(
(probe_root / "consumer.py").write_text(
"import os, sys\n"
"sys.stdout.write('__HERMES_IMPORT_HEALTH__[]')\n"
"sys.stdout.flush()\n"
@@ -178,7 +178,7 @@ def test_import_guard_does_not_accept_forged_static_marker(monkeypatch, tmp_path
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = update_cmd._validate_critical_modules_import(tmp_path)
ok, module, error = update_cmd._validate_critical_modules_import(probe_root)
assert ok is False
assert module == "critical-module probe"
@@ -278,21 +278,21 @@ def test_import_guard_uses_the_selected_runtime_command(monkeypatch, tmp_path):
assert seen["command"][:3] == ["/pm/python", "-I", "-c"]
assert seen["cwd"] == str(tmp_path)
def test_import_guard_ignores_missing_third_party_dependency(monkeypatch, tmp_path):
def test_import_guard_ignores_missing_third_party_dependency(monkeypatch, probe_root):
"""A new third-party requirement is not a partially-updated tree.
On the git path this guard runs BEFORE the dependency sync, so a release
that adds a dependency would otherwise look like breakage and trigger a
spurious `git reset --hard` rollback of a perfectly good update.
"""
(tmp_path / "consumer.py").write_text("import totally_not_installed_pkg\n")
(probe_root / "consumer.py").write_text("import totally_not_installed_pkg\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
assert update_cmd._validate_critical_modules_import(tmp_path) == (True, None, None)
assert update_cmd._validate_critical_modules_import(probe_root) == (True, None, None)
def test_import_guard_rejects_module_satisfied_only_by_inherited_pythonpath(
monkeypatch, tmp_path
monkeypatch, probe_root
):
"""A stale checkout on PYTHONPATH must not stand in for a candidate module.
@@ -301,7 +301,7 @@ def test_import_guard_rejects_module_satisfied_only_by_inherited_pythonpath(
fine from there and a candidate lacking it entirely read as healthy
(#115032).
"""
stale = tmp_path / "stale"
stale = probe_root / "stale"
stale.mkdir()
(stale / "hermes_stale_supply.py").write_text("VALUE = 'from the stale tree'\n")
@@ -313,19 +313,19 @@ def test_import_guard_rejects_module_satisfied_only_by_inherited_pythonpath(
)
monkeypatch.setenv("PYTHONPATH", str(stale))
ok, module, error = update_cmd._validate_critical_modules_import(tmp_path)
ok, module, error = update_cmd._validate_critical_modules_import(probe_root)
assert ok is False
assert module == "hermes_stale_supply"
assert error is not None and "hermes_stale_supply" in error
def test_import_guard_accepts_candidate_with_foreign_pythonpath(monkeypatch, tmp_path):
def test_import_guard_accepts_candidate_with_foreign_pythonpath(monkeypatch, probe_root):
"""The env scrub must not overreach: a candidate that carries the module
still passes while a foreign PYTHONPATH is set (#115032)."""
stale = tmp_path / "stale"
stale = probe_root / "stale"
stale.mkdir()
(stale / "hermes_stale_supply.py").write_text("VALUE = 'stale'\n")
(tmp_path / "hermes_stale_supply.py").write_text("VALUE = 'candidate'\n")
(probe_root / "hermes_stale_supply.py").write_text("VALUE = 'candidate'\n")
monkeypatch.setattr(
update_cmd, "_UPDATE_CRITICAL_MODULES", ("hermes_stale_supply",)
@@ -335,17 +335,17 @@ def test_import_guard_accepts_candidate_with_foreign_pythonpath(monkeypatch, tmp
)
monkeypatch.setenv("PYTHONPATH", str(stale))
assert update_cmd._validate_critical_modules_import(tmp_path) == (True, None, None)
assert update_cmd._validate_critical_modules_import(probe_root) == (True, None, None)
def test_import_guard_flags_missing_first_party_module(monkeypatch, tmp_path):
def test_import_guard_flags_missing_first_party_module(monkeypatch, probe_root):
"""A missing *first-party* module IS skew — the update dropped a file."""
(tmp_path / "tools").mkdir()
(tmp_path / "tools" / "__init__.py").write_text("")
(tmp_path / "consumer.py").write_text("import tools.nonexistent_module\n")
(probe_root / "tools").mkdir()
(probe_root / "tools" / "__init__.py").write_text("")
(probe_root / "consumer.py").write_text("import tools.nonexistent_module\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
monkeypatch.setattr(update_cmd_validation, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = update_cmd._validate_critical_modules_import(tmp_path)
ok, module, error = update_cmd._validate_critical_modules_import(probe_root)
assert ok is False
assert module == "consumer"
assert error is not None and "tools.nonexistent_module" in error

View File

@@ -1,25 +0,0 @@
"""Tests for interrupted-install self-heal (the ``.update-incomplete`` marker).
Covers the breadcrumb lifecycle so a ``hermes update`` killed mid-install
(Ctrl-C, terminal close, WSL OOM) leaves a marker the next launch can act on.
"""
from __future__ import annotations
import hermes_cli.main as m
def test_marker_round_trip(tmp_path, monkeypatch):
monkeypatch.setattr(m, "PROJECT_ROOT", tmp_path)
marker = m._update_marker_path()
assert marker == tmp_path / ".update-incomplete"
assert not marker.exists()
m._write_update_incomplete_marker()
assert marker.exists()
body = marker.read_text()
assert "started=" in body
assert "pid=" in body
m._clear_update_incomplete_marker()
assert not marker.exists()

View File

@@ -37,7 +37,7 @@ def fleet(monkeypatch, tmp_path):
monkeypatch.setattr("hermes_cli.gateway.supports_systemd_services", lambda: True)
monkeypatch.setattr("hermes_cli.gateway.find_profile_gateway_processes", lambda exclude_pids=None: [])
monkeypatch.setattr(
"hermes_cli.build_info.get_code_identity",
"hermes_cli.version_info.get_code_identity",
lambda refresh=False: {"sha": "a" * 40, "short_sha": "a" * 8, "version": "1.0", "source": "git"},
)
monkeypatch.setattr("hermes_cli.config.detect_install_method", lambda *a, **k: "git")
@@ -116,7 +116,7 @@ class TestCollectInventory:
for target in (
"hermes_cli.config.detect_install_method",
"hermes_cli.build_info.get_code_identity",
"hermes_cli.version_info.get_code_identity",
"hermes_cli.profiles._get_default_hermes_home",
"hermes_cli.gateway._get_service_pids",
"hermes_cli.gateway.find_profile_gateway_processes",
@@ -143,7 +143,6 @@ class TestReceiptIntegration:
home = tmp_path / "receipt_home"
home.mkdir()
monkeypatch.setattr("hermes_cli.config.get_hermes_home", lambda: home, raising=False)
ur._current = None
ur.begin_update_receipt()
plan = ui.collect_runtime_inventory()
ui.record_plan_in_receipt(plan)

View File

@@ -1,215 +0,0 @@
"""Tests for the orphaned-Desktop-backend reap in the venv-holder guard.
The GUI-updater handoff race (ryanc's 2026-08-09 failures): the Desktop app
fires SIGTERM + app.quit() and spawns hermes-setup, but its Python backend
(``python.exe -m hermes_cli.main serve``) survives the teardown race. The
Desktop is gone — nothing will respawn that backend — yet the venv-holder
guard refused on it and the update dead-ended with "Hermes is still running"
while the user had zero windows open.
``_orphaned_desktop_backend_pids`` classifies holders: a ``serve``/
``dashboard`` backend whose supervising parent is provably dead is safe to
reap (with its full child tree — the managed .hermes-runtime interpreter
child included, #70026); anything else keeps the refusal.
All paths run on any host via a fake psutil module (same approach as
test_update_venv_health.py).
"""
from __future__ import annotations
import sys
import types
from unittest.mock import MagicMock, patch
from hermes_cli import main as cli_main
from hermes_cli import update_cmd
class _FakeNoSuchProcess(Exception):
pass
def _fake_psutil(procs: dict[int, MagicMock]):
"""Build a psutil stand-in whose Process(pid) serves from *procs*."""
def _process(pid: int):
if pid not in procs:
raise _FakeNoSuchProcess(pid)
return procs[pid]
return types.SimpleNamespace(
Process=_process, NoSuchProcess=_FakeNoSuchProcess
)
def _proc(
pid: int,
cmdline: list[str],
*,
ppid: int = 0,
create_time: float = 100.0,
parents: list[MagicMock] | None = None,
):
proc = MagicMock()
proc.pid = pid
proc.cmdline.return_value = cmdline
proc.ppid.return_value = ppid
proc.create_time.return_value = create_time
proc.is_running.return_value = True
proc.parents.return_value = parents or []
return proc
_SERVE_ARGV = [
"C:\\hermes\\venv\\Scripts\\python.exe",
"-m",
"hermes_cli.main",
"serve",
"--host",
"127.0.0.1",
]
def _holders(pid=200, cmdline="python.exe -m hermes_cli.main serve"):
return [(pid, "python.exe", cmdline)]
# ---------------------------------------------------------------------------
# _orphaned_desktop_backend_pids classification
# ---------------------------------------------------------------------------
def test_orphan_backend_dead_parent_qualifies():
backend = _proc(200, _SERVE_ARGV, ppid=999) # 999 not in table → dead
fake = _fake_psutil({200: backend})
with patch.dict(sys.modules, {"psutil": fake}):
assert cli_main._orphaned_desktop_backend_pids(_holders()) == [(200, 10000)]
def test_backend_with_live_parent_keeps_refusal():
parent = _proc(50, ["Hermes.exe"], create_time=10.0)
backend = _proc(200, _SERVE_ARGV, ppid=50, create_time=100.0)
fake = _fake_psutil({50: parent, 200: backend})
with patch.dict(sys.modules, {"psutil": fake}):
assert cli_main._orphaned_desktop_backend_pids(_holders()) is None
def test_recycled_parent_pid_counts_as_orphan():
# "Parent" created AFTER the child = PID reuse; real supervisor is dead.
recycled = _proc(50, ["notepad.exe"], create_time=500.0)
backend = _proc(200, _SERVE_ARGV, ppid=50, create_time=100.0)
fake = _fake_psutil({50: recycled, 200: backend})
with patch.dict(sys.modules, {"psutil": fake}):
assert cli_main._orphaned_desktop_backend_pids(_holders()) == [(200, 10000)]
def test_non_backend_holder_keeps_refusal():
repl = _proc(300, ["python.exe", "-m", "hermes_cli.main", "chat"], ppid=999)
fake = _fake_psutil({300: repl})
with patch.dict(sys.modules, {"psutil": fake}):
holders = _holders(pid=300, cmdline="python.exe -m hermes_cli.main chat")
assert cli_main._orphaned_desktop_backend_pids(holders) is None
def test_mixed_holders_keep_refusal():
# One orphan backend + one operator REPL → the whole set is refused.
backend = _proc(200, _SERVE_ARGV, ppid=999)
repl = _proc(300, ["python.exe", "some_script.py"], ppid=998)
fake = _fake_psutil({200: backend, 300: repl})
with patch.dict(sys.modules, {"psutil": fake}):
holders = _holders() + [(300, "python.exe", "python.exe some_script.py")]
assert cli_main._orphaned_desktop_backend_pids(holders) is None
def test_orphan_root_plus_managed_runtime_descendant_qualifies():
# helix4u's review case (#82179): the scanner returns BOTH the orphaned
# serve root and its .hermes-runtime interpreter child. The child's live
# parent IS the orphan root, so the set is safe — only the root is
# returned (taskkill /T reaps the descendant with it).
backend = _proc(200, _SERVE_ARGV, ppid=999)
child_argv = [
"C:\\hermes\\.hermes-runtime\\python\\generation-1\\python.exe",
"worker.py",
]
child = _proc(210, child_argv, ppid=200, parents=[backend])
fake = _fake_psutil({200: backend, 210: child})
with patch.dict(sys.modules, {"psutil": fake}):
holders = _holders() + [(210, "python.exe", " ".join(child_argv))]
assert cli_main._orphaned_desktop_backend_pids(holders) == [(200, 10000)]
def test_descendant_of_grandchild_depth_qualifies():
# Descendant two hops below the orphan root (root → child → grandchild):
# psutil.parents() walks the full chain, so ancestry still matches.
backend = _proc(200, _SERVE_ARGV, ppid=999)
mid = _proc(210, ["python.exe", "mid.py"], ppid=200, parents=[backend])
grand = _proc(
220, ["python.exe", "leaf.py"], ppid=210, parents=[mid, backend]
)
fake = _fake_psutil({200: backend, 210: mid, 220: grand})
with patch.dict(sys.modules, {"psutil": fake}):
holders = _holders() + [(220, "python.exe", "python.exe leaf.py")]
assert cli_main._orphaned_desktop_backend_pids(holders) == [(200, 10000)]
def test_non_descendant_alongside_orphan_root_keeps_refusal():
# A stray process that is NOT under the orphan root disqualifies the set
# even though an orphan root exists.
backend = _proc(200, _SERVE_ARGV, ppid=999)
unrelated_parent = _proc(50, ["explorer.exe"])
stray = _proc(
300, ["python.exe", "stray.py"], ppid=50, parents=[unrelated_parent]
)
fake = _fake_psutil({50: unrelated_parent, 200: backend, 300: stray})
with patch.dict(sys.modules, {"psutil": fake}):
holders = _holders() + [(300, "python.exe", "python.exe stray.py")]
assert cli_main._orphaned_desktop_backend_pids(holders) is None
def test_descendant_exited_between_scan_and_classify_is_skipped():
backend = _proc(200, _SERVE_ARGV, ppid=999)
fake = _fake_psutil({200: backend}) # descendant 210 already gone
with patch.dict(sys.modules, {"psutil": fake}):
holders = _holders() + [(210, "python.exe", "python.exe worker.py")]
assert cli_main._orphaned_desktop_backend_pids(holders) == [(200, 10000)]
def test_holder_gone_between_scan_and_classify_is_skipped():
fake = _fake_psutil({}) # PID vanished entirely
with patch.dict(sys.modules, {"psutil": fake}):
assert cli_main._orphaned_desktop_backend_pids(_holders()) == []
def test_missing_psutil_keeps_refusal():
import builtins
real_import = builtins.__import__
def _no_psutil(name, *args, **kwargs):
if name == "psutil":
raise ImportError("no psutil")
return real_import(name, *args, **kwargs)
with patch.dict(sys.modules, {"psutil": None}), patch.object(
builtins, "__import__", _no_psutil
):
assert cli_main._orphaned_desktop_backend_pids(_holders()) is None
# ---------------------------------------------------------------------------
# _stop_process_trees
# ---------------------------------------------------------------------------
def test_stop_process_trees_kills_full_tree():
with patch("gateway.status.get_process_start_time", return_value=123), patch(
"hermes_cli._subprocess_compat.pid_is_hermes", return_value=True
), patch.object(update_cmd.subprocess, "run") as run:
cli_main._stop_process_trees([111, 222])
calls = [c.args[0] for c in run.call_args_list]
assert calls == [
["taskkill", "/PID", "111", "/T", "/F"],
["taskkill", "/PID", "222", "/T", "/F"],
]

View File

@@ -231,8 +231,6 @@ def _patch_update_flow(monkeypatch, repo, run_real_git=True):
monkeypatch.setattr(
hermes_main, "_resume_windows_gateways_after_update", lambda *a, **k: None
)
monkeypatch.setattr(hermes_main, "_capture_active_lazy_features", lambda: [])
monkeypatch.setattr(hermes_main, "_capture_active_tool_dependencies", lambda: [])
def test_update_skips_and_warns_on_dirty_parked_branch(
@@ -279,8 +277,8 @@ def test_update_switches_unmerged_parked_branch_with_kept_notice(
pass
monkeypatch.setattr(
hermes_main,
"_abort_dependency_sync_if_self_locked",
update_cmd,
"_complete_source_update",
lambda *a, **k: (_ for _ in ()).throw(_StopFlow()),
)
args = SimpleNamespace(branch=None, yes=False, force=False, force_venv=False)
@@ -330,8 +328,8 @@ def test_update_updates_unmerged_branch_in_place_when_configured(
pass
monkeypatch.setattr(
hermes_main,
"_abort_dependency_sync_if_self_locked",
update_cmd,
"_complete_source_update",
lambda *a, **k: (_ for _ in ()).throw(_StopFlow()),
)
args = SimpleNamespace(branch=None, yes=False, force=False, force_venv=False)
@@ -384,8 +382,8 @@ def test_switch_branch_flag_overrides_in_place_strategy(
pass
monkeypatch.setattr(
hermes_main,
"_abort_dependency_sync_if_self_locked",
update_cmd,
"_complete_source_update",
lambda *a, **k: (_ for _ in ()).throw(_StopFlow()),
)
args = SimpleNamespace(
@@ -424,8 +422,8 @@ def test_update_auto_switches_clean_merged_parked_branch(
pass
monkeypatch.setattr(
hermes_main,
"_abort_dependency_sync_if_self_locked",
update_cmd,
"_complete_source_update",
lambda *a, **k: (_ for _ in ()).throw(_StopFlow()),
)
args = SimpleNamespace(branch=None, yes=False, force=False, force_venv=False)
@@ -472,11 +470,9 @@ def test_update_up_to_date_path_does_not_repark_merged_branch(tmp_path, monkeypa
class _StopFlow(Exception):
pass
import hermes_cli.managed_uv as managed_uv
monkeypatch.setattr(
managed_uv,
"update_managed_uv",
update_cmd,
"_complete_source_update",
lambda *a, **k: (_ for _ in ()).throw(_StopFlow()),
)
args = SimpleNamespace(branch=None, yes=False, force=False, force_venv=False)
@@ -500,8 +496,8 @@ def test_update_on_main_fast_path_unchanged(repo_pair, monkeypatch, capsys):
pass
monkeypatch.setattr(
hermes_main,
"_abort_dependency_sync_if_self_locked",
update_cmd,
"_complete_source_update",
lambda *a, **k: (_ for _ in ()).throw(_StopFlow()),
)
args = SimpleNamespace(branch=None, yes=False, force=False, force_venv=False)

View File

@@ -28,10 +28,9 @@ def receipt_home(tmp_path, monkeypatch):
# ``_receipt_dir`` resolves through ``hermes_constants.get_hermes_home`` (env var), not
# ``hermes_cli.config`` — patch where production reads.
monkeypatch.setenv("HERMES_HOME", str(home))
# ensure no receipt bleeds between tests
ur._current = None
yield home
ur._current = None
# The open receipt is per-context; the scope isolates it from any enclosing receipt.
with ur.update_receipt_scope():
yield home
def _finalize(outcome="success", fleet=None):
@@ -159,7 +158,7 @@ class TestCommandBoundaryFinalization:
assert payload["exit_code"] == 2
assert payload["stop_reason"] == "sys.exit(2)"
assert payload["finished_at"] is not None
assert ur._current is None
assert ur.current_correlation_id() is None
def test_pending_receipt_persisted_on_exit_1_failure(self, receipt_home):
ur.begin_update_receipt()
@@ -239,7 +238,7 @@ class TestCommandBoundaryFinalization:
assert latest["exit_code"] == 2
assert latest["stop_reason"] == "sys.exit(2)"
assert latest["steps"][0]["name"] == "windows_preflight"
assert ur._current is None
assert ur.current_correlation_id() is None
# exactly-once: exactly one receipt file
directory = receipt_home / "logs" / "update_receipts"
assert len(list(directory.glob("update_*.json"))) == 1
@@ -260,7 +259,7 @@ class TestFleetClassification:
json.dumps(gateway_record), encoding="utf-8"
)
monkeypatch.setattr(
"hermes_cli.build_info.get_code_identity",
"hermes_cli.version_info.get_code_identity",
lambda refresh=False: {"sha": expected_sha, "short_sha": expected_sha[:8],
"version": "1.0", "source": "git"},
)
@@ -286,7 +285,7 @@ class TestFleetClassification:
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(
"hermes_cli.build_info.get_code_identity",
"hermes_cli.version_info.get_code_identity",
lambda refresh=False: {"sha": "a" * 40, "short_sha": "a" * 8,
"version": "1.0", "source": "git"},
)
@@ -425,7 +424,7 @@ class TestGatewayStatusStamping:
import gateway.status as gs
monkeypatch.setattr(
"hermes_cli.build_info.get_code_identity",
"hermes_cli.version_info.get_code_identity",
lambda refresh=False: {"sha": "c" * 40, "short_sha": "c" * 8,
"version": "2.0", "source": "git"},
)
@@ -439,7 +438,7 @@ class TestGatewayStatusStamping:
def _boom(refresh=False):
raise RuntimeError("no build info")
monkeypatch.setattr("hermes_cli.build_info.get_code_identity", _boom)
monkeypatch.setattr("hermes_cli.version_info.get_code_identity", _boom)
record = gs._build_runtime_status_record()
# Must not raise, and must not stamp bogus values.
assert "code_sha" not in record

View File

@@ -1,92 +0,0 @@
"""Test: _install_python_dependencies_with_optional_fallback with stale VIRTUAL_ENV.
Simulates the real crash: a pip/system-Python install where PROJECT_ROOT is
site-packages and VIRTUAL_ENV=PROJECT_ROOT/venv does not exist.
"""
import sys
import unittest
from pathlib import Path
from unittest import mock
import hermes_cli.main as main_mod
from hermes_cli import main_install_repair
class StaleVirtualEnvTest(unittest.TestCase):
def _call(self, uv_cmd, venv_path, fake_executable, is_windows=False):
"""Run the function with a mocked uv/env and capture the subprocess call."""
captured = []
def fake_quarantine(cmd, *, env=None, scripts_dir=None, strict_quarantine=False):
captured.append((list(cmd), dict(env or {}), scripts_dir))
return None
def fake_verify(prefix, *, env=None):
return None
with mock.patch.object(main_install_repair, "_run_quarantined_install", fake_quarantine), \
mock.patch.object(main_install_repair, "_verify_console_scripts_installed", fake_verify), \
mock.patch.object(main_install_repair, "_venv_scripts_dir", return_value=None), \
mock.patch.object(main_install_repair, "_is_windows", return_value=is_windows), \
mock.patch.object(main_install_repair.sys, "executable", fake_executable), \
mock.patch.object(main_mod, "PROJECT_ROOT", Path("/fake/project")):
main_install_repair._install_python_dependencies_with_optional_fallback(
list(uv_cmd),
env={"VIRTUAL_ENV": str(venv_path)},
group="all",
)
return captured
def test_stale_virtualenv_pins_python(self):
"""VIRTUAL_ENV points at a nonexistent venv -> --python sys.executable."""
captured = self._call(
uv_cmd=[Path("/fake/uv"), "pip"],
venv_path=Path("/fake/project/venv"), # does not exist
fake_executable="/fake/python311/python.exe",
)
self.assertTrue(captured, "no subprocess call captured")
cmd, env, _ = captured[0]
# --python must come after 'install': uv pip install --python <exe> ...
self.assertIn("install", cmd)
self.assertIn("--python", cmd)
self.assertEqual(cmd[cmd.index("--python") + 1], "/fake/python311/python.exe")
# VIRTUAL_ENV removed from env
self.assertNotIn("VIRTUAL_ENV", env)
def test_existing_virtualenv_keeps_env(self):
"""VIRTUAL_ENV points at an existing venv -> unchanged, no --python."""
real_venv = Path(sys.executable).resolve().parent.parent
if not real_venv.is_dir():
self.skipTest("no real venv available in this test run")
captured = self._call(
uv_cmd=[Path("/fake/uv"), "pip"],
venv_path=real_venv,
fake_executable=sys.executable,
)
cmd, env, _ = captured[0]
self.assertNotIn("--python", cmd)
self.assertEqual(env.get("VIRTUAL_ENV"), str(real_venv))
def test_python_dash_m_uv_is_detected(self):
"""python -m uv must also trigger the pin (naive basename check misses it)."""
captured = self._call(
uv_cmd=[Path("/fake/python"), "-m", "uv", "pip"],
venv_path=Path("/fake/project/venv"), # does not exist
fake_executable="/fake/python311/python.exe",
)
self.assertTrue(captured, "no subprocess call captured")
cmd, _, _ = captured[0]
self.assertIn("--python", cmd)
self.assertEqual(cmd[cmd.index("--python") + 1], "/fake/python311/python.exe")
def test_existing_python_flag_wins(self):
"""A caller-supplied --python is not duplicated by the pin."""
args = ["install", "--python", "/caller/choice/python.exe", "hermes"]
pinned = main_install_repair._insert_python_pin(args)
self.assertEqual(pinned, args, "existing --python must win")
self.assertEqual(pinned.count("--python"), 1)
if __name__ == "__main__":
unittest.main(verbosity=2)

View File

@@ -1,88 +0,0 @@
"""Tests for the Windows half-updated-venv hardening (July 2026 incident).
Covers the commit_count == 0 repair branch: an "Already up to date" checkout
must still re-sync a venv whose installed distribution lags the checkout.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import patch
from hermes_cli import main as cli_main
from hermes_cli import update_cmd
# ---------------------------------------------------------------------------
# _venv_core_imports_healthy
# ---------------------------------------------------------------------------
def _fake_venv_python(tmp_path, *, windows: bool = False):
bin_dir = tmp_path / "venv" / ("Scripts" if windows else "bin")
bin_dir.mkdir(parents=True)
py = bin_dir / ("python.exe" if windows else "python")
py.write_bytes(b"")
return py
# ---------------------------------------------------------------------------
# "Already up to date" must not hide a venv the last pull never re-synced (#97208)
# ---------------------------------------------------------------------------
def test_venv_dependency_set_stale_compares_installed_distribution_with_checkout(tmp_path):
"""Reporter's state: git current at 0.21.2, venv metadata still 0.20.6 (the hand-off child
refused the sync); core imports pass, so only the distribution version reveals the drift."""
from hermes_cli import update_cmd_deps
(tmp_path / "pyproject.toml").write_text('[project]\nname = "hermes-agent"\nversion = "0.21.2"\n', encoding="utf-8")
venv_python = _fake_venv_python(tmp_path)
probes = []
def fake_run(cmd, **kwargs):
probes.append(cmd)
return SimpleNamespace(returncode=0, stdout=f"{fake_run.installed}\n", stderr="")
with patch.object(cli_main, "PROJECT_ROOT", tmp_path), patch.object(cli_main, "_is_windows", return_value=False), \
patch.object(update_cmd_deps.subprocess, "run", fake_run):
fake_run.installed = "0.20.6"
assert update_cmd._venv_dependency_set_stale() == (True, "installed hermes-agent 0.20.6, checkout is 0.21.2")
fake_run.installed = "0.21.2"
assert update_cmd._venv_dependency_set_stale() == (False, "")
# Asked in the venv's own interpreter (the updater may run under another Python).
assert {cmd[0] for cmd in probes} == {str(venv_python)}
def test_current_checkout_with_stale_dependency_set_runs_the_sync(monkeypatch, capsys):
"""Drift on the commit_count == 0 path triggers the same repair as an unhealthy venv instead
of ``✓ Already up to date!``; a synced venv keeps the cheap node-only path."""
from hermes_cli import main as hm
calls = []
monkeypatch.setattr(update_cmd, "_venv_core_imports_healthy", lambda: (True, ""))
monkeypatch.setattr(hm, "_is_windows", lambda: False)
monkeypatch.delenv(hm._UPDATE_REEXEC_ENV, raising=False)
monkeypatch.setattr("hermes_cli.managed_uv.update_managed_uv", lambda **kwargs: None)
monkeypatch.setattr("hermes_cli.managed_uv.ensure_uv", lambda **kwargs: "uv")
monkeypatch.setattr(update_cmd, "_repair_venv_on_current_checkout",
lambda **kwargs: calls.append("sync") or True)
monkeypatch.setattr(update_cmd, "_repair_node_deps_on_current_checkout",
lambda *a, **kwargs: calls.append(kwargs["completion_message"]) or True)
def run(stale):
monkeypatch.setattr(update_cmd, "_venv_dependency_set_stale", lambda: stale)
return update_cmd._repair_current_checkout(
assume_yes=True, gateway_mode=False, pre_update_snapshot_id=None,
had_desktop_app_before_update=False, active_lazy_features=[], active_tool_dependencies=[],
upstream_checked=True, _windows_gateway_resume=None)
assert run((True, "installed hermes-agent 0.20.6, checkout is 0.21.2")) is True
assert calls == ["sync"]
assert run((False, "")) is True
assert calls == ["sync", "✓ Already up to date!"]

View File

@@ -1,117 +0,0 @@
"""Tests for _verify_core_dependencies_installed.
Regression coverage for the partial-install bug where uv's incremental
resolver silently failed to land ``pathspec`` (and similar newly-added
base deps) during ``hermes update``, leaving the venv in a broken state
that only surfaced hours later when a downstream subprocess imported the
missing module.
The verification step:
1. Reads pyproject.toml's [project.dependencies] directly.
2. Filters by environment markers so cross-platform exclusions don't
false-positive (e.g. ``ptyprocess ; sys_platform != 'win32'`` on Windows).
3. Probes ``importlib.metadata.version()`` in the venv interpreter.
4. Reinstalls with --reinstall, then per-package, if anything's missing.
"""
from __future__ import annotations
import sys
import textwrap
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture
def temp_pyproject(tmp_path, monkeypatch):
"""Point hermes_cli.main.PROJECT_ROOT at a tmp dir with a minimal pyproject.
The verification helper opens ``PROJECT_ROOT / 'pyproject.toml'`` directly;
redirecting PROJECT_ROOT keeps the test hermetic.
"""
pyproject = tmp_path / "pyproject.toml"
pyproject.write_text(textwrap.dedent("""\
[project]
name = "fake"
version = "0.0.0"
dependencies = [
"pathspec==1.1.1",
"pydantic==2.13.4",
"ptyprocess>=0.7.0,<1; sys_platform != 'win32'",
"tzdata>=2024.1; sys_platform == 'win32'",
]
"""))
import hermes_cli.main as main_mod
monkeypatch.setattr(main_mod, "PROJECT_ROOT", tmp_path)
return tmp_path
@pytest.fixture
def fake_venv_python(tmp_path):
"""Create a fake venv python shim path that exists on disk."""
venv_root = tmp_path / "venv"
scripts = venv_root / "Scripts"
scripts.mkdir(parents=True)
py = scripts / "python.exe"
py.write_text("#!/bin/sh\necho fake python")
return py, venv_root
class TestVerifyCoreDependencies:
def test_skips_deps_excluded_by_environment_markers(self, temp_pyproject, fake_venv_python):
"""A dep whose ``sys_platform`` marker excludes THIS host must not be
probed (and so never reported missing). Without marker evaluation the
verification step would false-positive on every cross-platform
exclusion and chase its tail installing something inapplicable here.
Deliberately host-invariant rather than ``windows_only``: the subject
is ``packaging``'s marker *evaluation*, not any OS facility. The
pyproject fixture declares one dep gated to non-Windows and one gated
to Windows, so exactly one of the pair is filtered on any host — the
old ``patch("sys.platform", "win32")`` bought nothing but a fake host.
"""
py, venv_root = fake_venv_python
env = {"VIRTUAL_ENV": str(venv_root)}
captured_argv: list[list[str]] = []
def fake_subprocess_run(cmd, **kwargs):
captured_argv.append(list(cmd))
return MagicMock(returncode=0, stdout="", stderr="")
with patch("hermes_cli.main_install_repair._resolve_install_target_python", return_value=py), \
patch("hermes_cli.main_install_repair.subprocess.run", side_effect=fake_subprocess_run), \
patch("hermes_cli.main_install_repair._run_install_with_heartbeat"):
from hermes_cli.main_install_repair import _verify_core_dependencies_installed
_verify_core_dependencies_installed(["uv", "pip"], env=env)
# Find the probe argv — it's the call that passed the dep names.
probe = next(
(argv for argv in captured_argv if any("importlib.metadata" in str(a) for a in argv)),
None,
)
assert probe is not None, "verification probe should have run"
# The dep names are tacked on after the -c script. Exactly one of the
# marker-gated pair applies to this host; the other must be filtered.
on_windows = sys.platform == "win32"
assert ("ptyprocess" in probe) is not on_windows, (
"ptyprocess is gated by sys_platform != 'win32', so it must be "
f"probed off Windows and filtered on it; probe argv was: {probe}"
)
assert ("tzdata" in probe) is on_windows, (
"tzdata is gated by sys_platform == 'win32', so it must be probed "
f"on Windows and filtered elsewhere; probe argv was: {probe}"
)
assert "pathspec" in probe, "core deps without markers must be checked"
def test_no_pyproject_is_noop(self, tmp_path, monkeypatch):
"""If pyproject.toml is missing (unusual but possible in some test
envs), the verification step must short-circuit, not crash."""
import hermes_cli.main as main_mod
monkeypatch.setattr(main_mod, "PROJECT_ROOT", tmp_path)
# No pyproject.toml in tmp_path.
with patch("hermes_cli.main_install_repair._resolve_install_target_python") as mock_resolve, \
patch("hermes_cli.main_install_repair._run_install_with_heartbeat") as mock_install:
from hermes_cli.main_install_repair import _verify_core_dependencies_installed
_verify_core_dependencies_installed(["uv", "pip"], env={})
assert not mock_resolve.called
assert not mock_install.called

View File

@@ -32,7 +32,7 @@ def _sync_checkout(tmp_path: Path):
with (root / "calls.jsonl").open("a", encoding="utf-8") as stream:
stream.write(json.dumps(record) + "\\n")
assert all(value is None for value in record["python_env"].values()), record
assert sys.argv[1:] == ["runtime-only", "--trust-recorded"], record
assert sys.argv[1:] == ["runtime-only"], record
print("setup progress")
if (root / "fail").exists():
sys.exit(42)
@@ -60,10 +60,12 @@ def _sync_checkout(tmp_path: Path):
binary.parent.mkdir(parents=True)
binary.write_text(f"#!/bin/sh\nexec {shlex.quote(str(python))} \"$@\"\n", encoding="utf-8")
binary.chmod(0o755)
# Activation's contract with setup is the runtime-only switch; setup
# itself maps that to PM's --trust-recorded install.
(root / "setup-hermes.sh").write_text(
'test "$#" = 2 && test "$1" = --runtime-only && test "$2" = --trust-recorded || exit 2\n'
'test "$#" = 1 && test "$1" = --runtime-only || exit 2\n'
f'cd {shlex.quote(str(root))} || exit 3\n'
f'exec {shlex.quote(str(python))} sync.py runtime-only --trust-recorded\n', encoding="utf-8",
f'exec {shlex.quote(str(python))} sync.py runtime-only\n', encoding="utf-8",
)
(root / "setup-hermes.ps1").write_text(
"param([switch]$RuntimeOnly)\n"
@@ -73,14 +75,14 @@ def _sync_checkout(tmp_path: Path):
"argv=[Environment]::GetCommandLineArgs()}\n"
"$record | ConvertTo-Json -Compress | Add-Content -LiteralPath \"$PSScriptRoot\\ps-calls.jsonl\"\n"
"Set-Location -LiteralPath $PSScriptRoot\n"
f"& '{python}' sync.py runtime-only --trust-recorded\nexit $LASTEXITCODE\n", encoding="utf-8",
f"& '{python}' sync.py runtime-only\nexit $LASTEXITCODE\n", encoding="utf-8",
)
return root, env
def _assert_syncs(root: Path):
calls = [json.loads(line) for line in (root / "calls.jsonl").read_text(encoding="utf-8").splitlines()]
assert calls == [{"argv": ["runtime-only", "--trust-recorded"], "python_env": {
assert calls == [{"argv": ["runtime-only"], "python_env": {
"PYTHONHOME": None, "PYTHONPATH": None, "VIRTUAL_ENV": None,
}}] * 3
assert (root / "builds").read_text(encoding="utf-8").splitlines() == ["first", "second"]

View File

@@ -133,7 +133,13 @@ def test_selection_refuses_config_edits_during_preparation(client, tmp_path, mon
@pytest.mark.parametrize("invalid", ["name: [", "manifest_version: 999", "requires_hermes: '>=999'", "name: other"])
def test_worker_rejects_unloadable_staged_plugin_without_app_dependencies(client, tmp_path, monkeypatch, invalid):
from pm.store import tree_digest
_current_environment(tmp_path, monkeypatch, [])
repo = _current_environment(tmp_path, monkeypatch, [])
# The install stamp is the running version identity; without a release
# base (a tagless checkout) the requires_hermes gate is permissive.
monkeypatch.delenv("HERMES_INSTALL_ROOT", raising=False)
(repo / "install-stamp.json").write_text(json.dumps({
"commit": "1" * 40, "updateMechanism": "self", "baseVersion": "1.0.0", "source": "local",
}), encoding="utf-8")
target = tmp_path / "home/plugins/example"
target.mkdir(parents=True)
(target / "plugin.yaml").write_text("name: example\n")

View File

@@ -146,23 +146,24 @@ def _assert_forwarded(
inherited_home=inherited_home,
)
assert result.returncode == 0, result.stdout + result.stderr
assert len(calls) == 2, calls
expected_args = ["update", "--yes", "--gateway"]
if windows:
expected_args += ["--force"]
expected_args += expected
assert (
calls
== [
{
"argv": expected_args,
"home": str(home),
"cwd": str(install),
"install_root": str(install),
}
]
* 2
)
expected_argvs = [expected_args] * 2
if windows:
# Desktop stopped the local gateways before handing off; a verified
# update restores the whole fleet in the same home and install.
expected_argvs.append(["gateway", "start", "--all"])
assert calls == [
{
"argv": argv,
"home": str(home),
"cwd": str(install),
"install_root": str(install),
}
for argv in expected_argvs
], calls
receipt = json.loads(
(home / ".hermes-update-result.json").read_text(encoding="utf-8-sig")
)

View File

@@ -38,6 +38,11 @@ _STOCK_NPM_BY_NODE_MAJOR = {
def _root_manifest() -> dict:
return json.loads((REPO_ROOT / "package.json").read_text())
def _pm_lock():
from pm.lock import Lockfile
return Lockfile(REPO_ROOT / "pm" / "lock.json")
def _parse_major_minor_patch(version: str) -> tuple[int, int, int]:
parts = version.split("-", 1)[0].split(".")
nums = [int(p) for p in parts[:3]]
@@ -101,49 +106,26 @@ class TestEnginesAreSatisfiable:
)
def test_node_floor_is_met_by_the_managed_runtime(self):
"""The Node major the installers provision must clear engines.node."""
"""The Node PM provisions must clear engines.node."""
node_range = _root_manifest()["engines"]["node"]
install_sh = (REPO_ROOT / "scripts" / "install.sh").read_text()
for line in install_sh.splitlines():
if line.startswith("NODE_VERSION="):
managed_major = int(line.split("=", 1)[1].strip().strip('"').strip("'"))
break
else: # pragma: no cover - install.sh always defines it
pytest.fail("install.sh does not define NODE_VERSION")
# install.sh fetches latest-v{major}.x, not {major}.0.0. Use a high
# representative release from that major so ranges that enumerate LTS
# lines (rather than one continuous floor) are checked correctly.
managed_release = f"{managed_major}.999.999"
assert _satisfies_range(managed_release, node_range), (
f"engines.node is {node_range!r} but install.sh provisions Node "
f"{managed_major}.x. The runtime we ship must satisfy the floor we "
managed_node = _pm_lock().version("node")
assert managed_node, "pm/lock.json does not pin node"
assert _satisfies_range(managed_node, node_range), (
f"engines.node is {node_range!r} but PM provisions Node "
f"{managed_node}. The runtime we ship must satisfy the floor we "
"declare, or the install we just performed cannot install deps."
)
def test_managed_node_bundles_an_npm_the_engines_accept(self):
"""The Node major install.sh fetches must ship an npm that clears
engines.npm. Node 22 bundles 11.16.0, which is in the excluded
11.10–11.16 band — fresh Hermes-managed installs then die at
`npm ci` with EBADENGINE (#80769).
def test_managed_npm_is_accepted_by_the_engines(self):
"""The npm PM provisions must clear engines.npm, or fresh
Hermes-managed installs die at `npm ci` with EBADENGINE (#80769).
"""
npm_range = _root_manifest()["engines"]["npm"]
install_sh = (REPO_ROOT / "scripts" / "install.sh").read_text()
for line in install_sh.splitlines():
if line.startswith("NODE_VERSION="):
managed_major = int(line.split("=", 1)[1].strip().strip('"').strip("'"))
break
else: # pragma: no cover
pytest.fail("install.sh does not define NODE_VERSION")
stock_npm = _STOCK_NPM_BY_NODE_MAJOR.get(managed_major)
assert stock_npm is not None, (
f"install.sh NODE_VERSION={managed_major} is not in the known "
f"stock map {_STOCK_NPM_BY_NODE_MAJOR}"
)
assert _satisfies_range(stock_npm, npm_range), (
f"install.sh provisions Node {managed_major}.x (stock npm "
f"{stock_npm}), but engines.npm is {npm_range!r}. A fresh "
"Hermes-managed install cannot run npm ci."
managed_npm = _pm_lock().version("npm")
assert managed_npm, "pm/lock.json does not pin npm"
assert _satisfies_range(managed_npm, npm_range), (
f"PM provisions npm {managed_npm}, but engines.npm is "
f"{npm_range!r}. A fresh Hermes-managed install cannot run npm ci."
)
class TestExcludedNpmBand:

View File

@@ -64,11 +64,12 @@ class TestApprovalTimeoutOverflowClamp:
def test_human_wait_ceiling_inherits_clamp(self):
from tools.approval_human_wait import HUMAN_WAIT_MARGIN_S, human_wait_ceiling
from tools.approval_human_wait import human_wait_ceiling
with _with_configured_timeout(10**18):
ceiling = human_wait_ceiling()
assert ceiling == float(int(MAX_SAFE_TIMEOUT_S)) + HUMAN_WAIT_MARGIN_S
# The margin must not push the clamped timeout back past the safe cap.
assert int(MAX_SAFE_TIMEOUT_S) <= ceiling <= MAX_SAFE_TIMEOUT_S
lock = threading.Lock()
assert lock.acquire(timeout=ceiling)
lock.release()

View File

@@ -29,29 +29,32 @@ def fake_fal_client(monkeypatch):
class TestImportFalClient:
def test_lazy_ensure_import_error_is_swallowed(self, monkeypatch, fake_fal_client):
"""If lazy_deps.ensure raises ImportError, it's swallowed (fal_client still imported)."""
def test_pm_ensure_import_error_is_swallowed(self, monkeypatch, fake_fal_client):
"""If pm.ensure_import raises ImportError, the plain import decides."""
monkeypatch.setattr(
"tools.lazy_deps.ensure",
MagicMock(side_effect=ImportError("no lazy_deps")),
"pm.ensure_import",
MagicMock(side_effect=ImportError("not installed")),
)
assert import_fal_client() is fake_fal_client
def test_lazy_ensure_other_exception_raises_import_error(self):
"""If lazy_deps.ensure raises a non-ImportError, it's re-raised as ImportError."""
with patch("tools.lazy_deps.ensure", side_effect=RuntimeError("install hint")):
with pytest.raises(ImportError, match="install hint"):
import_fal_client()
def test_pm_ensure_other_exception_raises_import_error(self, monkeypatch):
"""A non-ImportError from pm (an install hint) surfaces as ImportError."""
monkeypatch.setattr(
"pm.ensure_import",
MagicMock(side_effect=RuntimeError("install hint")),
)
with pytest.raises(ImportError, match="install hint"):
import_fal_client()
def test_lazy_ensure_module_missing_is_swallowed(self, fake_fal_client):
"""If tools.lazy_deps itself can't be imported, ImportError is swallowed."""
def test_pm_missing_is_swallowed(self, fake_fal_client):
"""If pm itself can't be imported, the plain import decides."""
import builtins
original_import = builtins.__import__
def failing_import(name, *args, **kwargs):
if name == "tools.lazy_deps":
if name == "pm":
raise ImportError("no module")
return original_import(name, *args, **kwargs)

View File

@@ -139,17 +139,22 @@ class TestNeedsOcrPath(unittest.TestCase):
self.assertEqual(len(calls), 1) # no hosted attempt
def test_pin_lockstep(self):
"""pyproject core pin and lazy_deps self-heal pin must match."""
import re
"""The doc-extract extra PM uses to self-heal must pin core's anydoc."""
import tomllib
from pathlib import Path
from tools.lazy_deps import LAZY_DEPS
from packaging.requirements import Requirement
py = Path(__file__).resolve().parents[2].joinpath("pyproject.toml").read_text(encoding="utf-8")
m1 = re.search(r'"(firecrawl-anydoc==[\d.]+)"', py)
self.assertIsNotNone(m1)
self.assertEqual(LAZY_DEPS["tool.doc_extract"], (m1.group(1),))
pyproject = Path(__file__).resolve().parents[2] / "pyproject.toml"
project = tomllib.loads(pyproject.read_text(encoding="utf-8"))["project"]
def anydoc_pins(reqs):
return {str(r.specifier) for r in map(Requirement, reqs) if r.name == "firecrawl-anydoc"}
core = anydoc_pins(project["dependencies"])
extra = anydoc_pins(project["optional-dependencies"]["doc-extract"])
self.assertEqual(len(core), 1)
self.assertEqual(extra, core)
if __name__ == "__main__":
unittest.main()

View File

@@ -134,8 +134,8 @@ class TestGetSubprocessHome:
assert home_a is not None
assert home_b is not None
assert home_a != home_b
assert home_a.endswith(os.path.join("alpha", "home"))
assert home_b.endswith(os.path.join("beta", "home"))
assert Path(home_a).parts[-2:] == ("alpha", "home")
assert Path(home_b).parts[-2:] == ("beta", "home")

View File

@@ -1,10 +1,7 @@
"""Tests for the tirith security scanning subprocess wrapper."""
import io
import json
import os
import subprocess
import tarfile
import time
from unittest.mock import MagicMock, patch
@@ -14,25 +11,19 @@ import tools.tirith_security as _tirith_mod
from tools.tirith_security import check_command_security, ensure_installed
def _reset_state():
_tirith_mod._install_attempted.clear()
_tirith_mod._install_threads.clear()
_tirith_mod._crash_count = 0
_tirith_mod._circuit_open = False
_tirith_mod._circuit_open_at = 0.0
@pytest.fixture(autouse=True)
def _reset_resolved_path():
"""Pre-set cached path to skip auto-install in scan tests.
Tests that specifically test ensure_installed / resolve behavior
reset this to None themselves.
"""
_tirith_mod._resolved_path = "tirith"
_tirith_mod._install_thread = None
_tirith_mod._install_failure_reason = ""
_tirith_mod._crash_count = 0
_tirith_mod._circuit_open = False
_tirith_mod._circuit_open_at = 0.0
def _reset_tirith_state():
_reset_state()
yield
_tirith_mod._resolved_path = None
_tirith_mod._install_thread = None
_tirith_mod._install_failure_reason = ""
_tirith_mod._crash_count = 0
_tirith_mod._circuit_open = False
_tirith_mod._circuit_open_at = 0.0
_reset_state()
# ---------------------------------------------------------------------------
@@ -272,7 +263,6 @@ class TestEnsureInstalled:
def test_disabled_returns_none(self, mock_cfg):
mock_cfg.return_value = {"tirith_enabled": False, "tirith_path": "tirith",
"tirith_timeout": 5, "tirith_fail_open": True}
_tirith_mod._resolved_path = None
assert ensure_installed() is None
@patch("tools.tirith_security.shutil.which", return_value="/usr/local/bin/tirith")
@@ -280,12 +270,7 @@ class TestEnsureInstalled:
def test_found_on_path_returns_immediately(self, mock_cfg, mock_which):
mock_cfg.return_value = {"tirith_enabled": True, "tirith_path": "tirith",
"tirith_timeout": 5, "tirith_fail_open": True}
_tirith_mod._resolved_path = None
with patch("os.path.isfile", return_value=True), \
patch("os.access", return_value=True):
result = ensure_installed()
assert result == "/usr/local/bin/tirith"
_tirith_mod._resolved_path = None
assert ensure_installed() == "/usr/local/bin/tirith"
# ---------------------------------------------------------------------------
@@ -293,28 +278,21 @@ class TestEnsureInstalled:
# ---------------------------------------------------------------------------
class TestUnsupportedPlatform:
"""When _detect_target() returns None (no tirith binary for this OS+arch),
the entire subsystem must stay silent: no PATH probes, no download thread,
no disk failure marker, no spawn attempts, no CLI banner. Pattern-matching
"""When PM has no tirith build for this OS+arch, the entire subsystem
must stay silent: no install thread, no spawn attempts, no CLI banner. Pattern-matching
guards still cover the gap; tirith content scanning is just absent."""
@pytest.mark.parametrize("system, machine, expected", [
("Linux", "x86_64", True),
("Windows", "AMD64", False),
("Linux", "riscv64", False),
@pytest.mark.parametrize("target, expected", [
("linux-x64", True),
("win32-x64", False),
(RuntimeError("unsupported architecture: riscv64"), False),
])
def test_is_platform_supported(self, system, machine, expected):
# The patched (system, machine) pairs are table inputs, not a host
# fake: is_platform_supported() is a pure string mapping that touches
# no OS facility beneath the check, so there is nothing for a real
# host to falsify. Two of the rows (Windows/AMD64, Linux/riscv64)
# could never execute honestly anyway — the second has no CI runner
# on any lane.
with patch("tools.tirith_security.platform.system", return_value=system), \
patch("tools.tirith_security.platform.machine", return_value=machine):
def test_is_platform_supported(self, target, expected):
# Table inputs, not a host fake: support is PM's per-target mapping.
current_target = MagicMock(side_effect=[target])
with patch("pm.current_target", current_target):
assert _tirith_mod.is_platform_supported() is expected
@patch("tools.tirith_security._load_security_config")
def test_check_command_security_unsupported_allows_silently(self, mock_cfg):
"""Windows: skip the resolver and spawn entirely — return allow with
@@ -330,212 +308,72 @@ class TestUnsupportedPlatform:
mock_run.assert_not_called()
mock_resolve.assert_not_called()
@patch("tools.tirith_security._load_security_config")
def test_explicit_path_still_honored_on_unsupported_platform(self, mock_cfg):
def test_explicit_path_still_honored_on_unsupported_platform(self, tmp_path):
"""If a user explicitly configured a tirith_path (e.g. they built it
themselves under WSL), the unsupported-platform short-circuit must
NOT override that — explicit config wins."""
mock_cfg.return_value = {"tirith_enabled": True,
"tirith_path": "/opt/custom/tirith",
"tirith_timeout": 5, "tirith_fail_open": True}
_tirith_mod._resolved_path = None
with patch("tools.tirith_security.is_platform_supported", return_value=False), \
patch("os.path.isfile", return_value=True), \
patch("os.access", return_value=True):
result = _tirith_mod._resolve_tirith_path("/opt/custom/tirith")
assert result == "/opt/custom/tirith"
assert _tirith_mod._resolved_path == "/opt/custom/tirith"
custom = tmp_path / "tirith"
custom.write_text("#!/bin/sh\nexit 0\n")
custom.chmod(0o755)
with patch("tools.tirith_security.is_platform_supported", return_value=False):
assert _tirith_mod._resolve_tirith_path(str(custom)) == str(custom)
# ---------------------------------------------------------------------------
# Failed download caches the miss (Finding #1)
# PM-provisioned binary: one install attempt per home, explicit paths never download
# ---------------------------------------------------------------------------
class TestFailedDownloadCaching:
@patch("tools.tirith_security._mark_install_failed")
@patch("tools.tirith_security._is_install_failed_on_disk", return_value=False)
@patch("tools.tirith_security._install_tirith", return_value=(None, "download_failed"))
@patch("tools.tirith_security.shutil.which", return_value=None)
def test_failed_install_cached_no_retry(self, mock_which, mock_install,
mock_disk_check, mock_mark):
"""After a failed download, subsequent resolves must not retry."""
from tools.tirith_security import _resolve_tirith_path, _INSTALL_FAILED
_tirith_mod._resolved_path = None
# First call: tries install, fails
_resolve_tirith_path("tirith")
assert mock_install.call_count == 1
assert _tirith_mod._resolved_path is _INSTALL_FAILED
mock_mark.assert_called_once_with("download_failed") # reason persisted
# Second call: hits the cache, does NOT call _install_tirith again
_resolve_tirith_path("tirith")
assert mock_install.call_count == 1 # still 1, not 2
_tirith_mod._resolved_path = None
_BARE_CFG = {"tirith_enabled": True, "tirith_path": "tirith",
"tirith_timeout": 5, "tirith_fail_open": True}
# ---------------------------------------------------------------------------
# Explicit path must not auto-download (Finding #2)
# ---------------------------------------------------------------------------
@pytest.fixture
def pm_tirith(monkeypatch):
"""Nothing on PATH, nothing installed yet, lazy installs allowed."""
import pm
class TestExplicitPathNoAutoDownload:
@patch("tools.tirith_security._install_tirith")
@patch("tools.tirith_security.shutil.which", return_value=None)
def test_tilde_explicit_path_missing_no_download(self, mock_which, mock_install):
"""An explicit ~/path that doesn't exist must NOT trigger download."""
from tools.tirith_security import _resolve_tirith_path, _INSTALL_FAILED
_tirith_mod._resolved_path = None
installed = MagicMock(return_value=None)
ensure = MagicMock()
monkeypatch.setattr("tools.tirith_security.shutil.which", lambda _name: None)
monkeypatch.setattr(pm, "installed_package", installed)
monkeypatch.setattr(pm, "ensure", ensure)
monkeypatch.setattr(pm, "lazy_installs_allowed", lambda: True)
return ensure, installed
result = _resolve_tirith_path("~/bin/tirith")
mock_install.assert_not_called()
assert _tirith_mod._resolved_path is _INSTALL_FAILED
class TestPmInstall:
def test_default_path_installs_through_pm(self, pm_tirith):
"""The default bare 'tirith' is provisioned by PM on a cold scan."""
ensure, installed = pm_tirith
ensure.side_effect = lambda *_a, **_k: setattr(
installed, "return_value", MagicMock(binary="/pm/tirith"))
assert _tirith_mod._resolve_tirith_path("tirith") == "/pm/tirith"
ensure.assert_called_once_with("tirith")
def test_failed_install_is_not_retried(self, pm_tirith):
"""After a failed install, subsequent resolves fall back without retrying."""
ensure, _ = pm_tirith
ensure.side_effect = RuntimeError("download failed")
assert _tirith_mod._resolve_tirith_path("tirith") == "tirith"
assert _tirith_mod._resolve_tirith_path("tirith") == "tirith"
assert ensure.call_count == 1
def test_tilde_explicit_path_missing_no_download(self, pm_tirith):
"""An explicit ~/path that doesn't exist must NOT trigger an install."""
ensure, _ = pm_tirith
result = _tirith_mod._resolve_tirith_path("~/bin/tirith")
ensure.assert_not_called()
assert "~" not in result # tilde still expanded
_tirith_mod._resolved_path = None
@patch("tools.tirith_security._mark_install_failed")
@patch("tools.tirith_security._is_install_failed_on_disk", return_value=False)
@patch("tools.tirith_security._install_tirith", return_value=("/auto/tirith", ""))
@patch("tools.tirith_security.shutil.which", return_value=None)
def test_default_path_does_auto_download(self, mock_which, mock_install,
mock_disk_check, mock_mark):
"""The default bare 'tirith' SHOULD trigger auto-download."""
from tools.tirith_security import _resolve_tirith_path
_tirith_mod._resolved_path = None
result = _resolve_tirith_path("tirith")
mock_install.assert_called_once()
assert result == "/auto/tirith"
_tirith_mod._resolved_path = None
# ---------------------------------------------------------------------------
# Cosign provenance verification (P1)
# ---------------------------------------------------------------------------
class TestCosignVerification:
@patch("tools.tirith_security.subprocess.run")
@patch("tools.tirith_security.shutil.which", return_value="/usr/bin/cosign")
def test_cosign_identity_pinned_to_release_workflow(self, mock_which, mock_run):
"""Identity regexp must pin to the release workflow, not the whole repo."""
from tools.tirith_security import _verify_cosign
mock_run.return_value = _mock_run(0, "Verified OK")
_verify_cosign("/tmp/checksums.txt", "/tmp/sig", "/tmp/cert")
args = mock_run.call_args[0][0]
# Find the value after --certificate-identity-regexp
idx = args.index("--certificate-identity-regexp")
identity = args[idx + 1]
# The identity contains regex-escaped dots
assert "workflows/release" in identity
assert "refs/tags/v" in identity
@patch("tools.tirith_security.tarfile.open")
@patch("tools.tirith_security._verify_checksum", return_value=True)
@patch("tools.tirith_security.shutil.which", return_value=None)
@patch("tools.tirith_security._download_file")
@patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin")
def test_install_proceeds_without_cosign(self, mock_target, mock_dl,
mock_which, mock_checksum,
mock_tarfile):
"""_install_tirith proceeds with SHA-256 only when cosign is not on PATH."""
from tools.tirith_security import _install_tirith
mock_tar = MagicMock()
mock_tar.__enter__ = MagicMock(return_value=mock_tar)
mock_tar.__exit__ = MagicMock(return_value=False)
mock_tar.getmembers.return_value = []
mock_tarfile.return_value = mock_tar
path, reason = _install_tirith()
# Reaches extraction (no binary in mock archive), but got past cosign
assert path is None
assert reason == "binary_not_in_archive"
assert mock_checksum.called # SHA-256 verification ran
class TestInstallArchiveMemberValidation:
def _write_archive(self, tmp_path, member: tarfile.TarInfo, data: bytes | None = None):
archive = tmp_path / "tirith-aarch64-apple-darwin.tar.gz"
checksums = tmp_path / "checksums.txt"
with tarfile.open(archive, "w:gz") as tar:
if data is None:
tar.addfile(member)
else:
tar.addfile(member, io.BytesIO(data))
checksums.write_text(
"ignored tirith-aarch64-apple-darwin.tar.gz\n",
encoding="utf-8",
)
return archive, checksums
def _download_side_effect(self, archive, checksums):
def _download(url, dest, timeout=10):
del timeout
if url.endswith(".tar.gz"):
with open(archive, "rb") as src, open(dest, "wb") as dst:
dst.write(src.read())
return
if url.endswith("checksums.txt"):
with open(checksums, "rb") as src, open(dest, "wb") as dst:
dst.write(src.read())
return
raise AssertionError(f"unexpected download URL: {url}")
return _download
@patch("tools.tirith_security._verify_checksum", return_value=True)
@patch("tools.tirith_security.shutil.which", return_value=None)
@patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin")
def test_install_extracts_regular_tirith_member(self, mock_target, mock_which,
mock_checksum, tmp_path, monkeypatch):
"""A valid regular-file tirith member is installed as a plain file."""
del mock_target, mock_which, mock_checksum
from tools.tirith_security import _install_tirith
payload = b"#!/bin/sh\nexit 0\n"
member = tarfile.TarInfo("bin/tirith")
member.mode = 0o755
member.size = len(payload)
archive, checksums = self._write_archive(tmp_path, member, payload)
hermes_home = tmp_path / "hermes-home"
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
with patch("tools.tirith_security._download_file",
side_effect=self._download_side_effect(archive, checksums)):
path, reason = _install_tirith(log_failures=False)
assert reason == ""
assert path == str(hermes_home / "bin" / "tirith")
assert os.path.isfile(path)
assert not os.path.islink(path)
with open(path, "rb") as f:
assert f.read() == payload
@patch("tools.tirith_security._verify_checksum", return_value=True)
@patch("tools.tirith_security.shutil.which", return_value=None)
@patch("tools.tirith_security._detect_target", return_value="aarch64-apple-darwin")
def test_install_rejects_non_regular_tirith_member(self, mock_target, mock_which,
mock_checksum, tmp_path, monkeypatch):
"""Symlink or hardlink tar members must not be installed as tirith."""
del mock_target, mock_which, mock_checksum
from tools.tirith_security import _install_tirith
member = tarfile.TarInfo("bin/tirith")
member.type = tarfile.SYMTYPE
member.linkname = "/bin/sh"
archive, checksums = self._write_archive(tmp_path, member)
hermes_home = tmp_path / "hermes-home"
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
with patch("tools.tirith_security._download_file",
side_effect=self._download_side_effect(archive, checksums)):
path, reason = _install_tirith(log_failures=False)
assert path is None
assert reason == "binary_not_regular_file"
assert not os.path.lexists(hermes_home / "bin" / "tirith")
def test_install_proceeds_without_cosign(self, tmp_path):
"""Provenance is optional without cosign: SHA-256 verification alone proceeds."""
with patch("tools.tirith_security.shutil.which", return_value=None):
verified, reason = _tirith_mod.verify_release_provenance(tmp_path, MagicMock())
assert (verified, reason) == (False, "")
# ---------------------------------------------------------------------------
@@ -543,96 +381,25 @@ class TestInstallArchiveMemberValidation:
# ---------------------------------------------------------------------------
class TestBackgroundInstall:
def test_ensure_installed_non_blocking(self):
"""ensure_installed must return immediately when download needed."""
_tirith_mod._resolved_path = None
with patch("tools.tirith_security._load_security_config",
return_value={"tirith_enabled": True, "tirith_path": "tirith",
"tirith_timeout": 5, "tirith_fail_open": True}), \
patch("tools.tirith_security.shutil.which", return_value=None), \
patch("tools.tirith_security._hermes_bin_dir", return_value="/nonexistent"), \
patch("tools.tirith_security._is_install_failed_on_disk", return_value=False), \
def test_ensure_installed_non_blocking(self, pm_tirith):
"""ensure_installed must return immediately when an install is needed."""
with patch("tools.tirith_security._load_security_config", return_value=_BARE_CFG), \
patch("tools.tirith_security.is_platform_supported", return_value=True), \
patch("tools.tirith_security.threading.Thread") as MockThread:
mock_thread = MagicMock()
mock_thread.is_alive.return_value = False
MockThread.return_value = mock_thread
result = ensure_installed()
assert result is None # not available yet
assert ensure_installed() is None # not available yet
MockThread.assert_called_once()
mock_thread.start.assert_called_once()
MockThread.return_value.start.assert_called_once()
_tirith_mod._resolved_path = None
def test_scan_does_not_wait_on_startup_install(self, pm_tirith):
"""A scan during the startup install returns the default instead of installing again."""
ensure, _ = pm_tirith
with patch("tools.tirith_security._load_security_config", return_value=_BARE_CFG), \
patch("tools.tirith_security.is_platform_supported", return_value=True), \
patch("tools.tirith_security.threading.Thread"):
ensure_installed()
def test_resolve_returns_default_when_thread_alive(self):
"""_resolve_tirith_path returns default while background thread runs."""
from tools.tirith_security import _resolve_tirith_path
_tirith_mod._resolved_path = None
mock_thread = MagicMock()
mock_thread.is_alive.return_value = True
_tirith_mod._install_thread = mock_thread
with patch("tools.tirith_security.shutil.which", return_value=None), \
patch("tools.tirith_security._hermes_bin_dir", return_value="/nonexistent"):
result = _resolve_tirith_path("tirith")
assert result == "tirith" # returns configured default, doesn't block
_tirith_mod._install_thread = None
_tirith_mod._resolved_path = None
# ---------------------------------------------------------------------------
# Disk failure marker persistence (P2)
# ---------------------------------------------------------------------------
class TestDiskFailureMarker:
def test_expired_marker_ignored(self):
"""Marker older than TTL should be ignored."""
import tempfile
tmpdir = tempfile.mkdtemp()
marker = os.path.join(tmpdir, ".tirith-install-failed")
with patch("tools.tirith_security._failure_marker_path", return_value=marker):
from tools.tirith_security import _mark_install_failed, _is_install_failed_on_disk
assert not _is_install_failed_on_disk()
_mark_install_failed("download_failed")
assert _is_install_failed_on_disk()
# Backdate the file past 24h TTL
old_time = time.time() - 90000 # 25 hours ago
os.utime(marker, (old_time, old_time))
assert not _is_install_failed_on_disk()
def test_in_memory_cosign_exec_failed_not_retried(self):
"""In-memory _INSTALL_FAILED with cosign_exec_failed is NOT retried."""
from tools.tirith_security import _resolve_tirith_path, _INSTALL_FAILED
_tirith_mod._resolved_path = _INSTALL_FAILED
_tirith_mod._install_failure_reason = "cosign_exec_failed"
with patch("tools.tirith_security.shutil.which", return_value=None), \
patch("tools.tirith_security._hermes_bin_dir", return_value="/nonexistent"), \
patch("tools.tirith_security._install_tirith") as mock_install:
result = _resolve_tirith_path("tirith")
assert result == "tirith" # fallback
mock_install.assert_not_called()
_tirith_mod._resolved_path = None
# ---------------------------------------------------------------------------
# HERMES_HOME isolation
# ---------------------------------------------------------------------------
class TestHermesHomeIsolation:
def test_hermes_bin_dir_respects_hermes_home(self):
"""_hermes_bin_dir must use HERMES_HOME, not hardcoded ~/.hermes."""
from tools.tirith_security import _hermes_bin_dir
import tempfile
tmpdir = tempfile.mkdtemp()
with patch.dict(os.environ, {"HERMES_HOME": tmpdir}):
result = _hermes_bin_dir()
assert result == os.path.join(tmpdir, "bin")
assert os.path.isdir(result)
assert _tirith_mod._resolve_tirith_path("tirith") == "tirith"
ensure.assert_not_called()
# ---------------------------------------------------------------------------
@@ -762,37 +529,3 @@ class TestEmojiVariationSelectorSuppression:
assert result["action"] == "warn"
assert result["findings"] == findings
# ---------------------------------------------------------------------------
# mkdtemp OSError → no_space (disk-full leak prevention)
# ---------------------------------------------------------------------------
class TestMkdtempOSErrorNoSpace:
"""When tempfile.mkdtemp raises OSError (e.g. disk full), _install_tirith
must return (None, "no_space") instead of propagating the exception.
This prevents the unbounded retry + temp-dir leak described in #51826.
"""
def test_mkdtemp_oserror_returns_no_space(self):
from tools.tirith_security import _install_tirith
with patch("tools.tirith_security.tempfile.mkdtemp",
side_effect=OSError(28, "No space left on device")):
result, reason = _install_tirith(log_failures=False)
assert result is None
assert reason == "no_space"
def test_mkdtemp_oserror_does_not_leak_tempdir(self):
"""No temp directory should remain after a mkdtemp failure."""
import glob
from tools.tirith_security import _install_tirith
before = set(glob.glob("/tmp/tirith-install-*"))
with patch("tools.tirith_security.tempfile.mkdtemp",
side_effect=OSError(28, "No space left on device")):
_install_tirith(log_failures=False)
after = set(glob.glob("/tmp/tirith-install-*"))
assert after - before == set()