hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 29 Packages Projects Releases Wiki Activity
Files
8f6d98e4c3a8aed53980db92912793a28cbf4b4c
hermes-agent/tests/test_iron_proxy.py
ethernet 92686159d1 fix(pm): integrate audited runtime and lifecycle repairs 
Prepare dependency generations before selecting them. Keep shipped tool
bytes separate from writable additions, and store facts beside their entries.
Validate proposed plugin sets before config publication. Restore the previous
config if the facts write fails.

Consolidate duplicate updater, backup, setup, and voice helpers. Repair
launcher selection, dependency consumers, download ownership, update feeds,
and native Windows process and file handling.

Verification: 206 changed/prior-failing Python files reported 4630 passed,
one failed, and 330 skipped. Fix the remaining Hindsight fixture boundary.
The final targeted rerun reported 234 passed and two skipped. The store
review regression batch reported 83 passed and one skipped. Desktop
TypeScript checks, 56 selected Electron tests, 24 release tests, and the
removed-import/compatibility guards passed.

This is an integration checkpoint, not full audit acceptance. The complete
Python suite has not run on this fixed tree. Crash-atomic plugin publication,
generation cleanup, receipt correlation, and packaged lifecycle acceptance
remain open in docs/pm-audit-status.md.
2026-09-05 22:36:48 -04:00

841 lines
27 KiB
Python
Raw Blame History

"""Hermetic tests for the iron-proxy egress integration.
Covers the pure-function surface (token mint, mapping discovery, config build,
config + mappings I/O), the binary install path (HTTP downloads + tar
extraction + checksum verification fully mocked), the subprocess lifecycle
(spawn / PID / pid_alive / stop, with subprocess.Popen mocked), and the
docker backend's egress arg builder.
Live network and the real ``iron-proxy`` binary are NEVER touched. See
``tests/test_iron_proxy_e2e.py`` (gated behind a marker) for the real-binary
smoke test.
"""
from __future__ import annotations
import io
import os
import sys
import tarfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from agent.proxy_sources import iron_proxy as ip
# ---------------------------------------------------------------------------
# Per-test isolation
# ---------------------------------------------------------------------------
@pytest.fixture
def hermes_home(tmp_path, monkeypatch):
"""Point HERMES_HOME at a temp dir so install paths don't touch the real $HOME."""
home = tmp_path / "hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
# Make sure no stale provider keys influence discovery.
for key in list(os.environ):
if key.endswith("_API_KEY"):
monkeypatch.delenv(key, raising=False)
return home
# ---------------------------------------------------------------------------
# Token mint + mapping discovery
# ---------------------------------------------------------------------------
def test_mint_proxy_token_has_prefix_and_length():
t = ip.mint_proxy_token("alpha")
assert t.startswith("alpha-")
assert len(t) >= len("alpha-") + 32
def test_management_token_path_is_single_authority(hermes_home):
"""One token path: <hermes_home>/proxy/management.token, shared by mint, reuse and readers."""
assert ip._management_token_path() == ip._proxy_state_dir_ro() / "management.token"
assert not (hermes_home / "proxy").exists()
token = ip.ensure_management_token()
assert token
p = ip._management_token_path()
assert p.is_file()
assert p.read_text(encoding="utf-8-sig").strip() == token
# 0600-style private write: reuse without minting a second token.
assert ip.ensure_management_token() == token
# Forced rotation mints a new token at the same single path.
rotated = ip.ensure_management_token(force=True)
assert rotated != token
assert ip._management_token_path().read_text(encoding="utf-8-sig").strip() == rotated
# Unknown providers (no entry in _BEARER_PROVIDERS) are skipped, not warned.
# ---------------------------------------------------------------------------
# Config / mapping serialization
# ---------------------------------------------------------------------------
def _sample_mapping(env_name: str = "OPENROUTER_API_KEY") -> ip.TokenMapping:
return ip.TokenMapping(
proxy_token=ip.mint_proxy_token("test"),
real_env_name=env_name,
upstream_hosts=("openrouter.ai", "*.openrouter.ai"),
)
def test_build_proxy_config_custom_allowed_hosts(tmp_path):
m = _sample_mapping("OPENAI_API_KEY")
cfg = ip.build_proxy_config(
mappings=[m],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
allowed_hosts=["custom-host.test"],
)
domains = cfg["transforms"][0]["config"]["domains"]
# Custom allowed_hosts wins as the base; mapping's hosts get appended.
assert "custom-host.test" in domains
assert "openrouter.ai" in domains # comes from the mapping
# ---------------------------------------------------------------------------
# Default SSRF deny list (regression: docs promise cloud metadata is denied)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Bind policy (regression: must not bind 0.0.0.0)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# audit_log file pre-creation (parameter still accepted; v0.39 doesn't
# wire it into the binary config but ensure_audit_log() still creates
# the file at 0o600 as a logrotate / monitoring sentinel)
# ---------------------------------------------------------------------------
def test_audit_log_kwarg_does_not_inject_audit_path_v039(tmp_path):
"""v0.39 of iron-proxy rejects ``log.audit_path`` (not a struct
field). build_proxy_config still accepts the audit_log kwarg for
forward compatibility but MUST NOT emit it into the rendered yaml
until the upstream binary supports it. See the kwarg's docstring
for the upgrade path."""
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
audit_log=tmp_path / "audit.log",
)
assert "audit_path" not in cfg["log"], (
"iron-proxy v0.39 has no log.audit_path field; emitting it "
"causes 'field audit_path not found in type config.Log' at "
"daemon start. ensure_audit_log() still creates the file as "
"an operator-facing logrotate target."
)
def test_load_mappings_handles_corrupt_json(hermes_home):
state = ip._proxy_state_dir()
(state / "mappings.json").write_text("{not json", encoding="utf-8")
assert ip.load_mappings() == []
# ---------------------------------------------------------------------------
# Token-preservation on re-setup (regression: clobbered live sandboxes)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Uncovered provider detection (regression: signature-auth providers bypass)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Binary discovery + lazy install
# ---------------------------------------------------------------------------
def _make_fake_tar(binary_name: str, payload: bytes = b"#!/bin/sh\necho ok\n") -> bytes:
"""Build a tar.gz with one file at the root, named ``binary_name``."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
info = tarfile.TarInfo(name=binary_name)
info.size = len(payload)
info.mode = 0o755
tf.addfile(info, io.BytesIO(payload))
return buf.getvalue()
# ── GPG release-signature verification (maxpetrusenko P1) ────────────────────
def test_verify_checksums_signature_skips_without_gpg(hermes_home, monkeypatch, tmp_path):
"""No gpg on PATH → degrade gracefully (return False), do not raise."""
monkeypatch.setattr(ip.shutil, "which", lambda name: None)
cks = tmp_path / "checksums.txt"
cks.write_text("abc iron-proxy.tar.gz\n")
assert ip._verify_checksums_signature(tmp_path, cks) is False
def test_pick_tar_member_rejects_path_traversal():
"""A malicious tar that escapes via '..' must be refused."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
info = tarfile.TarInfo(name="../iron-proxy")
info.size = 1
info.mode = 0o755
tf.addfile(info, io.BytesIO(b"x"))
buf.seek(0)
with tarfile.open(fileobj=buf, mode="r:gz") as tf:
with pytest.raises(RuntimeError, match="Could not find iron-proxy"):
ip._pick_tar_member(tf, "iron-proxy")
# ---------------------------------------------------------------------------
# Subprocess lifecycle
# ---------------------------------------------------------------------------
def test_start_proxy_idempotent_when_already_running(hermes_home, monkeypatch):
state = ip._proxy_state_dir()
pid_file = state / "iron-proxy.pid"
pid_file.write_text("12345")
monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
monkeypatch.setattr(ip, "_port_listening", lambda h, p: True)
monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "test")
# Materialize config so we get past that check (we shouldn't reach it,
# but if the idempotent path regresses we want a clean failure mode).
(state / "proxy.yaml").write_text("proxy: {}")
# Sentinel: subprocess.Popen must NOT be called.
with patch("subprocess.Popen", lambda *a, **k: pytest.fail("should not spawn")):
status = ip.start_proxy()
# Should return without spawning anything.
assert status is not None
# ---------------------------------------------------------------------------
# Docker integration
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Platform asset name resolution
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Subprocess env minimization (regression: host secrets leaked to proxy)
# ---------------------------------------------------------------------------
def test_subprocess_env_strips_unrelated_secrets(hermes_home, monkeypatch):
"""``_build_proxy_subprocess_env`` must NOT carry every host secret
over to the proxy. /proc/<pid>/environ on the proxy would otherwise
expose all of them to same-uid local processes."""
# Unrelated env vars that should NOT propagate.
monkeypatch.setenv("MY_PRIVATE_TOKEN", "should-not-leak")
monkeypatch.setenv("DATABASE_URL", "postgres://very-private")
monkeypatch.setenv("SLACK_BOT_TOKEN", "xoxb-very-secret")
# Provider keys that ARE in load_mappings should propagate.
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-real")
ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
env = ip._build_proxy_subprocess_env()
assert "MY_PRIVATE_TOKEN" not in env
assert "DATABASE_URL" not in env
assert "SLACK_BOT_TOKEN" not in env
assert env.get("OPENROUTER_API_KEY") == "sk-or-real"
# ---------------------------------------------------------------------------
# CA generation TOCTOU (regression: 0o600 only set AFTER copy)
# ---------------------------------------------------------------------------
@pytest.mark.platforms("linux")
def test_ca_key_created_with_0o600(hermes_home, monkeypatch):
"""The CA private key must NEVER exist on disk with default umask
permissions, even transiently. Fix: open with explicit mode=0o600
so the very first byte is written under tight perms."""
# ensure_ca_cert shells out to openssl; mock the subprocess.run calls
# so we don't need openssl on the test host AND don't depend on its
# output format.
def fake_run(args, **kwargs):
# First call: genrsa → -out is at args[-2]
if args[1] == "genrsa":
out = args[-2]
Path(out).write_bytes(b"-----BEGIN RSA PRIVATE KEY-----\nfake\n-----END RSA PRIVATE KEY-----\n")
elif args[1] == "req":
# Find -out path
i = args.index("-out")
Path(args[i + 1]).write_bytes(b"-----BEGIN CERTIFICATE-----\nfake\n-----END CERTIFICATE-----\n")
result = MagicMock()
result.returncode = 0
return result
monkeypatch.setattr(ip.shutil, "which", lambda name: "/usr/bin/openssl" if name == "openssl" else None)
monkeypatch.setattr(ip.subprocess, "run", fake_run)
ca_crt, ca_key = ip.ensure_ca_cert()
assert ca_key.exists()
mode = ca_key.stat().st_mode & 0o777
assert mode == 0o600, f"CA key has perms {oct(mode)}, expected 0o600"
# ---------------------------------------------------------------------------
# Audit log permissions (regression: depended on umask)
# ---------------------------------------------------------------------------
@pytest.mark.platforms("linux")
def test_ensure_audit_log_creates_with_0o600(hermes_home, tmp_path):
audit = tmp_path / "audit.log"
ip.ensure_audit_log(audit)
assert audit.exists()
mode = audit.stat().st_mode & 0o777
assert mode == 0o600
@pytest.mark.platforms("linux")
def test_ensure_audit_log_tightens_existing_perms(hermes_home, tmp_path):
audit = tmp_path / "audit.log"
audit.write_text("preexisting content\n")
os.chmod(audit, 0o644)
ip.ensure_audit_log(audit)
mode = audit.stat().st_mode & 0o777
assert mode == 0o600
# ---------------------------------------------------------------------------
# State dir hardening (regression: world-traversable on multi-user hosts)
# ---------------------------------------------------------------------------
@pytest.mark.platforms("linux")
def test_proxy_state_dir_is_0o700(hermes_home):
state = ip._proxy_state_dir()
mode = state.stat().st_mode & 0o777
assert mode == 0o700
# ---------------------------------------------------------------------------
# Mappings clobber refused when corrupt (regression: silent 403s)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# CA missing → enforce_on_docker semantics (regression: silent fail-open)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Docker env collision detection (regression: docker_env silently bypassed proxy)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# v3 round: bridge-IP parser hardening (P1 #1)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# v3: default deny-list adjacency (P2 IPv4-mapped-v6 + CGNAT)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Header-auth providers (x-api-key family) — match_headers + aliases
# ---------------------------------------------------------------------------
def test_mappings_roundtrip_preserves_headers_and_aliases(hermes_home):
m = ip.TokenMapping(
proxy_token=ip.mint_proxy_token("gemini"),
real_env_name="GEMINI_API_KEY",
upstream_hosts=("generativelanguage.googleapis.com",),
match_headers=("x-goog-api-key",),
alias_env_names=("GOOGLE_API_KEY",),
)
ip.write_mappings([m])
loaded = ip.load_mappings()
assert loaded[0].match_headers == ("x-goog-api-key",)
assert loaded[0].alias_env_names == ("GOOGLE_API_KEY",)
# ---------------------------------------------------------------------------
# Management API (hot reload)
# ---------------------------------------------------------------------------
@pytest.mark.platforms("linux")
def test_ensure_management_token_persists_and_is_stable(hermes_home):
t1 = ip.ensure_management_token()
t2 = ip.ensure_management_token()
assert t1 == t2
assert t1.startswith("hermes-mgmt-")
p = ip._proxy_state_dir() / "management.token"
assert p.exists()
assert (p.stat().st_mode & 0o777) == 0o600
def test_reload_proxy_refuses_when_not_running(hermes_home, monkeypatch):
monkeypatch.setattr(ip, "_read_pid", lambda: None)
with pytest.raises(RuntimeError, match="not running"):
ip.reload_proxy()
def test_reload_proxy_posts_bearer_to_management_endpoint(hermes_home, monkeypatch):
monkeypatch.setattr(ip, "_read_pid", lambda: 4242)
monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
monkeypatch.setattr(
ip, "_read_management_listen_from_config",
lambda config_path=None: ("127.0.0.1", 9092),
)
ip.ensure_management_token()
captured = {}
class _FakeResp:
status = 200
def __enter__(self):
return self
def __exit__(self, *a):
return False
def fake_urlopen(req, timeout=None):
captured["url"] = req.full_url
captured["method"] = req.get_method()
captured["auth"] = req.get_header("Authorization")
return _FakeResp()
import urllib.request
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
assert ip.reload_proxy() is True
assert captured["url"] == "http://127.0.0.1:9092/v1/reload"
assert captured["method"] == "POST"
token = (ip._proxy_state_dir() / "management.token").read_text().strip()
assert captured["auth"] == f"Bearer {token}"
@pytest.mark.platforms("linux")
def test_start_proxy_injects_management_key_env(hermes_home, monkeypatch):
"""When the generated config has a management listener, start_proxy
must inject the bearer key env var — v0.39 refuses to start when
api_key_env is empty."""
cfg_path = ip._proxy_state_dir() / "proxy.yaml"
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=hermes_home / "ca.crt",
ca_key=hermes_home / "ca.key",
http_listen=["127.0.0.1:9090"],
)
ip.write_proxy_config(cfg)
(hermes_home / "bin").mkdir(parents=True, exist_ok=True)
fake_bin = hermes_home / "bin" / "iron-proxy"
fake_bin.write_text("#!/bin/sh\nsleep 60\n")
fake_bin.chmod(0o755)
captured_env = {}
class _FakeProc:
pid = 99999
def poll(self):
return None
def fake_popen(cmd, **kw):
captured_env.update(kw.get("env") or {})
return _FakeProc()
monkeypatch.setattr(ip.subprocess, "Popen", fake_popen)
monkeypatch.setattr(ip, "_write_pidfile_safely", lambda pf, pid: None)
monkeypatch.setattr(ip, "_port_listening", lambda h, p: True)
monkeypatch.setattr(ip, "get_status", lambda: ip.ProxyStatus(pid=99999, listening=True))
ip.start_proxy(binary=fake_bin, config_path=cfg_path, install_if_missing=False)
assert captured_env.get(ip._MGMT_API_KEY_ENV)
assert captured_env[ip._MGMT_API_KEY_ENV].startswith("hermes-mgmt-")
# ---------------------------------------------------------------------------
# v3: _pid_proc_starttime parser (handles comm with parens, brackets)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# v3: stop_proxy SIGKILL suppression on pid recycle (P3 #5 coverage gap)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# v3: _reset_for_tests actually clears module state (P3 #1)
# ---------------------------------------------------------------------------
def test_reset_for_tests_clears_version_cache_and_nonce():
"""_reset_for_tests must clear _VERSION_CACHE and _proxy_nonce so
in-process callers don't see leakage between tests."""
ip._VERSION_CACHE["dummy"] = "v0.0.0-fake"
ip._proxy_nonce = "fake-nonce-12345"
ip._reset_for_tests()
assert ip._VERSION_CACHE == {}
assert ip._proxy_nonce is None
# ---------------------------------------------------------------------------
# v3: version cache doesn't poison on empty stdout (P2 _VERSION_CACHE bug B)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# v3: NODE_OPTIONS append-merge in docker env (arshkumarsingh #1)
# ---------------------------------------------------------------------------
def test_docker_egress_node_options_uses_sentinel(hermes_home, monkeypatch):
"""``_egress_proxy_args_for_docker`` should NOT put NODE_OPTIONS in
env_overrides directly; it uses a sentinel key
``_HERMES_EGRESS_NODE_OPTIONS_APPEND`` so DockerEnvironment can
append-merge with the operator's existing NODE_OPTIONS."""
from tools.environments.docker import _egress_proxy_args_for_docker
from hermes_cli.config import load_config, save_config
state = ip._proxy_state_dir()
ca = state / "ca.crt"
ca.write_text("fake-ca")
(state / "ca.key").write_text("fake-key")
mapping = _sample_mapping("OPENROUTER_API_KEY")
proxy_cfg = ip.build_proxy_config(
mappings=[mapping], ca_cert=ca, ca_key=state / "ca.key", tunnel_port=9090,
)
ip.write_proxy_config(proxy_cfg)
ip.write_mappings([mapping])
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = True
cfg["proxy"]["enforce_on_docker"] = True
save_config(cfg)
(state / "iron-proxy.pid").write_text("99999")
monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
monkeypatch.setattr(ip, "_port_listening", lambda h, p: True)
_, env, _ = _egress_proxy_args_for_docker()
# The egress dict should contain the sentinel, NOT a raw NODE_OPTIONS.
assert env.get("_HERMES_EGRESS_NODE_OPTIONS_APPEND") == "--use-openssl-ca"
assert "NODE_OPTIONS" not in env, (
"NODE_OPTIONS in egress env_overrides would clobber the operator's "
"docker_env NODE_OPTIONS — that's exactly the bug arshkumarsingh "
"flagged."
)
# ---------------------------------------------------------------------------
# v3: ensure_audit_log fails loud on OSError (P2 promise mismatch)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# v3: persisted nonce roundtrip (stephenschoettler #3 cross-CLI defense)
# ---------------------------------------------------------------------------
def test_persisted_nonce_roundtrip(hermes_home, monkeypatch):
"""Write the nonce next to the pidfile (simulating one CLI invocation
finishing start_proxy), then verify a fresh _read_persisted_nonce
can pick it up — that's what cross-process _pid_alive uses."""
nonce_path = ip._persisted_nonce_path()
nonce_path.parent.mkdir(parents=True, exist_ok=True)
nonce_path.write_text("test-nonce-abc123")
assert ip._read_persisted_nonce() == "test-nonce-abc123"
# ---------------------------------------------------------------------------
# v4 round (GodsBoy follow-up): bind-host-aware liveness probes +
# allow_env_fallback on the partial-secret path
# ---------------------------------------------------------------------------
def test_get_status_probes_configured_bind_host(hermes_home, monkeypatch):
"""get_status must probe the configured bind host (e.g. the docker
bridge IP), not loopback unconditionally."""
state = ip._proxy_state_dir()
(state / "proxy.yaml").write_text(
"proxy:\n http_listen: 172.17.0.1:9123\n", encoding="utf-8"
)
(state / "ca.crt").write_text("cert")
ip._write_pidfile_safely(ip._pidfile(), 99999)
monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
monkeypatch.setattr(ip, "find_iron_proxy", lambda **kw: None)
probed = {}
def fake_probe(host, port):
probed["host"] = host
probed["port"] = port
return True
monkeypatch.setattr(ip, "_port_listening", fake_probe)
status = ip.get_status()
assert probed == {"host": "172.17.0.1", "port": 9123}
assert status.listening is True
assert status.tunnel_port == 9123
def test_partial_bitwarden_secrets_honor_allow_env_fallback(
hermes_home, monkeypatch,
):
"""The missing-secret branch's own error message tells operators to
set proxy.allow_env_fallback — so the flag must actually work there
(previously only the empty-token branch honored it)."""
ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-host-fallback")
import agent.secret_sources.bitwarden as bw
monkeypatch.setattr(
bw, "fetch_bitwarden_secrets", lambda **kw: ({}, []),
)
monkeypatch.setenv("BWS_ACCESS_TOKEN", "tok")
bw_cfg = {
"project_id": "proj",
"access_token_env": "BWS_ACCESS_TOKEN",
"allow_env_fallback": True,
}
env = ip._build_proxy_subprocess_env(
refresh_from_bitwarden=True, bitwarden_config=bw_cfg,
)
# Falls back to the host env value instead of raising.
assert env.get("OPENROUTER_API_KEY") == "sk-host-fallback"
def test_partial_bitwarden_secrets_raise_without_fallback(
hermes_home, monkeypatch,
):
"""Strict default: missing BWS secrets raise."""
ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-host")
import agent.secret_sources.bitwarden as bw
monkeypatch.setattr(
bw, "fetch_bitwarden_secrets", lambda **kw: ({}, []),
)
monkeypatch.setenv("BWS_ACCESS_TOKEN", "tok")
bw_cfg = {"project_id": "proj", "access_token_env": "BWS_ACCESS_TOKEN"}
with pytest.raises(RuntimeError, match="did not return secrets"):
ip._build_proxy_subprocess_env(
refresh_from_bitwarden=True, bitwarden_config=bw_cfg,
)
def test_bitwarden_importerror_raise_without_fallback(
hermes_home, monkeypatch,
):
"""Strict default: ImportError on BWS module raises when
allow_env_fallback is unset, matching the sibling branches."""
ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-host")
# Simulate the BWS SDK not being installed. The lazy import
# ``from agent.secret_sources import bitwarden`` inside
# _build_proxy_subprocess_env resolves through the parent package's
# cached attribute; deleting both the sys.modules entry AND the
# parent-package attribute forces a real import that we intercept.
#
# In addition, block importlib.reload in case the test infra used it.
import agent.secret_sources as ss
monkeypatch.delitem(sys.modules, "agent.secret_sources.bitwarden", raising=False)
monkeypatch.delitem(sys.modules, "agent.secret_sources.bitwarden.bws", raising=False)
monkeypatch.delattr(ss, "bitwarden", raising=False)
# Now block the re-import. ``from agent.secret_sources import
# bitwarden`` resolves to a submodule attribute; setting it to a
# sentinel that raises on attribute access is more reliable than
# trying to intercept __import__ at the C level.
class _MissingBWS:
"""Sentinel: accessing any attribute raises ImportError."""
def __getattr__(self, _name):
raise ImportError("bws SDK not installed")
def __call__(self, *a, **kw):
raise ImportError("bws SDK not installed")
monkeypatch.setattr(ss, "bitwarden", _MissingBWS(), raising=False)
monkeypatch.setenv("BWS_ACCESS_TOKEN", "tok")
bw_cfg = {"project_id": "proj", "access_token_env": "BWS_ACCESS_TOKEN"}
with pytest.raises(RuntimeError, match="Bitwarden refresh module unavailable"):
ip._build_proxy_subprocess_env(
refresh_from_bitwarden=True, bitwarden_config=bw_cfg,
)
Reference in New Issue View Git Blame Copy Permalink