fix(auth): hoist get_env_value import + strengthen .env fallback tests

Follow-up to cherry-picked PR #15920:

- agent/credential_pool.py: hoist 'from hermes_cli.config import get_env_value'
  to module top instead of inline try/except in each seed site (3 sites).
  No import cycle — hermes_cli/config.py doesn't depend on agent.credential_pool.
- hermes_cli/auth.py: same hoist for the _resolve_api_key_provider_secret loop.
- tests/tools/test_credential_pool_env_fallback.py: replace smoke-only tests
  with real .env file I/O. Each test writes a temp ~/.hermes/.env, verifies
  _seed_from_env / _resolve_api_key_provider_secret read from it, and asserts
  the full priority chain: os.environ > .env > credential_pool. Uses
  'deepseek' as the test provider since 'openai' isn't in PROVIDER_REGISTRY
  and _seed_from_env's generic path requires a real pconfig lookup.
This commit is contained in:
Teknium
2026-04-26 08:30:56 -07:00
committed by Teknium
parent 27f4dba5ce
commit f2d655529a
3 changed files with 184 additions and 98 deletions

View File

@@ -14,6 +14,7 @@ from datetime import datetime
from typing import Any, Dict, List, Optional, Set, Tuple from typing import Any, Dict, List, Optional, Set, Tuple
from hermes_constants import OPENROUTER_BASE_URL from hermes_constants import OPENROUTER_BASE_URL
from hermes_cli.config import get_env_value
import hermes_cli.auth as auth_mod import hermes_cli.auth as auth_mod
from hermes_cli.auth import ( from hermes_cli.auth import (
CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS, CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
@@ -1274,11 +1275,7 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
return False return False
if provider == "openrouter": if provider == "openrouter":
# Check both os.environ and ~/.hermes/.env file # Check both os.environ and ~/.hermes/.env file
try: token = (get_env_value("OPENROUTER_API_KEY") or "").strip()
from hermes_cli.config import get_env_value
token = (get_env_value("OPENROUTER_API_KEY") or "").strip()
except Exception:
token = os.getenv("OPENROUTER_API_KEY", "").strip()
if token: if token:
source = "env:OPENROUTER_API_KEY" source = "env:OPENROUTER_API_KEY"
if _is_source_suppressed(provider, source): if _is_source_suppressed(provider, source):
@@ -1304,11 +1301,7 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
env_url = "" env_url = ""
if pconfig.base_url_env_var: if pconfig.base_url_env_var:
try: env_url = (get_env_value(pconfig.base_url_env_var) or "").strip().rstrip("/")
from hermes_cli.config import get_env_value
env_url = (get_env_value(pconfig.base_url_env_var) or "").strip().rstrip("/")
except Exception:
env_url = os.getenv(pconfig.base_url_env_var, "").strip().rstrip("/")
env_vars = list(pconfig.api_key_env_vars) env_vars = list(pconfig.api_key_env_vars)
if provider == "anthropic": if provider == "anthropic":
@@ -1320,11 +1313,7 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
for env_var in env_vars: for env_var in env_vars:
# Check both os.environ and ~/.hermes/.env file # Check both os.environ and ~/.hermes/.env file
try: token = (get_env_value(env_var) or "").strip()
from hermes_cli.config import get_env_value
token = (get_env_value(env_var) or "").strip()
except Exception:
token = os.getenv(env_var, "").strip()
if not token: if not token:
continue continue
source = f"env:{env_var}" source = f"env:{env_var}"

View File

@@ -467,13 +467,10 @@ def _resolve_api_key_provider_secret(
pass pass
return "", "" return "", ""
from hermes_cli.config import get_env_value
for env_var in pconfig.api_key_env_vars: for env_var in pconfig.api_key_env_vars:
# Check both os.environ and ~/.hermes/.env file # Check both os.environ and ~/.hermes/.env file
try: val = (get_env_value(env_var) or "").strip()
from hermes_cli.config import get_env_value
val = (get_env_value(env_var) or "").strip()
except Exception:
val = os.getenv(env_var, "").strip()
if has_usable_secret(val): if has_usable_secret(val):
return val, env_var return val, env_var

View File

@@ -1,110 +1,210 @@
"""Tests for credential_pool .env fallback and auth credential pool lookup.""" """Tests for credential_pool .env fallback and auth credential_pool lookup.
Covers the fix from #15914 / PR #15920:
- _seed_from_env reads API keys from ~/.hermes/.env when not in os.environ
- _resolve_api_key_provider_secret falls back to credential_pool when env vars are empty
- env vars take priority over .env file (handled by get_env_value itself)
- env vars take priority over credential pool (fallback only kicks in when env is empty)
"""
import os import os
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest import pytest
from unittest.mock import patch, MagicMock
def _make_pconfig(env_vars=None): def _make_pconfig(provider_id="deepseek", env_vars=None):
"""Create a minimal ProviderConfig for testing.""" """Create a minimal ProviderConfig for testing.
Default provider_id is 'deepseek' because it's a real api_key provider
in PROVIDER_REGISTRY (needed for _seed_from_env's generic path).
"""
from hermes_cli.auth import ProviderConfig from hermes_cli.auth import ProviderConfig
return ProviderConfig( return ProviderConfig(
id="openai", id=provider_id,
name="OpenAI", name=provider_id.title(),
auth_type="api_key", auth_type="api_key",
api_key_env_vars=tuple(env_vars or ["OPENAI_API_KEY"]), api_key_env_vars=tuple(env_vars or [f"{provider_id.upper()}_API_KEY"]),
) )
class TestCredentialPoolEnvFallback: @pytest.fixture
"""Verify _seed_from_env resolves keys from both os.environ and .env file.""" def isolated_hermes_home(tmp_path, monkeypatch):
"""Point HERMES_HOME at a temp dir and clear known API key env vars.
Also invalidates any cached get_env_value state by patching Path.home().
"""
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setattr(Path, "home", lambda: tmp_path)
monkeypatch.setenv("HERMES_HOME", str(home))
# Clear all known API key env vars so get_env_value falls through to .env
for key in [
"OPENAI_API_KEY", "ANTHROPIC_API_KEY", "OPENROUTER_API_KEY",
"ZAI_API_KEY", "DEEPSEEK_API_KEY", "ANTHROPIC_TOKEN",
"CLAUDE_CODE_OAUTH_TOKEN", "OPENAI_BASE_URL",
]:
monkeypatch.delenv(key, raising=False)
return home
def _write_env_file(home: Path, **kwargs) -> None:
"""Write key=value pairs to ~/.hermes/.env."""
lines = [f"{k}={v}" for k, v in kwargs.items()]
(home / ".env").write_text("\n".join(lines) + "\n")
class TestCredentialPoolSeedsFromDotEnv:
"""_seed_from_env must read keys from ~/.hermes/.env, not just os.environ.
This is the load-bearing behaviour for the fix: when a user adds a key to
.env mid-session or via a non-CLI entry point that doesn't run
load_hermes_dotenv, the credential pool must still discover it.
"""
def test_deepseek_key_from_dotenv_only(self, isolated_hermes_home):
"""Key in .env but not os.environ → _seed_from_env adds a pool entry."""
_write_env_file(isolated_hermes_home, DEEPSEEK_API_KEY="sk-dotenv-only-12345")
assert "DEEPSEEK_API_KEY" not in os.environ
def test_os_environ_still_works(self):
"""Existing os.environ resolution must not break.
_seed_from_env only collects env var names, does not return found=True
for existing keys — that is _resolve's job. Just verify no crash."""
from agent.credential_pool import _seed_from_env from agent.credential_pool import _seed_from_env
# Should not raise entries = []
found, entries = _seed_from_env("openai", []) changed, active_sources = _seed_from_env("deepseek", entries)
def test_get_env_value_import_does_not_crash(self): assert changed is True
"""Importing get_env_value from hermes_cli.config should not raise.""" assert "env:DEEPSEEK_API_KEY" in active_sources
try: assert any(
from hermes_cli.config import get_env_value e.access_token == "sk-dotenv-only-12345"
assert callable(get_env_value) and e.source == "env:DEEPSEEK_API_KEY"
except ImportError: for e in entries
pytest.skip("hermes_cli.config not available in test environment") ), f"Expected seeded entry with dotenv key, got: {[(e.source, e.access_token) for e in entries]}"
def test_openrouter_key_from_dotenv_only(self, isolated_hermes_home):
"""OpenRouter path has its own branch — verify it also reads .env."""
_write_env_file(isolated_hermes_home, OPENROUTER_API_KEY="sk-or-dotenv-abc")
assert "OPENROUTER_API_KEY" not in os.environ
from agent.credential_pool import _seed_from_env
entries = []
changed, active_sources = _seed_from_env("openrouter", entries)
assert changed is True
assert "env:OPENROUTER_API_KEY" in active_sources
assert any(
e.access_token == "sk-or-dotenv-abc" for e in entries
)
def test_empty_dotenv_no_entries(self, isolated_hermes_home):
"""No .env file, no env vars → no entries seeded (and no crash)."""
from agent.credential_pool import _seed_from_env
entries = []
changed, active_sources = _seed_from_env("deepseek", entries)
assert changed is False
assert active_sources == set()
assert entries == []
def test_os_environ_still_wins_over_dotenv(self, isolated_hermes_home, monkeypatch):
"""get_env_value checks os.environ first — verify seeding picks that up."""
_write_env_file(isolated_hermes_home, DEEPSEEK_API_KEY="sk-dotenv-stale")
monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-env-fresh-xyz")
from agent.credential_pool import _seed_from_env
entries = []
changed, _ = _seed_from_env("deepseek", entries)
assert changed is True
seeded = [e for e in entries if e.source == "env:DEEPSEEK_API_KEY"]
assert len(seeded) == 1
assert seeded[0].access_token == "sk-env-fresh-xyz"
class TestAuthResolvesFromDotEnv:
"""_resolve_api_key_provider_secret must also read from ~/.hermes/.env."""
def test_key_from_dotenv_only(self, isolated_hermes_home):
"""Key in .env but not os.environ → _resolve returns it with the env var source."""
_write_env_file(isolated_hermes_home, DEEPSEEK_API_KEY="sk-dotenv-resolve-789")
assert "DEEPSEEK_API_KEY" not in os.environ
from hermes_cli.auth import _resolve_api_key_provider_secret
key, source = _resolve_api_key_provider_secret(
provider_id="deepseek",
pconfig=_make_pconfig(),
)
assert key == "sk-dotenv-resolve-789"
assert source == "DEEPSEEK_API_KEY"
class TestAuthCredentialPoolFallback: class TestAuthCredentialPoolFallback:
"""Verify auth.py falls back to credential pool when env vars are empty.""" """_resolve_api_key_provider_secret falls back to credential pool when env + dotenv are empty."""
def _clear_api_keys(self): def test_credential_pool_fallback_structure(self, isolated_hermes_home):
"""Temporarily clear API key env vars, return backup dict.""" """Empty env + empty .env → auth falls back to credential pool."""
backup = {}
for key in ["OPENAI_API_KEY", "ANTHROPIC_API_KEY", "OPENROUTER_API_KEY",
"ZAI_API_KEY", "DEEPSEEK_API_KEY"]:
if key in os.environ:
backup[key] = os.environ.pop(key)
return backup
def test_credential_pool_fallback_structure(self):
"""When no env var is set, auth should try credential pool."""
from hermes_cli.auth import _resolve_api_key_provider_secret
mock_entry = MagicMock() mock_entry = MagicMock()
mock_entry.access_token = "test-pool-key-12345" mock_entry.access_token = "test-pool-key-12345"
mock_entry.runtime_api_key = "" mock_entry.runtime_api_key = ""
mock_pool = MagicMock() mock_pool = MagicMock()
mock_pool.has_credentials.return_value = True mock_pool.has_credentials.return_value = True
mock_pool.peek.return_value = mock_entry mock_pool.peek.return_value = mock_entry
backup = self._clear_api_keys()
try:
with patch("agent.credential_pool.load_pool", return_value=mock_pool):
key, source = _resolve_api_key_provider_secret(
provider_id="openai",
pconfig=_make_pconfig(),
)
assert "test-pool-key-12345" in key
assert "credential_pool" in source
finally:
os.environ.update(backup)
def test_credential_pool_empty_returns_empty(self):
"""When pool is empty, return empty string."""
from hermes_cli.auth import _resolve_api_key_provider_secret from hermes_cli.auth import _resolve_api_key_provider_secret
with patch("agent.credential_pool.load_pool", return_value=mock_pool):
key, source = _resolve_api_key_provider_secret(
provider_id="deepseek",
pconfig=_make_pconfig(),
)
assert "test-pool-key-12345" in key
assert "credential_pool" in source
def test_credential_pool_empty_returns_empty(self, isolated_hermes_home):
"""Empty env + empty .env + empty pool → empty string."""
mock_pool = MagicMock() mock_pool = MagicMock()
mock_pool.has_credentials.return_value = False mock_pool.has_credentials.return_value = False
backup = self._clear_api_keys()
try:
with patch("agent.credential_pool.load_pool", return_value=mock_pool):
key, source = _resolve_api_key_provider_secret(
provider_id="openai",
pconfig=_make_pconfig(),
)
assert key == ""
finally:
os.environ.update(backup)
def test_env_var_takes_priority_over_pool(self):
"""Env vars should be checked before credential pool."""
from hermes_cli.auth import _resolve_api_key_provider_secret from hermes_cli.auth import _resolve_api_key_provider_secret
with patch("agent.credential_pool.load_pool", return_value=mock_pool):
key, source = _resolve_api_key_provider_secret(
provider_id="deepseek",
pconfig=_make_pconfig(),
)
assert key == ""
def test_env_var_takes_priority_over_pool(self, isolated_hermes_home, monkeypatch):
"""os.environ key wins — credential pool is NEVER consulted."""
monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-env-key-first-abc123")
mock_pool = MagicMock() mock_pool = MagicMock()
mock_pool.has_credentials.return_value = True mock_pool.has_credentials.return_value = True
with patch.dict(os.environ, {"OPENAI_API_KEY": "sk-env-key-first-abc123"}): from hermes_cli.auth import _resolve_api_key_provider_secret
with patch("agent.credential_pool.load_pool", return_value=mock_pool): with patch("agent.credential_pool.load_pool", return_value=mock_pool) as mp:
key, source = _resolve_api_key_provider_secret( key, source = _resolve_api_key_provider_secret(
provider_id="openai", provider_id="deepseek",
pconfig=_make_pconfig(), pconfig=_make_pconfig(),
) )
assert key == "sk-env-key-first-abc123" assert key == "sk-env-key-first-abc123"
# Source is the env var name itself (e.g. "OPENAI_API_KEY") assert source == "DEEPSEEK_API_KEY"
assert "OPENAI_API_KEY" in source # Pool should not even have been loaded — env var satisfied the request first
# Pool peek should NOT have been called — env var found first mp.assert_not_called()
mock_pool.peek.assert_not_called()
def test_dotenv_takes_priority_over_pool(self, isolated_hermes_home):
"""Key in .env beats credential pool — pool only fires when both env sources are empty."""
_write_env_file(isolated_hermes_home, DEEPSEEK_API_KEY="sk-dotenv-priority-xyz")
assert "DEEPSEEK_API_KEY" not in os.environ
mock_pool = MagicMock()
mock_pool.has_credentials.return_value = True
from hermes_cli.auth import _resolve_api_key_provider_secret
with patch("agent.credential_pool.load_pool", return_value=mock_pool) as mp:
key, source = _resolve_api_key_provider_secret(
provider_id="deepseek",
pconfig=_make_pconfig(),
)
assert key == "sk-dotenv-priority-xyz"
assert source == "DEEPSEEK_API_KEY"
mp.assert_not_called()