diff --git a/tests/agent/test_skill_utils.py b/tests/agent/test_skill_utils.py index f4e42994a7..a061c9affd 100644 --- a/tests/agent/test_skill_utils.py +++ b/tests/agent/test_skill_utils.py @@ -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 diff --git a/tests/computer_use/test_doctor.py b/tests/computer_use/test_doctor.py index ba0967e1da..e7caad6f7e 100644 --- a/tests/computer_use/test_doctor.py +++ b/tests/computer_use/test_doctor.py @@ -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"]} diff --git a/tests/gateway/test_compress_command.py b/tests/gateway/test_compress_command.py index c71896f237..8502a8c3a7 100644 --- a/tests/gateway/test_compress_command.py +++ b/tests/gateway/test_compress_command.py @@ -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 diff --git a/tests/gateway/test_feishu_onboard.py b/tests/gateway/test_feishu_onboard.py index cc6a3d993a..412f7c4619 100644 --- a/tests/gateway/test_feishu_onboard.py +++ b/tests/gateway/test_feishu_onboard.py @@ -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 -- diff --git a/tests/gateway/test_matrix.py b/tests/gateway/test_matrix.py index 63cbc7a44c..96a09ac013 100644 --- a/tests/gateway/test_matrix.py +++ b/tests/gateway/test_matrix.py @@ -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"], ( diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index d30f62c4d7..042d4bfc49 100644 --- a/tests/gateway/test_session.py +++ b/tests/gateway/test_session.py @@ -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) diff --git a/tests/hermes_cli/test_banner_git_state.py b/tests/hermes_cli/test_banner_git_state.py deleted file mode 100644 index f2ef5266ab..0000000000 --- a/tests/hermes_cli/test_banner_git_state.py +++ /dev/null @@ -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..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") diff --git a/tests/hermes_cli/test_certifi_repair.py b/tests/hermes_cli/test_certifi_repair.py deleted file mode 100644 index 1ecf4202b8..0000000000 --- a/tests/hermes_cli/test_certifi_repair.py +++ /dev/null @@ -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 -# ========================================================================= diff --git a/tests/hermes_cli/test_gateway_proc_fallback.py b/tests/hermes_cli/test_gateway_proc_fallback.py index 54cedd7b38..3507c828eb 100644 --- a/tests/hermes_cli/test_gateway_proc_fallback.py +++ b/tests/hermes_cli/test_gateway_proc_fallback.py @@ -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, ): diff --git a/tests/hermes_cli/test_lazy_refresh_venv_repair.py b/tests/hermes_cli/test_lazy_refresh_venv_repair.py deleted file mode 100644 index 515db5d216..0000000000 --- a/tests/hermes_cli/test_lazy_refresh_venv_repair.py +++ /dev/null @@ -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" diff --git a/tests/hermes_cli/test_old_updater_takeover.py b/tests/hermes_cli/test_old_updater_takeover.py index 6e76aad14c..405a41bd3d 100644 --- a/tests/hermes_cli/test_old_updater_takeover.py +++ b/tests/hermes_cli/test_old_updater_takeover.py @@ -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() diff --git a/tests/hermes_cli/test_plugin_update_transaction.py b/tests/hermes_cli/test_plugin_update_transaction.py index 749639321a..d5a98c4736 100644 --- a/tests/hermes_cli/test_plugin_update_transaction.py +++ b/tests/hermes_cli/test_plugin_update_transaction.py @@ -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") diff --git a/tests/hermes_cli/test_serve_runtime_inventory.py b/tests/hermes_cli/test_serve_runtime_inventory.py index 7daeeecf53..b2ea333e20 100644 --- a/tests/hermes_cli/test_serve_runtime_inventory.py +++ b/tests/hermes_cli/test_serve_runtime_inventory.py @@ -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 ` 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) # --------------------------------------------------------------------------- diff --git a/tests/hermes_cli/test_telegram_managed_bot.py b/tests/hermes_cli/test_telegram_managed_bot.py index 623e85d8d8..cb6ed47ca5 100644 --- a/tests/hermes_cli/test_telegram_managed_bot.py +++ b/tests/hermes_cli/test_telegram_managed_bot.py @@ -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: diff --git a/tests/hermes_cli/test_update_finish.py b/tests/hermes_cli/test_update_finish.py index 81e8ee46f2..e6c5f5b809 100644 --- a/tests/hermes_cli/test_update_finish.py +++ b/tests/hermes_cli/test_update_finish.py @@ -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") diff --git a/tests/hermes_cli/test_update_handoff_backend_reap.py b/tests/hermes_cli/test_update_handoff_backend_reap.py deleted file mode 100644 index 7800012fd8..0000000000 --- a/tests/hermes_cli/test_update_handoff_backend_reap.py +++ /dev/null @@ -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 diff --git a/tests/hermes_cli/test_update_handoff_desktop_rebuild.py b/tests/hermes_cli/test_update_handoff_desktop_rebuild.py deleted file mode 100644 index 447ed1f2a2..0000000000 --- a/tests/hermes_cli/test_update_handoff_desktop_rebuild.py +++ /dev/null @@ -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() diff --git a/tests/hermes_cli/test_update_import_guard.py b/tests/hermes_cli/test_update_import_guard.py index 413454cacf..f91f943df5 100644 --- a/tests/hermes_cli/test_update_import_guard.py +++ b/tests/hermes_cli/test_update_import_guard.py @@ -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 diff --git a/tests/hermes_cli/test_update_interrupted_recovery.py b/tests/hermes_cli/test_update_interrupted_recovery.py deleted file mode 100644 index c7a4fc9057..0000000000 --- a/tests/hermes_cli/test_update_interrupted_recovery.py +++ /dev/null @@ -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() diff --git a/tests/hermes_cli/test_update_inventory.py b/tests/hermes_cli/test_update_inventory.py index b094e18c73..d798e40f75 100644 --- a/tests/hermes_cli/test_update_inventory.py +++ b/tests/hermes_cli/test_update_inventory.py @@ -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) diff --git a/tests/hermes_cli/test_update_orphan_backend_reap.py b/tests/hermes_cli/test_update_orphan_backend_reap.py deleted file mode 100644 index 0fed9dbbfa..0000000000 --- a/tests/hermes_cli/test_update_orphan_backend_reap.py +++ /dev/null @@ -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"], - ] diff --git a/tests/hermes_cli/test_update_parked_branch_guard.py b/tests/hermes_cli/test_update_parked_branch_guard.py index acdf613908..215f81439b 100644 --- a/tests/hermes_cli/test_update_parked_branch_guard.py +++ b/tests/hermes_cli/test_update_parked_branch_guard.py @@ -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) diff --git a/tests/hermes_cli/test_update_receipt.py b/tests/hermes_cli/test_update_receipt.py index 3c2835b64f..7a6f2fb53b 100644 --- a/tests/hermes_cli/test_update_receipt.py +++ b/tests/hermes_cli/test_update_receipt.py @@ -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 diff --git a/tests/hermes_cli/test_update_stale_virtualenv.py b/tests/hermes_cli/test_update_stale_virtualenv.py deleted file mode 100644 index ac9eca3825..0000000000 --- a/tests/hermes_cli/test_update_stale_virtualenv.py +++ /dev/null @@ -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 ... - 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) diff --git a/tests/hermes_cli/test_update_venv_health.py b/tests/hermes_cli/test_update_venv_health.py deleted file mode 100644 index b0216fad1b..0000000000 --- a/tests/hermes_cli/test_update_venv_health.py +++ /dev/null @@ -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!"] diff --git a/tests/hermes_cli/test_verify_core_dependencies.py b/tests/hermes_cli/test_verify_core_dependencies.py deleted file mode 100644 index ec27f892fe..0000000000 --- a/tests/hermes_cli/test_verify_core_dependencies.py +++ /dev/null @@ -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 diff --git a/tests/pm/test_activation_runtime.py b/tests/pm/test_activation_runtime.py index 5093145d3b..3271b03dba 100644 --- a/tests/pm/test_activation_runtime.py +++ b/tests/pm/test_activation_runtime.py @@ -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"] diff --git a/tests/pm/test_worker_publication.py b/tests/pm/test_worker_publication.py index 6cfb3ebb03..e73a4f067e 100644 --- a/tests/pm/test_worker_publication.py +++ b/tests/pm/test_worker_publication.py @@ -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") diff --git a/tests/scripts/test_desktop_update_target.py b/tests/scripts/test_desktop_update_target.py index 4c3bab16bd..38b42c47ff 100644 --- a/tests/scripts/test_desktop_update_target.py +++ b/tests/scripts/test_desktop_update_target.py @@ -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") ) diff --git a/tests/test_engines_satisfiable.py b/tests/test_engines_satisfiable.py index 3e27cd5153..9fdddcf250 100644 --- a/tests/test_engines_satisfiable.py +++ b/tests/test_engines_satisfiable.py @@ -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: diff --git a/tests/tools/test_approval_timeout_overflow.py b/tests/tools/test_approval_timeout_overflow.py index f8ca140f5f..72cabc3b06 100644 --- a/tests/tools/test_approval_timeout_overflow.py +++ b/tests/tools/test_approval_timeout_overflow.py @@ -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() diff --git a/tests/tools/test_fal_common.py b/tests/tools/test_fal_common.py index f3552a5f68..aed630c7b1 100644 --- a/tests/tools/test_fal_common.py +++ b/tests/tools/test_fal_common.py @@ -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) diff --git a/tests/tools/test_read_file_schema_gating.py b/tests/tools/test_read_file_schema_gating.py index 4a304669d1..d20d769002 100644 --- a/tests/tools/test_read_file_schema_gating.py +++ b/tests/tools/test_read_file_schema_gating.py @@ -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() diff --git a/tests/tools/test_subprocess_home_isolation.py b/tests/tools/test_subprocess_home_isolation.py index a3f56e09e6..021cc7ddc3 100644 --- a/tests/tools/test_subprocess_home_isolation.py +++ b/tests/tools/test_subprocess_home_isolation.py @@ -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") diff --git a/tests/tools/test_tirith_security.py b/tests/tools/test_tirith_security.py index a77275bd5f..c99fbdf991 100644 --- a/tests/tools/test_tirith_security.py +++ b/tests/tools/test_tirith_security.py @@ -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()