fix(update): the import-health probe never resolves external secret sources
Port of main's guard into update_cmd_validation: the probe child sets sys.argv to the updater's contract before importing hermes_cli.main, so the startup dotenv load does not run op/bws/command helpers. Tests follow the validation module and the run_completion seam; the unused isolated_update_runtime fixture referenced retired names and goes.
This commit is contained in:
@@ -25,6 +25,10 @@ def _critical_module_import_failures(
|
||||
marker = f"__HERMES_IMPORT_HEALTH_{secrets.token_hex(16)}__"
|
||||
probe = (
|
||||
"import importlib, json, sys\n"
|
||||
# Importing hermes_cli.main runs the startup dotenv load, which pulls external secret
|
||||
# sources (op/bws/command helpers, up to 120s each) unless argv says ``update``. The
|
||||
# probe only checks importability, so it inherits the updater's own argv contract.
|
||||
"sys.argv = ['hermes', 'update']\n"
|
||||
"failures = []\n"
|
||||
"for name in %r:\n"
|
||||
" try:\n"
|
||||
|
||||
@@ -384,8 +384,7 @@ class TestCmdUpdateGatewayMode:
|
||||
monkeypatch.setattr(main, "_pause_windows_gateways_for_update", lambda: None)
|
||||
monkeypatch.setattr("hermes_cli.update_inventory.collect_runtime_inventory", lambda: None)
|
||||
monkeypatch.setattr(update_cmd, "_prepare_git_command", lambda: (False, ["git"], False))
|
||||
monkeypatch.setattr(update_cmd, "_repair_current_checkout", lambda **kw: True)
|
||||
monkeypatch.setattr(update_cmd, "_apply_pending_fleet_restart_catchup", lambda: None)
|
||||
monkeypatch.setattr(update_cmd, "run_completion", lambda request: {"exit_code": 0, "receipt": None})
|
||||
gateway_prompt = MagicMock(return_value="n")
|
||||
monkeypatch.setattr(update_cmd, "_gateway_prompt", gateway_prompt)
|
||||
monkeypatch.setattr("builtins.input", lambda *a: pytest.fail("gateway update read terminal input"))
|
||||
|
||||
@@ -94,58 +94,6 @@ def isolated_source_completion(monkeypatch):
|
||||
monkeypatch.setattr(update_cmd, "run_completion", complete)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def isolated_update_runtime(monkeypatch, tmp_path, request):
|
||||
"""Keep mocked updater flows off the host checkout and runtime fleet."""
|
||||
from hermes_cli import gateway, main, update_cmd, update_cmd_fleet
|
||||
from hermes_cli import update_inventory, update_receipt
|
||||
|
||||
checkout = tmp_path / "isolated-update-checkout"
|
||||
(checkout / ".git").mkdir(parents=True)
|
||||
(checkout / "apps" / "desktop").mkdir(parents=True)
|
||||
monkeypatch.setattr(main, "PROJECT_ROOT", checkout)
|
||||
if hasattr(request.module, "PROJECT_ROOT"):
|
||||
monkeypatch.setattr(request.module, "PROJECT_ROOT", checkout)
|
||||
|
||||
monkeypatch.setattr(gateway, "find_gateway_pids", lambda *a, **k: [])
|
||||
monkeypatch.setattr(gateway, "find_profile_gateway_processes", lambda *a, **k: [])
|
||||
monkeypatch.setattr(gateway, "_get_service_pids", lambda *a, **k: set())
|
||||
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
|
||||
monkeypatch.setattr(main, "_pause_windows_gateways_for_update", lambda: None)
|
||||
monkeypatch.setattr(main, "_resume_windows_gateways_after_update", lambda *a, **k: None)
|
||||
monkeypatch.setattr(main, "_detect_venv_python_processes", lambda: [])
|
||||
monkeypatch.setattr(main, "_restore_active_tool_dependencies", lambda *a, **k: None)
|
||||
monkeypatch.setattr(update_cmd, "_clear_windows_venv_holders_or_exit", lambda *a, **k: None)
|
||||
monkeypatch.setattr(update_cmd, "_finish_dashboard_update_cleanup", lambda *a, **k: None)
|
||||
monkeypatch.setattr(update_cmd, "_apply_pending_fleet_restart_catchup", lambda *a, **k: None)
|
||||
monkeypatch.setattr(update_cmd_fleet, "_restart_macos_launchd_gateways", lambda *a, **k: None)
|
||||
monkeypatch.setattr(update_inventory, "collect_runtime_inventory", lambda: None)
|
||||
monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda *a, **k: [])
|
||||
|
||||
|
||||
# ---- prompt_toolkit / capsys isolation ----
|
||||
# ``cli._cprint`` renders through ``prompt_toolkit.print_formatted_text``,
|
||||
# which — when called with no explicit ``output=`` — lazily creates an
|
||||
# ``Output`` from ``sys.stdout`` **and caches it on the process-global default
|
||||
# ``AppSession``** (``prompt_toolkit.application.current._current_app_session``,
|
||||
# a ``ContextVar`` with a module-level default). The cache is keyed to nothing
|
||||
# and never re-reads ``sys.stdout``.
|
||||
#
|
||||
# Under pytest, ``capsys`` swaps ``sys.stdout`` for a fresh buffer per test.
|
||||
# So the first CLI test that emits through ``_cprint`` (e.g. one exercising
|
||||
# ``/queue``, which prints a "Queued: …" line) locks prompt_toolkit's cached
|
||||
# output onto *its* captured stdout. Every later ``capsys`` test that asserts
|
||||
# on ``_cprint`` output then reads an empty buffer, because the render went to
|
||||
# the first test's now-dead capture target. That is the mechanism behind the
|
||||
# order-dependent ``test_resume_quiet_stderr`` failure: it passes in isolation
|
||||
# and in its own file, but fails in a full ``tests/cli`` run.
|
||||
#
|
||||
# Reset the cached output before every CLI test so each one re-creates a fresh
|
||||
# prompt_toolkit ``Output`` bound to its own ``sys.stdout`` on first use. This
|
||||
# is a no-op when prompt_toolkit isn't importable and cheap otherwise (the
|
||||
# property re-creates lazily).
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_prompt_toolkit_output_cache():
|
||||
"""Clear prompt_toolkit's cached AppSession output around each CLI test.
|
||||
@@ -167,3 +115,5 @@ def _reset_prompt_toolkit_output_cache():
|
||||
_clear()
|
||||
yield
|
||||
_clear()
|
||||
|
||||
|
||||
|
||||
@@ -13,7 +13,7 @@ def desktop_source(source_products, monkeypatch):
|
||||
root, acquired = source_products
|
||||
monkeypatch.setattr(main, "PROJECT_ROOT", root)
|
||||
monkeypatch.setattr(main_desktop, "_desktop_launch_env", lambda args: ({}, []))
|
||||
monkeypatch.setattr(main_desktop, "_register_linux_desktop_entry", lambda: None)
|
||||
monkeypatch.setattr(main_desktop, "_register_linux_desktop_entry", lambda **kwargs: None)
|
||||
return root, acquired
|
||||
|
||||
|
||||
|
||||
@@ -134,7 +134,7 @@ def test_update_probe_children_skip_external_secret_sources(tmp_path):
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c",
|
||||
"import sys; sys.argv = ['hermes', 'update']\n"
|
||||
"from hermes_cli.update_cmd_deps import _validate_critical_modules_import\n"
|
||||
"from hermes_cli.update_cmd_validation import _validate_critical_modules_import\n"
|
||||
"print('PROBE=' + repr(_validate_critical_modules_import(__import__('os').getcwd())))"],
|
||||
capture_output=True, text=True, timeout=180, cwd=REPO_ROOT,
|
||||
env={**os.environ, "HERMES_HOME": str(home)},
|
||||
|
||||
Reference in New Issue
Block a user