mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6
hermes-agent/tests/test_iron_proxy.py
Teknium 6b81590c55
test: prune low-value tests suite-wide (wave 1) — 46,820 → 28,106 test functions 
Systematic prune per AGENTS.md test policy, one pass over every major
test tree (gateway, hermes_cli, tools, agent, run_agent, plugins, cli,
cron, tui_gateway, honcho/openviking, root-level):

- DELETE: source-reading tests (read_text/getsource on prod files),
  change-detector tests (exact catalog counts, model-name snapshots,
  config version literals), mock-echo tests (assert a mock returns what
  it was told), assertion-free/trivial tests, near-duplicate
  parametrizations (boundaries + one representative kept), async/sync
  twin duplicates, cosmetic within-file variations.
- KEEP (mandatory): security/redaction/approval guards, message-role
  alternation invariants, prompt-caching/deterministic-call-id
  invariants, issue-number regression tests (deduped), E2E tests.
- 6 test files deleted outright (script-style/no-assert or fully
  redundant); conftest.py, fakes/, fixtures/ untouched.
- tests/acp/conftest.py added: autouse fixture stubs the live
  models.dev/GitHub/Copilot/Anthropic inventory fetches that ACP server
  tests performed on every session create — test_server.py 147s → 3.4s,
  and the tests are now genuinely hermetic.
- Sleep-based slowness shrunk where safe (codex_ttfb_watchdog,
  compression_concurrent_fork, etc.); no wall-clock assertion tightened.

Verification: full hermetic suite via scripts/run_tests.sh —
2439 files, 31,130 tests passed, 0 failed, 0 flaky retries, 315s wall
(baseline: 583s wall, 13,564s subprocess CPU).
2026-07-29 13:10:23 -07:00

816 lines
26 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
# 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)
# ---------------------------------------------------------------------------
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)
# ---------------------------------------------------------------------------
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
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)
# ---------------------------------------------------------------------------
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)
# ---------------------------------------------------------------------------
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}"
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 a new issue View git blame Copy permalink