"""Shared fixtures for the hermes-agent test suite. Hermetic-test invariants enforced here (see AGENTS.md for rationale): 1. **No credential env vars.** All provider/credential-shaped env vars (ending in _API_KEY, _TOKEN, _SECRET, _PASSWORD, _CREDENTIALS, etc.) are unset before every test. Local developer keys cannot leak in. 2. **Isolated HERMES_HOME.** HERMES_HOME points to a per-test tempdir so code reading ``~/.hermes/*`` via ``get_hermes_home()`` can't see the real one. (We do NOT also redirect HOME — that broke subprocesses in CI. Code using ``Path.home() / ".hermes"`` instead of the canonical ``get_hermes_home()`` is a bug to fix at the callsite.) 3. **Deterministic runtime.** TZ=UTC, LANG=C.UTF-8, PYTHONHASHSEED=0. 4. **No HERMES_SESSION_* inheritance** — the agent's current gateway session must not leak into tests. These invariants make the local test run match CI closely. Gaps that remain (CPU count, xdist worker count) are addressed by the canonical test runner at ``scripts/run_tests.sh``. """ import asyncio import os import re import sys from pathlib import Path from unittest.mock import patch import pytest # Ensure project root is importable PROJECT_ROOT = Path(__file__).parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) # ── Plugin directory sys.path shadow fix ──────────────────────────────────── # pytest adds ``plugins/model-providers/`` to sys.path because # ``plugins/model-providers/anthropic/__init__.py`` (a provider profile) exists. # This makes ``import anthropic`` shadow the real SDK package with the plugin # directory. We fix this via pytest_load_initial_conftests (runs before pytest # adds package roots) AND inline here as belt-and-suspenders. _bad_path = str(PROJECT_ROOT / "plugins" / "model-providers") while _bad_path in sys.path: sys.path.remove(_bad_path) if "anthropic" in sys.modules and not hasattr(sys.modules["anthropic"], "Anthropic"): del sys.modules["anthropic"] def pytest_load_initial_conftests(early_config, parser, args): """Remove plugin dirs from sys.path before pytest adds package roots. This must run as early as possible — before hermes_agent_anthropic is imported — to prevent ``plugins/model-providers/anthropic/__init__.py`` (a provider profile) from shadowing the real ``anthropic`` SDK. """ _bad = str(PROJECT_ROOT / "plugins" / "model-providers") while _bad in sys.path: sys.path.remove(_bad) if "anthropic" in sys.modules and not hasattr(sys.modules["anthropic"], "Anthropic"): del sys.modules["anthropic"] # ── Per-file process isolation ────────────────────────────────────────────── # Tests run via ``scripts/run_tests_parallel.py``, which spawns a fresh # ``python -m pytest `` subprocess per test file. Cross-file state # leakage (module-level dicts, ContextVars, caches) is impossible: each # file gets a clean Python interpreter. Intra-file ordering is the test # author's responsibility — if test A in foo.py mutates state that test B # in foo.py reads, that's a real bug to fix in the file (it would also # bite anyone running ``pytest tests/foo.py`` directly). # # This replaces the historic _reset_module_state autouse fixture (manual # state clearing) and the brief experiment with subprocess-per-test # isolation (too slow at ~17k tests). # # See ``scripts/run_tests_parallel.py`` for the runner. # ── Credential env-var filter ────────────────────────────────────────────── # # Any env var in the current process matching ONE of these patterns is # unset for every test. Developers' local keys cannot leak into assertions # about "auto-detect provider when key present". _CREDENTIAL_SUFFIXES = ( "_API_KEY", "_TOKEN", "_SECRET", "_PASSWORD", "_CREDENTIALS", "_ACCESS_KEY", "_SECRET_ACCESS_KEY", "_PRIVATE_KEY", "_OAUTH_TOKEN", "_WEBHOOK_SECRET", "_ENCRYPT_KEY", "_APP_SECRET", "_CLIENT_SECRET", "_CORP_SECRET", "_AES_KEY", ) # Explicit names (for ones that don't fit the suffix pattern) _CREDENTIAL_NAMES = frozenset({ "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN", "ANTHROPIC_TOKEN", "FAL_KEY", "GH_TOKEN", "GITHUB_TOKEN", "OPENAI_API_KEY", "OPENROUTER_API_KEY", "NOUS_API_KEY", "GEMINI_API_KEY", "GOOGLE_API_KEY", "GROQ_API_KEY", "XAI_API_KEY", "MISTRAL_API_KEY", "DEEPSEEK_API_KEY", "KIMI_API_KEY", "MOONSHOT_API_KEY", "GLM_API_KEY", "ZAI_API_KEY", "MINIMAX_API_KEY", "OLLAMA_API_KEY", "OPENVIKING_API_KEY", "COPILOT_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", "BROWSERBASE_API_KEY", "FIRECRAWL_API_KEY", "PARALLEL_API_KEY", "EXA_API_KEY", "TAVILY_API_KEY", "WANDB_API_KEY", "ELEVENLABS_API_KEY", "HONCHO_API_KEY", "MEM0_API_KEY", "SUPERMEMORY_API_KEY", "RETAINDB_API_KEY", "HINDSIGHT_API_KEY", "HINDSIGHT_LLM_API_KEY", "DAYTONA_API_KEY", "TWILIO_AUTH_TOKEN", "TELEGRAM_BOT_TOKEN", "DISCORD_BOT_TOKEN", "SLACK_BOT_TOKEN", "SLACK_APP_TOKEN", "MATTERMOST_TOKEN", "MATRIX_ACCESS_TOKEN", "MATRIX_PASSWORD", "MATRIX_RECOVERY_KEY", "HASS_TOKEN", "EMAIL_PASSWORD", "BLUEBUBBLES_PASSWORD", "FEISHU_APP_SECRET", "FEISHU_ENCRYPT_KEY", "FEISHU_VERIFICATION_TOKEN", "DINGTALK_CLIENT_SECRET", "QQ_CLIENT_SECRET", "QQ_STT_API_KEY", "WECOM_SECRET", "WECOM_CALLBACK_CORP_SECRET", "WECOM_CALLBACK_TOKEN", "WECOM_CALLBACK_ENCODING_AES_KEY", "WEIXIN_TOKEN", "MODAL_TOKEN_ID", "MODAL_TOKEN_SECRET", "TERMINAL_SSH_KEY", "SUDO_PASSWORD", "GATEWAY_PROXY_KEY", "API_SERVER_KEY", "TOOL_GATEWAY_USER_TOKEN", "TELEGRAM_WEBHOOK_SECRET", "WEBHOOK_SECRET", "VOICE_TOOLS_OPENAI_KEY", "BROWSER_USE_API_KEY", "CUSTOM_API_KEY", "GATEWAY_PROXY_URL", "GEMINI_BASE_URL", "OPENAI_BASE_URL", "OPENROUTER_BASE_URL", "OLLAMA_BASE_URL", "GROQ_BASE_URL", "XAI_BASE_URL", "ANTHROPIC_BASE_URL", }) def _looks_like_credential(name: str) -> bool: """True if env var name matches a credential-shaped pattern.""" if name in _CREDENTIAL_NAMES: return True return any(name.endswith(suf) for suf in _CREDENTIAL_SUFFIXES) # HERMES_* vars that change test behavior by being set. Unset all of these # unconditionally — individual tests that need them set do so explicitly. _HERMES_BEHAVIORAL_VARS = frozenset({ "HERMES_YOLO_MODE", "HERMES_INTERACTIVE", "HERMES_QUIET", "HERMES_TOOL_PROGRESS", "HERMES_TOOL_PROGRESS_MODE", "HERMES_MAX_ITERATIONS", "HERMES_SESSION_PLATFORM", "HERMES_SESSION_CHAT_ID", "HERMES_SESSION_CHAT_NAME", "HERMES_SESSION_THREAD_ID", "HERMES_SESSION_SOURCE", "HERMES_SESSION_KEY", "HERMES_GATEWAY_SESSION", "HERMES_PLATFORM", "HERMES_MODEL", "HERMES_INFERENCE_MODEL", "HERMES_INFERENCE_PROVIDER", "HERMES_TUI_PROVIDER", "HERMES_MANAGED", "HERMES_DEV", "HERMES_CONTAINER", "HERMES_EPHEMERAL_SYSTEM_PROMPT", "HERMES_TIMEZONE", "HERMES_REDACT_SECRETS", "HERMES_BACKGROUND_NOTIFICATIONS", "HERMES_EXEC_ASK", "HERMES_HOME_MODE", "HERMES_AGENT_USE_LEGACY_SESSION_KEYS", # Kanban path/board pins must never leak from a developer shell or # dispatched worker into tests; otherwise tests can write fake tasks to # the real ~/.hermes/kanban.db instead of the per-test HERMES_HOME. "HERMES_KANBAN_DB", "HERMES_KANBAN_BOARD", "HERMES_KANBAN_HOME", "HERMES_KANBAN_WORKSPACES_ROOT", "HERMES_KANBAN_LOGS_ROOT", "HERMES_KANBAN_TASK", "HERMES_KANBAN_WORKSPACE", "HERMES_KANBAN_RUN_ID", "HERMES_KANBAN_CLAIM_LOCK", "HERMES_KANBAN_DISPATCH_IN_GATEWAY", "HERMES_TENANT", # Dashboard OAuth auth gate (PR #30156). When set, the bundled # dashboard-auth `nous` plugin auto-registers itself on plugin discovery, # which is triggered by any `/api/status` call. That leaks a provider # into the dashboard_auth registry across tests in the same worker and # makes assertions like `auth_providers == []` flaky. CI never sets # these, so production tests must not see them either. "HERMES_DASHBOARD_OAUTH_CLIENT_ID", "HERMES_DASHBOARD_PORTAL_URL", "TERMINAL_CWD", "TERMINAL_ENV", "TERMINAL_CONTAINER_CPU", "TERMINAL_CONTAINER_DISK", "TERMINAL_CONTAINER_MEMORY", "TERMINAL_CONTAINER_PERSISTENT", "TERMINAL_DOCKER_RUN_AS_HOST_USER", "BROWSER_CDP_URL", "CAMOFOX_URL", # Platform allowlists — not credentials, but if set from any source # (user shell, earlier leaky test, CI env), they change gateway auth # behavior and flake button-authorization tests. "TELEGRAM_ALLOWED_USERS", "DISCORD_ALLOWED_USERS", "WHATSAPP_ALLOWED_USERS", "SLACK_ALLOWED_USERS", "SIGNAL_ALLOWED_USERS", "SIGNAL_GROUP_ALLOWED_USERS", "EMAIL_ALLOWED_USERS", "SMS_ALLOWED_USERS", "MATTERMOST_ALLOWED_USERS", "MATRIX_ALLOWED_USERS", "DINGTALK_ALLOWED_USERS", "FEISHU_ALLOWED_USERS", "WECOM_ALLOWED_USERS", "GATEWAY_ALLOWED_USERS", "GATEWAY_ALLOW_ALL_USERS", "TELEGRAM_ALLOW_ALL_USERS", "DISCORD_ALLOW_ALL_USERS", "WHATSAPP_ALLOW_ALL_USERS", "SLACK_ALLOW_ALL_USERS", "SIGNAL_ALLOW_ALL_USERS", "EMAIL_ALLOW_ALL_USERS", "SMS_ALLOW_ALL_USERS", # Gateway home channels are set by /sethome in real profiles. Tests that # exercise dashboard notification toggles must opt in explicitly or they # can accidentally subscribe against a developer's real home channel. "TELEGRAM_HOME_CHANNEL", "TELEGRAM_HOME_CHANNEL_THREAD_ID", "TELEGRAM_HOME_CHANNEL_NAME", "TELEGRAM_CRON_THREAD_ID", "DISCORD_HOME_CHANNEL", "DISCORD_HOME_CHANNEL_THREAD_ID", "DISCORD_HOME_CHANNEL_NAME", "SLACK_HOME_CHANNEL", "SLACK_HOME_CHANNEL_THREAD_ID", "SLACK_HOME_CHANNEL_NAME", "WHATSAPP_HOME_CHANNEL", "WHATSAPP_HOME_CHANNEL_THREAD_ID", "WHATSAPP_HOME_CHANNEL_NAME", "SIGNAL_HOME_CHANNEL", "SIGNAL_HOME_CHANNEL_THREAD_ID", "SIGNAL_HOME_CHANNEL_NAME", "EMAIL_HOME_CHANNEL", "EMAIL_HOME_CHANNEL_THREAD_ID", "EMAIL_HOME_CHANNEL_NAME", "SMS_HOME_CHANNEL", "SMS_HOME_CHANNEL_THREAD_ID", "SMS_HOME_CHANNEL_NAME", "MATTERMOST_HOME_CHANNEL", "MATTERMOST_HOME_CHANNEL_THREAD_ID", "MATTERMOST_HOME_CHANNEL_NAME", "MATRIX_HOME_CHANNEL", "MATRIX_HOME_CHANNEL_THREAD_ID", "MATRIX_HOME_CHANNEL_NAME", "DINGTALK_HOME_CHANNEL", "DINGTALK_HOME_CHANNEL_THREAD_ID", "DINGTALK_HOME_CHANNEL_NAME", "FEISHU_HOME_CHANNEL", "FEISHU_HOME_CHANNEL_THREAD_ID", "FEISHU_HOME_CHANNEL_NAME", "WECOM_HOME_CHANNEL", "WECOM_HOME_CHANNEL_THREAD_ID", "WECOM_HOME_CHANNEL_NAME", # API server bind/auth settings are common in local gateway profiles and # change adapter defaults plus load_gateway_config() enablement. Tests that # need them set opt in explicitly with monkeypatch. "API_SERVER_ENABLED", "API_SERVER_HOST", "API_SERVER_PORT", "API_SERVER_KEY", "API_SERVER_CORS_ORIGINS", "API_SERVER_MODEL_NAME", # Platform gating — set by load_gateway_config() as a side effect when # a config.yaml is present, so individual test bodies that call the # loader leak these values into later tests in the same process. # Force-clear on every test setup so the leak can't happen. "SLACK_REQUIRE_MENTION", "SLACK_STRICT_MENTION", "SLACK_FREE_RESPONSE_CHANNELS", "SLACK_ALLOW_BOTS", "SLACK_REACTIONS", "DISCORD_REQUIRE_MENTION", "DISCORD_FREE_RESPONSE_CHANNELS", "TELEGRAM_REQUIRE_MENTION", "WHATSAPP_REQUIRE_MENTION", "DINGTALK_REQUIRE_MENTION", "MATRIX_REQUIRE_MENTION", }) @pytest.fixture(autouse=True) def _hermetic_environment(tmp_path, monkeypatch): """Blank out all credential/behavioral env vars so local and CI match. Also redirects HOME and HERMES_HOME to per-test tempdirs so code that reads ``~/.hermes/*`` can't touch the real one, and pins TZ/LANG so datetime/locale-sensitive tests are deterministic. """ # 1. Blank every credential-shaped env var that's currently set. for name in list(os.environ.keys()): if _looks_like_credential(name): monkeypatch.delenv(name, raising=False) # 2. Blank behavioral HERMES_* vars that could change test semantics. for name in _HERMES_BEHAVIORAL_VARS: monkeypatch.delenv(name, raising=False) # 3. Redirect HERMES_HOME to a per-test tempdir. Code that reads # ``~/.hermes/*`` via ``get_hermes_home()`` now gets the tempdir. # # NOTE: We do NOT also redirect HOME. Doing so broke CI because # some tests (and their transitive deps) spawn subprocesses that # inherit HOME and expect it to be stable. If a test genuinely # needs HOME isolated, it should set it explicitly in its own # fixture. Any code in the codebase reading ``~/.hermes/*`` via # ``Path.home() / ".hermes"`` instead of ``get_hermes_home()`` # is a bug to fix at the callsite. fake_hermes_home = tmp_path / "hermes_test" fake_hermes_home.mkdir() (fake_hermes_home / "sessions").mkdir() (fake_hermes_home / "cron").mkdir() (fake_hermes_home / "memories").mkdir() (fake_hermes_home / "skills").mkdir() monkeypatch.setenv("HERMES_HOME", str(fake_hermes_home)) # 4. Deterministic locale / timezone / hashseed. CI runs in UTC with # C.UTF-8 locale; local dev often doesn't. Pin everything. monkeypatch.setenv("TZ", "UTC") monkeypatch.setenv("LANG", "C.UTF-8") monkeypatch.setenv("LC_ALL", "C.UTF-8") monkeypatch.setenv("PYTHONHASHSEED", "0") # 4b. Disable AWS IMDS lookups. Without this, any test that ends up # calling has_aws_credentials() / resolve_aws_auth_env_var() # (e.g. provider auto-detect, status command, cron run_job) burns # ~2s waiting for the metadata service at 169.254.169.254 to time # out. Tests don't run on EC2 — IMDS is always unreachable here. monkeypatch.setenv("AWS_EC2_METADATA_DISABLED", "true") monkeypatch.setenv("AWS_METADATA_SERVICE_TIMEOUT", "1") monkeypatch.setenv("AWS_METADATA_SERVICE_NUM_ATTEMPTS", "1") # Tirith auto-installs from GitHub when enabled and missing. Unit tests # should never perform that implicit network/bootstrap path; Tirith-specific # tests opt back in by patching the security config directly. monkeypatch.setenv("TIRITH_ENABLED", "false") # 5. Reset plugin singleton so tests don't leak plugins from # ~/.hermes/plugins/ (which, per step 3, is now empty — but the # singleton might still be cached from a previous test). try: import hermes_cli.plugins as _plugins_mod monkeypatch.setattr(_plugins_mod, "_plugin_manager", None) except Exception: pass # Explicitly clear provider-specific base URL overrides that don't match # the generic credential-shaped env-var filter above. monkeypatch.delenv("GMI_API_KEY", raising=False) monkeypatch.delenv("GMI_BASE_URL", raising=False) # Backward-compat alias — old tests reference this fixture name. Keep it # as a no-op wrapper so imports don't break. @pytest.fixture(autouse=True) def _isolate_hermes_home(_hermetic_environment): """Alias preserved for any test that yields this name explicitly.""" return None # ── Module-level state reset — replaced by per-file process isolation ────── # # Each test FILE runs in a freshly-spawned ``python -m pytest `` # subprocess via ``scripts/run_tests_parallel.py``, so module-level dicts / # sets / ContextVars from tests in one file cannot leak into tests in # another file. No manual per-module clearing needed. # # Within a single file, ordering is the author's responsibility. If your # tests in the same file share mutable state, either reset it explicitly # in a fixture or split them across files. # # The skill ``test-suite-cascade-diagnosis`` documents the cascade patterns # this replaces; the running example was ``test_command_guards`` failing # 12/15 CI runs because ``tools.approval._session_approved`` carried # approvals from one test's session into another's. @pytest.fixture() def tmp_dir(tmp_path): """Provide a temporary directory that is cleaned up automatically.""" return tmp_path @pytest.fixture() def mock_config(): """Return a minimal hermes config dict suitable for unit tests.""" return { "model": "test/mock-model", "toolsets": ["terminal", "file"], "max_turns": 10, "terminal": { "backend": "local", "cwd": "/tmp", "timeout": 30, }, "compression": {"enabled": False}, "memory": {"memory_enabled": False, "user_profile_enabled": False}, "command_allowlist": [], } # ── Per-test timeout — handled by the isolation plugin ───────────────────── # # The subprocess-per-test plugin enforces the configured ``isolate_timeout`` # ini key by terminating the child if it overruns. The old SIGALRM-based # fixture (POSIX-only, didn't work on Windows) is gone. @pytest.fixture(autouse=True) def _ensure_current_event_loop(request): """Provide a default event loop for sync tests that call get_event_loop(). Python 3.11+ no longer guarantees a current loop for plain synchronous tests. A number of gateway tests still use asyncio.get_event_loop().run_until_complete(...). Ensure they always have a usable loop without interfering with pytest-asyncio's own loop management for @pytest.mark.asyncio tests. On Python 3.12+, ``asyncio.get_event_loop_policy().get_event_loop()`` with no *running* loop emits DeprecationWarning; skip that path and install a fresh loop via ``new_event_loop()`` instead. """ if request.node.get_closest_marker("asyncio") is not None: yield return loop = None try: loop = asyncio.get_running_loop() except RuntimeError: pass if loop is None and sys.version_info < (3, 12): try: loop = asyncio.get_event_loop_policy().get_event_loop() except RuntimeError: loop = None created = loop is None or loop.is_closed() if created: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: yield finally: if created and loop is not None: try: loop.close() finally: asyncio.set_event_loop(None) # ── Live-system guard ────────────────────────────────────────────────────── # # Several test files exercise the gateway-restart / kill code paths # (``cmd_update``, ``kill_gateway_processes``, ``stop_profile_gateway``). # When a single test forgets to mock either ``os.kill`` or the global # ``find_gateway_pids`` helper, the real call leaks out of the hermetic # environment and finds the developer's live ``hermes-gateway`` process # via ``psutil`` — sending it SIGTERM mid-test. The shutdown forensics in # PR #23285 caught this happening 5+ times in 3 days, every time # correlated with a ``tests/hermes_cli/`` pytest run starting up. # # This fixture makes the leak impossible by intercepting the two # primitives that actually do damage: # # • ``os.kill`` rejects any PID outside the test process subtree with # a hard ``RuntimeError`` so the offending test gets a stack trace # instead of silently murdering the real gateway. # • ``subprocess.run`` / ``subprocess.Popen`` / ``call`` / ``check_call`` / # ``check_output`` reject any ``systemctl ... hermes-gateway`` # invocation that would mutate the live unit. Read-only systemctl # calls (``status``, ``show``, ``list-units``) still pass through. # # We intentionally do NOT stub ``find_gateway_pids`` / ``_scan_gateway_pids`` # here — tests of those functions themselves need the real implementation. # Even if a test gets the live gateway PID back from a real scan, the # ``os.kill`` guard above catches the actual signal call, and the # ``systemctl`` guard catches the systemd path. Discovery without # delivery is harmless. _LIVE_SYSTEM_GUARD_BYPASS_MARK = "live_system_guard_bypass" def pytest_configure(config): # noqa: D401 — pytest hook """Register markers used by hermetic conftest.""" config.addinivalue_line( "markers", f"{_LIVE_SYSTEM_GUARD_BYPASS_MARK}: bypass the live-system guard " "(only for tests that genuinely need real os.kill / subprocess " "behaviour — e.g. PTY tests that signal their own child).", ) def pytest_collection_finish(session) -> None: # noqa: D401 — pytest hook """Warn when pytest is invoked directly instead of via the parallel runner. The canonical runner (scripts/run_tests_parallel.py) spawns one pytest subprocess per file, giving each a fresh Python interpreter. This prevents cross-file module-level state leakage (especially the plugin-registry singleton) from causing ordering-dependent failures. Running ``pytest tests/`` directly collapses all files into one process and WILL see spurious failures that don't exist in CI. The runner sets HERMES_PARALLEL_RUNNER=1 in each subprocess so we can detect the difference here. """ if os.environ.get("HERMES_PARALLEL_RUNNER"): return # launched by the runner — all good n = len(session.items) if n < 50: return # single-file or tiny run — fine to use bare pytest _YELLOW = "\033[33m" _BOLD = "\033[1m" _RESET = "\033[0m" print( f"\n{_BOLD}{_YELLOW}⚠ hermes-agent test suite warning{_RESET}\n" f" You're running {n} tests directly under pytest.\n" f" Cross-file module-state leakage (esp. the plugin registry\n" f" singleton) can cause ordering-dependent failures that don't\n" f" exist in CI.\n\n" f" Use the canonical runner instead:\n" f" {_BOLD}python scripts/run_tests_parallel.py{_RESET}\n\n" f" To suppress this warning (e.g. for a focused multi-file run):\n" f" {_BOLD}HERMES_PARALLEL_RUNNER=1 pytest ...{_RESET}\n", file=sys.stderr, ) @pytest.fixture(autouse=True) def _live_system_guard(request, monkeypatch): """Block real os.kill / systemctl / gateway-pid scans during tests. See block comment above for the why. Tests that genuinely need real signal delivery (e.g. PTY tests that SIGINT their own child) can opt out with ``@pytest.mark.live_system_guard_bypass``. Coverage (every primitive that can deliver a signal to or otherwise terminate a foreign process): • os.kill, os.killpg (POSIX) • subprocess.run / Popen / call / check_call / check_output • subprocess.getoutput / getstatusoutput • os.system / os.popen • pty.spawn • asyncio.create_subprocess_exec / create_subprocess_shell Subprocess inspection looks at the WHOLE command string (not just tokens[0]), so ``bash -c "systemctl restart hermes-gateway"``, ``sudo systemctl ...``, ``env systemctl ...``, ``setsid systemctl ...`` are all caught. ``pkill``/``killall``/``taskkill`` invocations targeting hermes/python patterns are also blocked. """ if request.node.get_closest_marker(_LIVE_SYSTEM_GUARD_BYPASS_MARK): yield return import os as _os import shlex as _shlex import subprocess as _subprocess test_pid = _os.getpid() # Capture the test process's existing children at fixture start — # any *new* children spawned by the test are also allowlisted via # the live psutil walk below. Static set keeps the fast path cheap. try: import psutil as _psutil _initial_children = { c.pid for c in _psutil.Process(test_pid).children(recursive=True) } except Exception: _psutil = None _initial_children = set() def _is_own_subtree(pid: int) -> bool: # PID 0 means "our own process group"; -1 means "every process we # can signal". Both are dangerous when paired with SIGTERM/SIGKILL, # but pid 0 is technically scoped to our group so allow it; pid -1 # is treated as foreign (refuse). if pid == 0: return True if pid < 0: return False if pid == test_pid or pid in _initial_children: return True if _psutil is None: return False try: walker = _psutil.Process(pid) except Exception: # Stale PID — kill would be a no-op anyway, allow it. return True try: for parent in walker.parents(): if parent.pid == test_pid: return True except Exception: return False return False real_kill = _os.kill def _guarded_kill(pid, sig, *args, **kwargs): if _is_own_subtree(int(pid)): return real_kill(pid, sig, *args, **kwargs) raise RuntimeError( f"tests/conftest.py live-system guard: blocked os.kill(" f"{pid}, {sig}) — PID is outside the test process subtree. " "If this fired in CI it means the test reached a real " "kill_gateway_processes / stop_profile_gateway / cmd_update " "code path without mocking find_gateway_pids and os.kill. " "Mock both, or mark the test with " "@pytest.mark.live_system_guard_bypass if real signal " "delivery is genuinely required." ) monkeypatch.setattr(_os, "kill", _guarded_kill) # ``os.killpg`` is the same risk class — sends a signal to every # process in a group. The gateway is a session leader (its own # PGID == its PID), so killpg(gateway_pid, SIGTERM) is a one-shot # kill of the live process. Allow it only when the target PGID is # the test process's own group. if hasattr(_os, "killpg"): real_killpg = _os.killpg own_pgid = _os.getpgrp() def _guarded_killpg(pgid, sig, *args, **kwargs): if int(pgid) == own_pgid or _is_own_subtree(int(pgid)): return real_killpg(pgid, sig, *args, **kwargs) raise RuntimeError( f"tests/conftest.py live-system guard: blocked " f"os.killpg({pgid}, {sig}) — PGID is outside the test " "process group. See _live_system_guard for the why." ) monkeypatch.setattr(_os, "killpg", _guarded_killpg) # ── Subprocess command-string inspection (whole-line) ────────── _HERMES_TOKENS = ( "hermes-gateway", "hermes.service", "hermes_cli.main gateway", "hermes_cli/main.py gateway", "gateway/run.py", "hermes gateway", ) _MUTATING_VERBS = ( "restart", "start", "stop", "kill", "reload", "reset-failed", "enable", "disable", "mask", "unmask", "daemon-reload", "try-restart", "reload-or-restart", ) _PROCESS_KILLERS = ("pkill", "killall", "taskkill", "skill", "fuser") def _cmd_to_string(cmd) -> str: if cmd is None: return "" if isinstance(cmd, (bytes, bytearray)): try: return bytes(cmd).decode(errors="replace") except Exception: return "" if isinstance(cmd, str): return cmd if isinstance(cmd, (list, tuple)): try: return " ".join(str(t) for t in cmd) except Exception: return "" return str(cmd) def _matches_hermes_gateway(cmd_str: str) -> bool: low = cmd_str.lower() return any(tok in low for tok in _HERMES_TOKENS) def _is_blocked_systemctl(cmd) -> bool: cmd_str = _cmd_to_string(cmd) if "systemctl" not in cmd_str: return False if not _matches_hermes_gateway(cmd_str): return False try: tokens = _shlex.split(cmd_str) except ValueError: tokens = cmd_str.split() return any(verb in tokens for verb in _MUTATING_VERBS) def _is_process_killer(cmd) -> bool: cmd_str = _cmd_to_string(cmd) try: tokens = _shlex.split(cmd_str) except ValueError: tokens = cmd_str.split() if not tokens: return False for tok in tokens: head = tok.rsplit("/", 1)[-1].rsplit("\\", 1)[-1] if head in _PROCESS_KILLERS: low = cmd_str.lower() # pkill -f pattern: catch hermes-themed patterns + a # plain "python" -f which would catch the live gateway # whose cmdline contains "python -m hermes_cli.main". if ( "hermes" in low or "gateway" in low or ("python" in low and "-f" in tokens) ): return True return False def _check_subprocess_cmd(name, cmd): if _is_blocked_systemctl(cmd): raise RuntimeError( f"tests/conftest.py live-system guard: blocked " f"subprocess.{name}({cmd!r}) — would mutate the " "live hermes-gateway systemd unit. Mock " "subprocess.run / _run_systemctl in the test, or " "mark with @pytest.mark.live_system_guard_bypass." ) if _is_process_killer(cmd): raise RuntimeError( f"tests/conftest.py live-system guard: blocked " f"subprocess.{name}({cmd!r}) — process-killer command " "targeting hermes/python could hit the live gateway. " "Mark with @pytest.mark.live_system_guard_bypass if " "intentional." ) def _wrap_subprocess(name, real): def _guarded(cmd, *args, **kwargs): _check_subprocess_cmd(name, cmd) return real(cmd, *args, **kwargs) _guarded.__name__ = f"_guarded_{name}" # Make the wrapper subscriptable like the wrapped callable when # the wrapped object is. ``subprocess.Popen[bytes]`` is used as # a type annotation in third-party packages (mcp, etc.); replacing # ``Popen`` with a plain function breaks ``Popen[bytes]`` at # import time. Defer ``__class_getitem__`` to the original. if hasattr(real, "__class_getitem__"): _guarded.__class_getitem__ = real.__class_getitem__ return _guarded def _wrap_popen(): """Subclass Popen so isinstance checks AND Popen[bytes] still work.""" real = _subprocess.Popen class _GuardedPopen(real): # type: ignore[misc, valid-type] def __init__(self, cmd, *args, **kwargs): _check_subprocess_cmd("Popen", cmd) super().__init__(cmd, *args, **kwargs) _GuardedPopen.__name__ = "Popen" _GuardedPopen.__qualname__ = "Popen" return _GuardedPopen real_run = _subprocess.run real_popen = _subprocess.Popen real_call = _subprocess.call real_check_call = _subprocess.check_call real_check_output = _subprocess.check_output real_getoutput = _subprocess.getoutput real_getstatusoutput = _subprocess.getstatusoutput monkeypatch.setattr(_subprocess, "run", _wrap_subprocess("run", real_run)) monkeypatch.setattr(_subprocess, "Popen", _wrap_popen()) monkeypatch.setattr(_subprocess, "call", _wrap_subprocess("call", real_call)) monkeypatch.setattr( _subprocess, "check_call", _wrap_subprocess("check_call", real_check_call) ) monkeypatch.setattr( _subprocess, "check_output", _wrap_subprocess("check_output", real_check_output), ) monkeypatch.setattr( _subprocess, "getoutput", _wrap_subprocess("getoutput", real_getoutput) ) monkeypatch.setattr( _subprocess, "getstatusoutput", _wrap_subprocess("getstatusoutput", real_getstatusoutput), ) # os.system / os.popen — same risk class, completely unwrapped before. real_os_system = _os.system real_os_popen = _os.popen def _guarded_os_system(command): _check_subprocess_cmd("os.system", command) return real_os_system(command) def _guarded_os_popen(cmd, *args, **kwargs): _check_subprocess_cmd("os.popen", cmd) return real_os_popen(cmd, *args, **kwargs) monkeypatch.setattr(_os, "system", _guarded_os_system) monkeypatch.setattr(_os, "popen", _guarded_os_popen) # pty.spawn — POSIX-only. try: import pty as _pty if hasattr(_pty, "spawn"): real_pty_spawn = _pty.spawn def _guarded_pty_spawn(argv, *args, **kwargs): _check_subprocess_cmd("pty.spawn", argv) return real_pty_spawn(argv, *args, **kwargs) monkeypatch.setattr(_pty, "spawn", _guarded_pty_spawn) except Exception: pass # asyncio.create_subprocess_* — bypasses subprocess module entirely. try: import asyncio as _asyncio real_async_exec = _asyncio.create_subprocess_exec real_async_shell = _asyncio.create_subprocess_shell async def _guarded_async_exec(program, *args, **kwargs): _check_subprocess_cmd( "asyncio.create_subprocess_exec", [program, *args] ) return await real_async_exec(program, *args, **kwargs) async def _guarded_async_shell(cmd, *args, **kwargs): _check_subprocess_cmd("asyncio.create_subprocess_shell", cmd) return await real_async_shell(cmd, *args, **kwargs) monkeypatch.setattr(_asyncio, "create_subprocess_exec", _guarded_async_exec) monkeypatch.setattr( _asyncio, "create_subprocess_shell", _guarded_async_shell ) except Exception: pass yield # ── Anthropic registry seed (shared across all test subdirs) ──────────────── # Defined in root conftest so tests/hermes_cli, tests/run_agent, # tests/tools, tests/gateway all get it. tests/agent has its own copy too. from contextlib import contextmanager from unittest.mock import MagicMock __all__ = ["mock_anthropic_provider"] def _mock_endpoint_speaks_anthropic_messages(base_url: str) -> bool: """Functional mock — detects Anthropic-wire endpoints by URL pattern. Reproduces the real plugin's logic without importing it. """ if not base_url: return False normalized = base_url.lower().rstrip("/") if normalized.endswith("/anthropic"): return True # api.anthropic.com if "api.anthropic.com" in normalized: return True # kimi coding plan if "api.kimi.com" in normalized and "/coding" in normalized: return True return False def _mock_is_anthropic_compat_endpoint(provider: str, base_url: str) -> bool: """Functional mock — detects Anthropic-compat endpoints. Reproduces the real plugin's logic: named compat providers OR /anthropic URL suffix. """ _COMPAT_PROVIDERS = frozenset({"minimax", "minimax-oauth", "minimax-cn"}) if provider in _COMPAT_PROVIDERS: return True url_lower = (base_url or "").lower() return "/anthropic" in url_lower def _mock_convert_openai_images_to_anthropic(messages: list) -> list: """Functional mock — converts OpenAI image_url blocks to Anthropic image blocks.""" converted = [] for msg in messages: content = msg.get("content") if not isinstance(content, list): converted.append(msg) continue new_content = [] changed = False for block in content: if block.get("type") == "image_url": image_url_val = (block.get("image_url") or {}).get("url", "") if image_url_val.startswith("data:"): header, _, b64data = image_url_val.partition(",") media_type = "image/png" if ":" in header and ";" in header: media_type = header.split(":", 1)[1].split(";", 1)[0] new_content.append({ "type": "image", "source": { "type": "base64", "media_type": media_type, "data": b64data, }, }) else: new_content.append({ "type": "image", "source": { "type": "url", "url": image_url_val, }, }) changed = True else: new_content.append(block) converted.append({**msg, "content": new_content} if changed else msg) return converted def _mock_maybe_wrap_anthropic(client_obj, model, api_key, base_url, api_mode=None): """Functional mock for maybe_wrap_anthropic — wraps when endpoint is Anthropic-wire. Reproduces the real plugin's wrapping logic without importing it. Uses the real AnthropicAuxiliaryClient from core (no SDK dependency). """ # Already wrapped — don't double-wrap from agent.anthropic_aux import (AnthropicAuxiliaryClient, AsyncAnthropicAuxiliaryClient) if isinstance(client_obj, (AnthropicAuxiliaryClient, AsyncAnthropicAuxiliaryClient)): return client_obj # Check for other specialized adapters we should never re-dispatch try: from agent.auxiliary_client import CodexAuxiliaryClient if isinstance(client_obj, CodexAuxiliaryClient): return client_obj except ImportError: pass # Explicit non-anthropic api_mode wins over URL heuristics if api_mode and api_mode != "anthropic_messages": return client_obj should_wrap = ( api_mode == "anthropic_messages" or _mock_endpoint_speaks_anthropic_messages(base_url) ) if not should_wrap: return client_obj # Use the registry's build_anthropic_client to construct a real(ish) client from agent.plugin_registries import registries build_fn = registries.get_provider_service("anthropic", "build_anthropic_client") if build_fn is None: return client_obj try: real_client = build_fn(api_key, base_url) except Exception: return client_obj return AnthropicAuxiliaryClient( real_client, model, api_key, base_url, is_oauth=False, ) def _make_base_anthropic_namespace() -> dict: """Build a minimal anthropic service namespace with safe mock stubs. Wire-format code (build_anthropic_kwargs, convert_messages_to_anthropic, AnthropicAuxiliaryClient, etc.) has moved to core modules and is no longer looked up via the registry. Only SDK-dependent orchestration (maybe_wrap_anthropic, is_anthropic_compat_endpoint, client building, auth) still needs mock stubs here. """ mock_client = MagicMock(name="anthropic_client") mock_client.base_url = "https://api.anthropic.com/v1" mock_client.api_key = "sk-ant-mock" def _resolve_token(): """Return token from env vars if set — mimics the real resolve_anthropic_token.""" import os return (os.environ.get("ANTHROPIC_TOKEN") or os.environ.get("ANTHROPIC_API_KEY")) return { # SDK-dependent client building "build_anthropic_client": MagicMock(return_value=mock_client), "build_anthropic_bedrock_client": MagicMock(return_value=mock_client), "resolve_anthropic_token": _resolve_token, "_is_oauth_token": lambda k: bool(k) and not (k or "").startswith("sk-ant-api"), "is_claude_code_token_valid": MagicMock(return_value=False), "read_claude_code_credentials": MagicMock(return_value=None), "write_claude_code_credentials": MagicMock(), "refresh_oauth_token": MagicMock(return_value=None), "run_hermes_oauth_login_pure": MagicMock(return_value=("mock-token", None)), "_HERMES_OAUTH_FILE": MagicMock(), # Resolve / endpoint detection (still plugin-provided, still needs mocking) "maybe_wrap_anthropic": _mock_maybe_wrap_anthropic, "endpoint_speaks_anthropic_messages": _mock_endpoint_speaks_anthropic_messages, "is_anthropic_compat_endpoint": _mock_is_anthropic_compat_endpoint, "convert_openai_images_to_anthropic": _mock_convert_openai_images_to_anthropic, "ANTHROPIC_DEFAULT_BASE_URL": "https://api.anthropic.com", "_ANTHROPIC_COMPAT_PROVIDERS": frozenset(), "resolve_auxiliary_client": MagicMock(return_value=(mock_client, "claude-3-5-sonnet-20241022")), } @contextmanager def mock_anthropic_provider(**overrides): """Patch the anthropic registry namespace. Use in core tests instead of patching hermes_agent_anthropic.* directly. Usage: with mock_anthropic_provider(build_anthropic_client=my_mock): result = resolve_provider_client(...) """ from agent.plugin_registries import registries base = _make_base_anthropic_namespace() base.update(overrides) with patch.dict(registries._provider_services, {"anthropic": base}): yield base @pytest.fixture(autouse=True) def _seed_anthropic_registry(request): """Install mock anthropic namespace before each test, restore after. Skips for plugin tests — they have their own conftest that registers the real plugin, and we must NOT block them with _guarded_register. Uses patch.dict so it's guaranteed to restore even when plugin tests in other directories (which use the real plugin) run before us in the same process. Function-scoped (not session) so it re-seeds after each plugin test that overwrites the registry. Also clears _provider_resolvers["anthropic"] so a real plugin registration that leaked from another test file doesn't affect core unit tests. Also blocks _ensure_plugins_discovered() so that code paths that lazily trigger plugin loading (e.g. get_plugin_auxiliary_tasks via _resolve_task_provider_model) don't overwrite the mock namespace. """ # Skip for plugin tests — they manage the registry themselves node_path = str(request.fspath) if "/plugins/" in node_path: yield return from unittest.mock import patch from agent.plugin_registries import registries ns = _make_base_anthropic_namespace() # Guard registries.register_provider_services so that if discover_and_load() # fires during a test (e.g. via get_plugin_auxiliary_tasks in # _resolve_task_provider_model), it can't overwrite our mock anthropic # namespace. We only block "anthropic" — other providers / hooks proceed # normally so tests like test_context_engine.py still work. _orig_register = registries.register_provider_services def _guarded_register(name, services): if name == "anthropic": return # mock namespace wins — don't let the real plugin clobber it return _orig_register(name, services) _orig_resolver = registries._provider_resolvers.pop("anthropic", None) with patch.dict(registries._provider_services, {"anthropic": ns}), \ patch.object(registries, "register_provider_services", _guarded_register): yield # Restore resolver (None means "not registered", which is correct for # core unit tests; plugin tests that need the real resolver load it themselves) if _orig_resolver is not None: registries._provider_resolvers["anthropic"] = _orig_resolver else: registries._provider_resolvers.pop("anthropic", None)