Merge branch 'main' of github.com:NousResearch/hermes-agent into bb/gui
# Conflicts: # cli.py # hermes_cli/main.py # run_agent.py # tests/hermes_cli/test_cmd_update.py # tools/mcp_tool.py # web/src/lib/gatewayClient.ts
This commit is contained in:
Brooklyn Nicholson
@@ -107,7 +107,7 @@ def test_auth_add_nous_oauth_persists_pool_entry(tmp_path, monkeypatch):
|
||||
"portal_base_url": "https://portal.example.com",
|
||||
"inference_base_url": "https://inference.example.com/v1",
|
||||
"client_id": "hermes-cli",
|
||||
"scope": "inference:mint_agent_key",
|
||||
"scope": "inference:invoke inference:mint_agent_key",
|
||||
"token_type": "Bearer",
|
||||
"access_token": token,
|
||||
"refresh_token": "refresh-token",
|
||||
@@ -228,7 +228,7 @@ def test_auth_add_nous_oauth_honors_custom_label(tmp_path, monkeypatch):
|
||||
"portal_base_url": "https://portal.example.com",
|
||||
"inference_base_url": "https://inference.example.com/v1",
|
||||
"client_id": "hermes-cli",
|
||||
"scope": "inference:mint_agent_key",
|
||||
"scope": "inference:invoke inference:mint_agent_key",
|
||||
"token_type": "Bearer",
|
||||
"access_token": token,
|
||||
"refresh_token": "refresh-token",
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
"""Regression tests for Nous OAuth refresh + agent-key mint interactions."""
|
||||
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
@@ -125,6 +128,11 @@ def _setup_nous_auth(
|
||||
*,
|
||||
access_token: str = "access-old",
|
||||
refresh_token: str = "refresh-old",
|
||||
scope: str = "inference:mint_agent_key",
|
||||
expires_at: str = "2026-02-01T00:00:00+00:00",
|
||||
expires_in: int = 0,
|
||||
agent_key: str | None = None,
|
||||
agent_key_expires_at: str | None = None,
|
||||
) -> None:
|
||||
hermes_home.mkdir(parents=True, exist_ok=True)
|
||||
auth_store = {
|
||||
@@ -136,15 +144,15 @@ def _setup_nous_auth(
|
||||
"inference_base_url": "https://inference.example.com/v1",
|
||||
"client_id": "hermes-cli",
|
||||
"token_type": "Bearer",
|
||||
"scope": "inference:mint_agent_key",
|
||||
"scope": scope,
|
||||
"access_token": access_token,
|
||||
"refresh_token": refresh_token,
|
||||
"obtained_at": "2026-02-01T00:00:00+00:00",
|
||||
"expires_in": 0,
|
||||
"expires_at": "2026-02-01T00:00:00+00:00",
|
||||
"agent_key": None,
|
||||
"expires_in": expires_in,
|
||||
"expires_at": expires_at,
|
||||
"agent_key": agent_key,
|
||||
"agent_key_id": None,
|
||||
"agent_key_expires_at": None,
|
||||
"agent_key_expires_at": agent_key_expires_at,
|
||||
"agent_key_expires_in": None,
|
||||
"agent_key_reused": None,
|
||||
"agent_key_obtained_at": None,
|
||||
@@ -164,6 +172,463 @@ def _mint_payload(api_key: str = "agent-key") -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _jwt_with_claims(claims: dict) -> str:
|
||||
def _part(payload: dict) -> str:
|
||||
raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
||||
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
|
||||
|
||||
return f"{_part({'alg': 'none', 'typ': 'JWT'})}.{_part(claims)}.sig"
|
||||
|
||||
|
||||
def _future_iso(seconds: int = 3600) -> str:
|
||||
return datetime.fromtimestamp(time.time() + seconds, tz=timezone.utc).isoformat()
|
||||
|
||||
|
||||
def _invoke_jwt(*, seconds: int = 3600, scope: object = "inference:invoke inference:mint_agent_key") -> str:
|
||||
return _jwt_with_claims({
|
||||
"sub": "test-user",
|
||||
"scope": scope,
|
||||
"exp": int(time.time() + seconds),
|
||||
})
|
||||
|
||||
|
||||
def test_resolve_nous_runtime_credentials_prefers_invoke_jwt_and_mirrors(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
token = _invoke_jwt(seconds=3600)
|
||||
_setup_nous_auth(
|
||||
hermes_home,
|
||||
access_token=token,
|
||||
scope=auth_mod.DEFAULT_NOUS_SCOPE,
|
||||
expires_at=_future_iso(3600),
|
||||
expires_in=3600,
|
||||
)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
def _unexpected_mint(*args, **kwargs):
|
||||
raise AssertionError("legacy agent-key mint should not run for invoke JWT")
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _unexpected_mint)
|
||||
|
||||
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
|
||||
|
||||
assert creds["api_key"] == token
|
||||
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
|
||||
assert creds["auth_path"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
|
||||
|
||||
payload = json.loads((hermes_home / "auth.json").read_text())
|
||||
singleton = payload["providers"]["nous"]
|
||||
assert singleton["agent_key"] == token
|
||||
assert datetime.fromisoformat(singleton["agent_key_expires_at"]).timestamp() > time.time() + 300
|
||||
|
||||
pool_entries = payload["credential_pool"]["nous"]
|
||||
assert len(pool_entries) == 1
|
||||
assert pool_entries[0]["agent_key"] == token
|
||||
assert pool_entries[0]["source"] == auth_mod.NOUS_DEVICE_CODE_SOURCE
|
||||
|
||||
|
||||
def test_resolve_nous_runtime_credentials_invoke_jwt_is_idempotent(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
hermes_home.mkdir(parents=True, exist_ok=True)
|
||||
exp = int(time.time() + 3600)
|
||||
expires_at = datetime.fromtimestamp(exp, tz=timezone.utc).isoformat()
|
||||
token = _jwt_with_claims({
|
||||
"sub": "test-user",
|
||||
"scope": auth_mod.DEFAULT_NOUS_SCOPE,
|
||||
"exp": exp,
|
||||
})
|
||||
original_obtained_at = "2026-04-17T22:00:10+00:00"
|
||||
auth_store = {
|
||||
"version": 1,
|
||||
"active_provider": "nous",
|
||||
"providers": {
|
||||
"nous": {
|
||||
"portal_base_url": "https://portal.example.com",
|
||||
"inference_base_url": "https://inference.example.com/v1",
|
||||
"client_id": "hermes-cli",
|
||||
"token_type": "Bearer",
|
||||
"scope": auth_mod.DEFAULT_NOUS_SCOPE,
|
||||
"access_token": token,
|
||||
"refresh_token": "refresh-token",
|
||||
"obtained_at": "2026-02-01T00:00:00+00:00",
|
||||
"expires_in": 123,
|
||||
"expires_at": expires_at,
|
||||
"agent_key": token,
|
||||
"agent_key_id": None,
|
||||
"agent_key_expires_at": expires_at,
|
||||
"agent_key_expires_in": 123,
|
||||
"agent_key_reused": False,
|
||||
"agent_key_obtained_at": original_obtained_at,
|
||||
"tls": {"insecure": False, "ca_bundle": None},
|
||||
},
|
||||
},
|
||||
}
|
||||
auth_path = hermes_home / "auth.json"
|
||||
auth_path.write_text(json.dumps(auth_store, indent=2))
|
||||
before_content = auth_path.read_text()
|
||||
before_mtime = auth_path.stat().st_mtime_ns
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
def _unexpected_mint(*args, **kwargs):
|
||||
raise AssertionError("stable invoke JWT should not mint a legacy key")
|
||||
|
||||
def _unexpected_shared_write(*args, **kwargs):
|
||||
raise AssertionError("unchanged invoke JWT resolution should not sync shared store")
|
||||
|
||||
sync_calls = []
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _unexpected_mint)
|
||||
monkeypatch.setattr(auth_mod, "_write_shared_nous_state", _unexpected_shared_write)
|
||||
monkeypatch.setattr(
|
||||
auth_mod,
|
||||
"_sync_nous_pool_from_auth_store",
|
||||
lambda: sync_calls.append(True),
|
||||
)
|
||||
|
||||
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
|
||||
|
||||
assert creds["api_key"] == token
|
||||
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
|
||||
assert auth_path.read_text() == before_content
|
||||
assert auth_path.stat().st_mtime_ns == before_mtime
|
||||
assert sync_calls == []
|
||||
payload = json.loads(auth_path.read_text())
|
||||
assert (
|
||||
payload["providers"]["nous"]["agent_key_obtained_at"]
|
||||
== original_obtained_at
|
||||
)
|
||||
|
||||
|
||||
def test_resolve_nous_runtime_credentials_trusts_invoke_jwt_exp_over_stale_metadata(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
token = _invoke_jwt(seconds=3600)
|
||||
_setup_nous_auth(
|
||||
hermes_home,
|
||||
access_token=token,
|
||||
scope=auth_mod.DEFAULT_NOUS_SCOPE,
|
||||
expires_at="2000-01-01T00:00:00+00:00",
|
||||
expires_in=0,
|
||||
agent_key=token,
|
||||
agent_key_expires_at="2000-01-01T00:00:00+00:00",
|
||||
)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
def _unexpected_refresh(*args, **kwargs):
|
||||
raise AssertionError("valid invoke JWT should not be refreshed because metadata is stale")
|
||||
|
||||
def _unexpected_mint(*args, **kwargs):
|
||||
raise AssertionError("valid invoke JWT should not fall back to legacy mint")
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_refresh_access_token", _unexpected_refresh)
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _unexpected_mint)
|
||||
|
||||
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
|
||||
|
||||
assert creds["api_key"] == token
|
||||
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
|
||||
payload = json.loads((hermes_home / "auth.json").read_text())
|
||||
singleton = payload["providers"]["nous"]
|
||||
assert singleton["agent_key"] == token
|
||||
assert datetime.fromisoformat(singleton["expires_at"]).timestamp() > time.time() + 300
|
||||
assert datetime.fromisoformat(singleton["agent_key_expires_at"]).timestamp() > time.time() + 300
|
||||
|
||||
|
||||
def test_resolve_nous_runtime_credentials_does_not_apply_legacy_ttl_to_invoke_jwt(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
token = _invoke_jwt(seconds=900)
|
||||
_setup_nous_auth(
|
||||
hermes_home,
|
||||
access_token=token,
|
||||
scope=auth_mod.DEFAULT_NOUS_SCOPE,
|
||||
expires_at=_future_iso(900),
|
||||
expires_in=900,
|
||||
)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
def _unexpected_mint(*args, **kwargs):
|
||||
raise AssertionError("1800s legacy min TTL should not force opaque mint for invoke JWT")
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _unexpected_mint)
|
||||
|
||||
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=1800)
|
||||
|
||||
assert creds["api_key"] == token
|
||||
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
|
||||
payload = json.loads((hermes_home / "auth.json").read_text())
|
||||
assert payload["providers"]["nous"]["agent_key"] == token
|
||||
assert payload["credential_pool"]["nous"][0]["agent_key"] == token
|
||||
|
||||
|
||||
def test_legacy_auth_mode_bypasses_usable_invoke_jwt(tmp_path, monkeypatch):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
token = _invoke_jwt(seconds=3600)
|
||||
_setup_nous_auth(
|
||||
hermes_home,
|
||||
access_token=token,
|
||||
scope=auth_mod.DEFAULT_NOUS_SCOPE,
|
||||
expires_at=_future_iso(3600),
|
||||
expires_in=3600,
|
||||
)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
mint_calls = []
|
||||
|
||||
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
|
||||
del client, portal_base_url, min_ttl_seconds
|
||||
mint_calls.append(access_token)
|
||||
return _mint_payload(api_key="legacy-after-jwt-401")
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
|
||||
|
||||
creds = auth_mod.resolve_nous_runtime_credentials(
|
||||
min_key_ttl_seconds=300,
|
||||
inference_auth_mode=auth_mod.NOUS_INFERENCE_AUTH_MODE_LEGACY,
|
||||
)
|
||||
|
||||
assert mint_calls == [token]
|
||||
assert creds["api_key"] == "legacy-after-jwt-401"
|
||||
assert creds["auth_path"] == auth_mod.NOUS_AUTH_PATH_LEGACY_SESSION_KEY_MINT
|
||||
payload = json.loads((hermes_home / "auth.json").read_text())
|
||||
assert payload["providers"]["nous"]["agent_key"] == "legacy-after-jwt-401"
|
||||
|
||||
|
||||
def test_resolve_nous_runtime_credentials_falls_back_when_invoke_scope_missing(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
token = _jwt_with_claims({
|
||||
"sub": "test-user",
|
||||
"scope": "inference:mint_agent_key",
|
||||
"exp": int(time.time() + 3600),
|
||||
})
|
||||
_setup_nous_auth(
|
||||
hermes_home,
|
||||
access_token=token,
|
||||
scope=auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
|
||||
expires_at=_future_iso(3600),
|
||||
expires_in=3600,
|
||||
)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
calls = []
|
||||
|
||||
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
|
||||
del client, portal_base_url, min_ttl_seconds
|
||||
calls.append(access_token)
|
||||
return _mint_payload(api_key="opaque-agent-key")
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
|
||||
|
||||
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
|
||||
|
||||
assert calls == [token]
|
||||
assert creds["api_key"] == "opaque-agent-key"
|
||||
assert creds["source"] == "portal"
|
||||
payload = json.loads((hermes_home / "auth.json").read_text())
|
||||
assert payload["providers"]["nous"]["agent_key"] == "opaque-agent-key"
|
||||
assert payload["credential_pool"]["nous"][0]["agent_key"] == "opaque-agent-key"
|
||||
|
||||
|
||||
def test_nous_device_code_login_retries_legacy_scope_when_invoke_refused(monkeypatch):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
scopes = []
|
||||
|
||||
def _fake_request_device_code(*, client, portal_base_url, client_id, scope):
|
||||
del client, portal_base_url, client_id
|
||||
scopes.append(scope)
|
||||
if len(scopes) == 1:
|
||||
request = httpx.Request("POST", "https://portal.example.com/api/oauth/device/code")
|
||||
response = httpx.Response(
|
||||
400,
|
||||
json={
|
||||
"error": "invalid_scope",
|
||||
"error_description": "unsupported inference:invoke",
|
||||
},
|
||||
request=request,
|
||||
)
|
||||
raise httpx.HTTPStatusError("invalid_scope", request=request, response=response)
|
||||
return {
|
||||
"device_code": "device",
|
||||
"user_code": "user",
|
||||
"verification_uri": "https://portal.example.com/device",
|
||||
"verification_uri_complete": "https://portal.example.com/device?code=user",
|
||||
"expires_in": 600,
|
||||
"interval": 1,
|
||||
}
|
||||
|
||||
def _fake_poll_for_token(**kwargs):
|
||||
del kwargs
|
||||
return {
|
||||
"access_token": "access-legacy",
|
||||
"refresh_token": "refresh-legacy",
|
||||
"expires_in": 900,
|
||||
"scope": auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
|
||||
}
|
||||
|
||||
def _fake_refresh(state, **kwargs):
|
||||
del kwargs
|
||||
refreshed = dict(state)
|
||||
refreshed["agent_key"] = "opaque-agent-key"
|
||||
refreshed["agent_key_expires_at"] = _future_iso(1800)
|
||||
return refreshed
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_request_device_code", _fake_request_device_code)
|
||||
monkeypatch.setattr(auth_mod, "_poll_for_token", _fake_poll_for_token)
|
||||
monkeypatch.setattr(auth_mod, "refresh_nous_oauth_from_state", _fake_refresh)
|
||||
|
||||
result = auth_mod._nous_device_code_login(
|
||||
portal_base_url="https://portal.example.com",
|
||||
inference_base_url="https://inference.example.com/v1",
|
||||
open_browser=False,
|
||||
timeout_seconds=1,
|
||||
)
|
||||
|
||||
assert scopes == [auth_mod.DEFAULT_NOUS_SCOPE, auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE]
|
||||
assert result["scope"] == auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE
|
||||
assert result["agent_key"] == "opaque-agent-key"
|
||||
|
||||
|
||||
def test_forced_legacy_env_skips_invoke_scope_and_jwt_storage(tmp_path, monkeypatch):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
token = _invoke_jwt(seconds=3600)
|
||||
_setup_nous_auth(
|
||||
hermes_home,
|
||||
access_token=token,
|
||||
scope=auth_mod.DEFAULT_NOUS_SCOPE,
|
||||
expires_at=_future_iso(3600),
|
||||
expires_in=3600,
|
||||
)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setenv(auth_mod.NOUS_LEGACY_SESSION_KEYS_ENV, "true")
|
||||
|
||||
mint_calls = []
|
||||
|
||||
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
|
||||
del client, portal_base_url, min_ttl_seconds
|
||||
mint_calls.append(access_token)
|
||||
return _mint_payload(api_key="forced-legacy-key")
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
|
||||
|
||||
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
|
||||
|
||||
assert mint_calls == [token]
|
||||
assert creds["api_key"] == "forced-legacy-key"
|
||||
payload = json.loads((hermes_home / "auth.json").read_text())
|
||||
assert payload["providers"]["nous"]["agent_key"] == "forced-legacy-key"
|
||||
|
||||
requested_scopes = []
|
||||
|
||||
def _fake_request_device_code(*, client, portal_base_url, client_id, scope):
|
||||
del client, portal_base_url, client_id
|
||||
requested_scopes.append(scope)
|
||||
return {
|
||||
"device_code": "device",
|
||||
"user_code": "user",
|
||||
"verification_uri": "https://portal.example.com/device",
|
||||
"verification_uri_complete": "https://portal.example.com/device?code=user",
|
||||
"expires_in": 600,
|
||||
"interval": 1,
|
||||
}
|
||||
|
||||
def _fake_poll_for_token(**kwargs):
|
||||
del kwargs
|
||||
return {
|
||||
"access_token": "access-legacy",
|
||||
"refresh_token": "refresh-legacy",
|
||||
"expires_in": 900,
|
||||
"scope": auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
|
||||
}
|
||||
|
||||
def _fake_refresh(state, **kwargs):
|
||||
del kwargs
|
||||
refreshed = dict(state)
|
||||
refreshed["agent_key"] = "forced-legacy-login-key"
|
||||
refreshed["agent_key_expires_at"] = _future_iso(1800)
|
||||
return refreshed
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_request_device_code", _fake_request_device_code)
|
||||
monkeypatch.setattr(auth_mod, "_poll_for_token", _fake_poll_for_token)
|
||||
monkeypatch.setattr(auth_mod, "refresh_nous_oauth_from_state", _fake_refresh)
|
||||
|
||||
auth_mod._nous_device_code_login(
|
||||
portal_base_url="https://portal.example.com",
|
||||
inference_base_url="https://inference.example.com/v1",
|
||||
open_browser=False,
|
||||
timeout_seconds=1,
|
||||
)
|
||||
|
||||
assert requested_scopes == [auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE]
|
||||
|
||||
|
||||
def test_nous_inference_auth_logs_do_not_include_secret_values(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
caplog,
|
||||
):
|
||||
import hermes_cli.auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
token = _jwt_with_claims({
|
||||
"sub": "secret-user",
|
||||
"scope": "inference:mint_agent_key",
|
||||
"exp": int(time.time() + 3600),
|
||||
})
|
||||
refresh_token = "refresh-secret-token"
|
||||
opaque_key = "opaque-secret-agent-key"
|
||||
_setup_nous_auth(
|
||||
hermes_home,
|
||||
access_token=token,
|
||||
refresh_token=refresh_token,
|
||||
scope=auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
|
||||
expires_at=_future_iso(3600),
|
||||
expires_in=3600,
|
||||
)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
|
||||
del client, portal_base_url, access_token, min_ttl_seconds
|
||||
return _mint_payload(api_key=opaque_key)
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
|
||||
|
||||
caplog.set_level(logging.INFO, logger="hermes_cli.auth")
|
||||
auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
|
||||
|
||||
logged = caplog.text
|
||||
assert "legacy session key path" in logged
|
||||
assert token not in logged
|
||||
assert refresh_token not in logged
|
||||
assert opaque_key not in logged
|
||||
|
||||
|
||||
def test_get_nous_auth_status_checks_credential_pool(tmp_path, monkeypatch):
|
||||
"""get_nous_auth_status() should find Nous credentials in the pool
|
||||
even when the auth store has no Nous provider entry — this is the
|
||||
@@ -373,6 +838,99 @@ def test_refresh_token_persisted_when_mint_times_out(tmp_path, monkeypatch):
|
||||
assert state_after_failure["access_token"] == "access-1"
|
||||
|
||||
|
||||
def test_terminal_refresh_failure_quarantines_tokens(
|
||||
tmp_path, monkeypatch, shared_store_env,
|
||||
):
|
||||
"""A revoked/invalid Nous refresh token must not be replayed forever."""
|
||||
from hermes_cli import auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
_setup_nous_auth(hermes_home, refresh_token="refresh-old")
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
from agent.credential_pool import load_pool
|
||||
|
||||
assert load_pool("nous").select() is not None
|
||||
|
||||
shared_state = _full_state_fixture()
|
||||
shared_state["access_token"] = "access-old"
|
||||
shared_state["refresh_token"] = "refresh-old"
|
||||
shared_state["expires_at"] = "2026-02-01T00:00:00+00:00"
|
||||
auth_mod._write_shared_nous_state(shared_state)
|
||||
|
||||
refresh_calls: list[str] = []
|
||||
|
||||
def _terminal_refresh_failure(*, client, portal_base_url, client_id, refresh_token):
|
||||
refresh_calls.append(refresh_token)
|
||||
raise AuthError(
|
||||
"Refresh session has been revoked",
|
||||
provider="nous",
|
||||
code="invalid_grant",
|
||||
relogin_required=True,
|
||||
)
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_refresh_access_token", _terminal_refresh_failure)
|
||||
|
||||
with pytest.raises(AuthError, match="Refresh session has been revoked"):
|
||||
auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
|
||||
|
||||
state_after_failure = auth_mod.get_provider_auth_state("nous")
|
||||
assert state_after_failure is not None
|
||||
assert not state_after_failure.get("refresh_token")
|
||||
assert not state_after_failure.get("access_token")
|
||||
assert not state_after_failure.get("agent_key")
|
||||
assert state_after_failure["last_auth_error"]["code"] == "invalid_grant"
|
||||
assert auth_mod._read_shared_nous_state() is None
|
||||
payload = json.loads((hermes_home / "auth.json").read_text())
|
||||
assert payload.get("credential_pool", {}).get("nous") == []
|
||||
|
||||
with pytest.raises(AuthError, match="No access token found"):
|
||||
auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
|
||||
|
||||
assert refresh_calls == ["refresh-old"]
|
||||
|
||||
|
||||
def test_managed_access_token_refresh_failure_quarantines_tokens(
|
||||
tmp_path, monkeypatch, shared_store_env,
|
||||
):
|
||||
from hermes_cli import auth as auth_mod
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
_setup_nous_auth(hermes_home, refresh_token="refresh-old")
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
from agent.credential_pool import load_pool
|
||||
|
||||
assert load_pool("nous").select() is not None
|
||||
|
||||
refresh_calls: list[str] = []
|
||||
|
||||
def _terminal_refresh_failure(*, client, portal_base_url, client_id, refresh_token):
|
||||
refresh_calls.append(refresh_token)
|
||||
raise AuthError(
|
||||
"Invalid refresh token",
|
||||
provider="nous",
|
||||
code="invalid_grant",
|
||||
relogin_required=True,
|
||||
)
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_refresh_access_token", _terminal_refresh_failure)
|
||||
|
||||
with pytest.raises(AuthError, match="Invalid refresh token"):
|
||||
auth_mod.resolve_nous_access_token()
|
||||
|
||||
state_after_failure = auth_mod.get_provider_auth_state("nous")
|
||||
assert state_after_failure is not None
|
||||
assert not state_after_failure.get("refresh_token")
|
||||
assert not state_after_failure.get("access_token")
|
||||
assert state_after_failure["last_auth_error"]["message"] == "Invalid refresh token"
|
||||
payload = json.loads((hermes_home / "auth.json").read_text())
|
||||
assert payload.get("credential_pool", {}).get("nous") == []
|
||||
|
||||
with pytest.raises(AuthError, match="No access token found"):
|
||||
auth_mod.resolve_nous_access_token()
|
||||
|
||||
assert refresh_calls == ["refresh-old"]
|
||||
|
||||
|
||||
def test_mint_retry_uses_latest_rotated_refresh_token(tmp_path, monkeypatch):
|
||||
hermes_home = tmp_path / "hermes"
|
||||
_setup_nous_auth(hermes_home, refresh_token="refresh-old")
|
||||
@@ -555,7 +1113,7 @@ class TestLoginNousSkipKeepsCurrent:
|
||||
auth_path = hermes_home / "auth.json"
|
||||
auth_after = json.loads(auth_path.read_text())
|
||||
# active_provider should NOT be set to "nous" after Skip
|
||||
assert auth_after.get("active_provider") in (None, "")
|
||||
assert auth_after.get("active_provider") in {None, ""}
|
||||
# But Nous creds are still saved
|
||||
assert "nous" in auth_after.get("providers", {})
|
||||
|
||||
@@ -640,7 +1198,11 @@ def test_persist_nous_credentials_allows_recovery_from_401(tmp_path, monkeypatch
|
||||
calls after a Nous 401 — before the fix it would raise AuthError because
|
||||
providers.nous was empty.
|
||||
"""
|
||||
from hermes_cli.auth import persist_nous_credentials, resolve_nous_runtime_credentials
|
||||
from hermes_cli.auth import (
|
||||
NOUS_INFERENCE_AUTH_MODE_FRESH,
|
||||
persist_nous_credentials,
|
||||
resolve_nous_runtime_credentials,
|
||||
)
|
||||
|
||||
hermes_home = tmp_path / "hermes"
|
||||
hermes_home.mkdir(parents=True, exist_ok=True)
|
||||
@@ -668,7 +1230,10 @@ def test_persist_nous_credentials_allows_recovery_from_401(tmp_path, monkeypatch
|
||||
monkeypatch.setattr("hermes_cli.auth._refresh_access_token", _fake_refresh_access_token)
|
||||
monkeypatch.setattr("hermes_cli.auth._mint_agent_key", _fake_mint_agent_key)
|
||||
|
||||
creds = resolve_nous_runtime_credentials(min_key_ttl_seconds=300, force_mint=True)
|
||||
creds = resolve_nous_runtime_credentials(
|
||||
min_key_ttl_seconds=300,
|
||||
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_FRESH,
|
||||
)
|
||||
assert creds["api_key"] == "new-agent-key"
|
||||
|
||||
|
||||
@@ -861,6 +1426,36 @@ def test_refresh_token_reuse_detection_surfaces_actionable_message():
|
||||
assert exc_info.value.relogin_required is True
|
||||
|
||||
|
||||
def test_refresh_token_reuse_error_code_is_terminal():
|
||||
"""Nous may return refresh_token_reused as the OAuth error code itself."""
|
||||
from hermes_cli import auth as auth_mod
|
||||
|
||||
class _FakeResponse:
|
||||
status_code = 400
|
||||
|
||||
def json(self):
|
||||
return {
|
||||
"error": "refresh_token_reused",
|
||||
"error_description": "Refresh token reuse detected",
|
||||
}
|
||||
|
||||
class _FakeClient:
|
||||
def post(self, *args, **kwargs):
|
||||
return _FakeResponse()
|
||||
|
||||
with pytest.raises(AuthError) as exc_info:
|
||||
auth_mod._refresh_access_token(
|
||||
client=_FakeClient(),
|
||||
portal_base_url="https://portal.nousresearch.com",
|
||||
client_id="hermes-cli",
|
||||
refresh_token="rt_consumed_elsewhere",
|
||||
)
|
||||
|
||||
assert exc_info.value.code == "refresh_token_reused"
|
||||
assert exc_info.value.relogin_required is True
|
||||
assert auth_mod._is_terminal_nous_refresh_error(exc_info.value) is True
|
||||
|
||||
|
||||
def test_refresh_token_exchange_sends_refresh_token_header():
|
||||
"""Nous refresh tokens must be sent in a header so sandbox proxies can
|
||||
substitute placeholder credentials without parsing form bodies.
|
||||
@@ -1118,6 +1713,47 @@ def test_try_import_shared_returns_none_on_refresh_failure(
|
||||
monkeypatch.setattr(auth_mod, "refresh_nous_oauth_from_state", _boom)
|
||||
|
||||
assert auth_mod._try_import_shared_nous_state() is None
|
||||
assert auth_mod._read_shared_nous_state() is None
|
||||
|
||||
|
||||
def test_try_import_shared_persists_rotated_token_when_mint_fails(
|
||||
shared_store_env, monkeypatch,
|
||||
):
|
||||
"""A forced shared import refresh rotates the single-use token before minting.
|
||||
|
||||
If the later agent-key mint fails, the shared store must still keep the
|
||||
rotated refresh token; otherwise the next import attempt replays the
|
||||
consumed token and trips refresh-token reuse.
|
||||
"""
|
||||
from hermes_cli import auth as auth_mod
|
||||
|
||||
shared_state = _full_state_fixture()
|
||||
shared_state["refresh_token"] = "refresh-old"
|
||||
shared_state["access_token"] = "access-old"
|
||||
auth_mod._write_shared_nous_state(shared_state)
|
||||
|
||||
def _fake_refresh_access_token(*, client, portal_base_url, client_id, refresh_token):
|
||||
assert refresh_token == "refresh-old"
|
||||
return {
|
||||
"access_token": "access-new",
|
||||
"refresh_token": "refresh-new",
|
||||
"expires_in": 900,
|
||||
"token_type": "Bearer",
|
||||
}
|
||||
|
||||
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
|
||||
assert access_token == "access-new"
|
||||
raise AuthError("credits exhausted", provider="nous", code="insufficient_credits")
|
||||
|
||||
monkeypatch.setattr(auth_mod, "_refresh_access_token", _fake_refresh_access_token)
|
||||
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
|
||||
|
||||
assert auth_mod._try_import_shared_nous_state() is None
|
||||
|
||||
shared_after = auth_mod._read_shared_nous_state()
|
||||
assert shared_after is not None
|
||||
assert shared_after["refresh_token"] == "refresh-new"
|
||||
assert shared_after["access_token"] == "access-new"
|
||||
|
||||
|
||||
def test_try_import_shared_rehydrates_on_success(shared_store_env, monkeypatch):
|
||||
@@ -1132,7 +1768,10 @@ def test_try_import_shared_rehydrates_on_success(shared_store_env, monkeypatch):
|
||||
def _fake_refresh(state, **kwargs):
|
||||
# Simulate portal returning fresh tokens + a new agent_key
|
||||
assert kwargs.get("force_refresh") is True
|
||||
assert kwargs.get("force_mint") is True
|
||||
assert (
|
||||
kwargs.get("inference_auth_mode")
|
||||
== auth_mod.NOUS_INFERENCE_AUTH_MODE_FRESH
|
||||
)
|
||||
return {
|
||||
**state,
|
||||
"access_token": "fresh-access-tok",
|
||||
@@ -1260,7 +1899,7 @@ def test_runtime_refresh_uses_newer_shared_token_before_local_stale_token(
|
||||
|
||||
creds = auth_mod.resolve_nous_runtime_credentials(
|
||||
min_key_ttl_seconds=300,
|
||||
force_mint=True,
|
||||
inference_auth_mode=auth_mod.NOUS_INFERENCE_AUTH_MODE_FRESH,
|
||||
)
|
||||
|
||||
assert creds["api_key"] == "agent-key-from-shared-token"
|
||||
|
||||
@@ -157,6 +157,24 @@ class TestCmdUpdateBranchFallback:
|
||||
(["/usr/bin/npm", "run", "build"], PROJECT_ROOT / "apps" / "dashboard"),
|
||||
]
|
||||
|
||||
# Regression for #18840: repo root + ui-tui installs must stream
|
||||
# output (capture_output=False) so postinstall progress is visible
|
||||
# to the user.
|
||||
repo_and_tui_calls = [
|
||||
call
|
||||
for call in mock_run.call_args_list
|
||||
if call.args
|
||||
and call.args[0][0] == "/usr/bin/npm"
|
||||
and call.args[0][1] == "ci"
|
||||
and call.kwargs.get("cwd") in {PROJECT_ROOT, PROJECT_ROOT / "ui-tui"}
|
||||
]
|
||||
assert len(repo_and_tui_calls) == 2
|
||||
for call in repo_and_tui_calls:
|
||||
assert call.kwargs.get("capture_output") is False, (
|
||||
"repo-root / ui-tui npm install must stream output "
|
||||
"(no capture_output) so postinstall progress is visible"
|
||||
)
|
||||
|
||||
def test_update_non_interactive_runs_safe_config_migrations(self, mock_args, capsys):
|
||||
"""Dashboard/web updates apply non-interactive migrations before restart."""
|
||||
with patch("shutil.which", return_value=None), patch(
|
||||
|
||||
@@ -105,7 +105,7 @@ class TestApply:
|
||||
assert "Cannot enable" in r.message
|
||||
assert "npm i -g @openai/codex" in r.message
|
||||
# Config NOT mutated on failure
|
||||
assert cfg.get("model", {}).get("openai_runtime") in (None, "")
|
||||
assert cfg.get("model", {}).get("openai_runtime") in {None, ""}
|
||||
|
||||
def test_enable_succeeds_when_codex_present(self):
|
||||
cfg = {}
|
||||
|
||||
@@ -107,6 +107,7 @@ class TestResolveCommand:
|
||||
assert resolve_command("gateway").name == "platforms"
|
||||
assert resolve_command("set-home").name == "sethome"
|
||||
assert resolve_command("reload_mcp").name == "reload-mcp"
|
||||
assert resolve_command("codex_runtime").name == "codex-runtime"
|
||||
assert resolve_command("tasks").name == "agents"
|
||||
|
||||
def test_topic_is_gateway_command(self):
|
||||
@@ -251,6 +252,12 @@ class TestTelegramBotCommands:
|
||||
assert "queue" in names
|
||||
assert "steer" in names
|
||||
|
||||
def test_hyphenated_codex_runtime_is_exposed_as_underscore_command(self):
|
||||
"""Telegram autocomplete exposes /codex-runtime as /codex_runtime."""
|
||||
names = {name for name, _ in telegram_bot_commands()}
|
||||
assert "codex_runtime" in names
|
||||
assert "codex-runtime" not in names
|
||||
|
||||
|
||||
class TestSlackSubcommandMap:
|
||||
def test_returns_dict(self):
|
||||
|
||||
@@ -320,6 +320,7 @@ class TestDoctorMemoryProviderSection:
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -426,6 +427,7 @@ def test_run_doctor_accepts_named_provider_from_providers_section(monkeypatch, t
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -463,6 +465,7 @@ def test_run_doctor_accepts_bare_custom_provider(monkeypatch, tmp_path):
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -474,6 +477,48 @@ def test_run_doctor_accepts_bare_custom_provider(monkeypatch, tmp_path):
|
||||
assert "model.provider 'custom' is not a recognised provider" not in out
|
||||
|
||||
|
||||
def test_run_doctor_flags_missing_credentials_for_active_openrouter_provider(monkeypatch, tmp_path):
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
(home / "config.yaml").write_text(
|
||||
"model:\n"
|
||||
" provider: openrouter\n"
|
||||
" default: openai/gpt-4.1-mini\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
|
||||
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", tmp_path / "project")
|
||||
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
|
||||
(tmp_path / "project").mkdir(exist_ok=True)
|
||||
|
||||
fake_model_tools = types.SimpleNamespace(
|
||||
check_tool_availability=lambda *a, **kw: ([], []),
|
||||
TOOLSET_REQUIREMENTS={},
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
|
||||
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
|
||||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||||
|
||||
try:
|
||||
from hermes_cli import auth as _auth_mod
|
||||
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_gemini_oauth_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_minimax_oauth_auth_status", lambda: {})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
doctor_mod.run_doctor(Namespace(fix=False))
|
||||
|
||||
out = buf.getvalue()
|
||||
assert "model.provider 'openrouter' is set but no API key is configured" in out
|
||||
assert "No credentials found for provider 'openrouter'." in out
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("provider", "default_model"),
|
||||
[
|
||||
@@ -510,6 +555,7 @@ def test_run_doctor_accepts_hermes_provider_ids_that_catalog_aliases(
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -556,6 +602,7 @@ def test_run_doctor_accepts_kimi_coding_cn_provider(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_auth_status", lambda provider: {"logged_in": True})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -594,6 +641,7 @@ def test_run_doctor_termux_does_not_mark_browser_available_without_agent_browser
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -633,6 +681,7 @@ def test_run_doctor_kimi_cn_env_is_detected_and_probe_is_null_safe(monkeypatch,
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -681,6 +730,7 @@ def test_run_doctor_dashscope_retries_china_endpoint_after_intl_unauthorized(mon
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
@@ -739,6 +789,7 @@ def test_run_doctor_opencode_go_skips_invalid_models_probe(monkeypatch, tmp_path
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {})
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
@@ -850,6 +901,7 @@ def _run_doctor_with_healthy_oauth_fallback(
|
||||
failing_host: str,
|
||||
gemini_oauth_status: dict,
|
||||
minimax_oauth_status: dict,
|
||||
xai_oauth_status: dict | None = None,
|
||||
) -> str:
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
@@ -886,6 +938,8 @@ def _run_doctor_with_healthy_oauth_fallback(
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
|
||||
monkeypatch.setattr(_auth_mod, "get_gemini_oauth_auth_status", lambda: gemini_oauth_status)
|
||||
monkeypatch.setattr(_auth_mod, "get_minimax_oauth_auth_status", lambda: minimax_oauth_status)
|
||||
_xai_status = xai_oauth_status if xai_oauth_status is not None else {}
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: _xai_status)
|
||||
|
||||
def fake_get(url, headers=None, timeout=None):
|
||||
status = 401 if failing_host in url else 200
|
||||
@@ -902,7 +956,7 @@ def _run_doctor_with_healthy_oauth_fallback(
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("env_key", "bad_key", "failing_host", "gemini_oauth_status", "minimax_oauth_status", "unexpected_issue"),
|
||||
("env_key", "bad_key", "failing_host", "gemini_oauth_status", "minimax_oauth_status", "xai_oauth_status", "unexpected_issue"),
|
||||
[
|
||||
(
|
||||
"GOOGLE_API_KEY",
|
||||
@@ -910,6 +964,7 @@ def _run_doctor_with_healthy_oauth_fallback(
|
||||
"googleapis.com",
|
||||
{"logged_in": True, "email": "user@example.com"},
|
||||
{},
|
||||
None,
|
||||
"Check GOOGLE_API_KEY in .env",
|
||||
),
|
||||
(
|
||||
@@ -918,8 +973,18 @@ def _run_doctor_with_healthy_oauth_fallback(
|
||||
"minimax.io",
|
||||
{},
|
||||
{"logged_in": True, "region": "global"},
|
||||
None,
|
||||
"Check MINIMAX_API_KEY in .env",
|
||||
),
|
||||
(
|
||||
"XAI_API_KEY",
|
||||
"bad-xai-key",
|
||||
"api.x.ai",
|
||||
{},
|
||||
{},
|
||||
{"logged_in": True, "auth_mode": "oauth_pkce"},
|
||||
"Check XAI_API_KEY in .env",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_run_doctor_ignores_invalid_direct_keys_when_oauth_fallback_is_healthy(
|
||||
@@ -930,6 +995,7 @@ def test_run_doctor_ignores_invalid_direct_keys_when_oauth_fallback_is_healthy(
|
||||
failing_host,
|
||||
gemini_oauth_status,
|
||||
minimax_oauth_status,
|
||||
xai_oauth_status,
|
||||
unexpected_issue,
|
||||
):
|
||||
out = _run_doctor_with_healthy_oauth_fallback(
|
||||
@@ -940,7 +1006,220 @@ def test_run_doctor_ignores_invalid_direct_keys_when_oauth_fallback_is_healthy(
|
||||
failing_host=failing_host,
|
||||
gemini_oauth_status=gemini_oauth_status,
|
||||
minimax_oauth_status=minimax_oauth_status,
|
||||
xai_oauth_status=xai_oauth_status,
|
||||
)
|
||||
|
||||
assert "invalid API key" in out
|
||||
assert unexpected_issue not in out
|
||||
|
||||
|
||||
def test_has_healthy_oauth_fallback_returns_false_for_unknown_provider():
|
||||
from hermes_cli.doctor import _has_healthy_oauth_fallback_for_apikey_provider
|
||||
assert _has_healthy_oauth_fallback_for_apikey_provider("unknown-provider") is False
|
||||
|
||||
|
||||
class TestHasHealthyOauthFallbackForXai:
|
||||
def test_returns_true_when_xai_oauth_healthy(self, monkeypatch):
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {"logged_in": True})
|
||||
from hermes_cli.doctor import _has_healthy_oauth_fallback_for_apikey_provider
|
||||
assert _has_healthy_oauth_fallback_for_apikey_provider("xai") is True
|
||||
|
||||
def test_returns_false_when_xai_oauth_not_logged_in(self, monkeypatch):
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: {"logged_in": False})
|
||||
from hermes_cli.doctor import _has_healthy_oauth_fallback_for_apikey_provider
|
||||
assert _has_healthy_oauth_fallback_for_apikey_provider("xai") is False
|
||||
|
||||
def test_returns_false_when_xai_oauth_returns_none(self, monkeypatch):
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", lambda: None)
|
||||
from hermes_cli.doctor import _has_healthy_oauth_fallback_for_apikey_provider
|
||||
assert _has_healthy_oauth_fallback_for_apikey_provider("xai") is False
|
||||
|
||||
def test_returns_false_when_xai_import_unavailable(self, monkeypatch):
|
||||
import sys
|
||||
# Simulate get_xai_oauth_auth_status missing from auth module
|
||||
monkeypatch.delattr("hermes_cli.auth.get_xai_oauth_auth_status", raising=False)
|
||||
# Force doctor module to re-import the function
|
||||
monkeypatch.delitem(sys.modules, "hermes_cli.doctor", raising=False)
|
||||
from hermes_cli.doctor import _has_healthy_oauth_fallback_for_apikey_provider
|
||||
assert _has_healthy_oauth_fallback_for_apikey_provider("xai") is False
|
||||
|
||||
def test_xai_import_failure_does_not_affect_gemini(self, monkeypatch):
|
||||
import sys
|
||||
from hermes_cli import auth as _auth_mod
|
||||
# xAI function missing, but Gemini is healthy
|
||||
monkeypatch.delattr(_auth_mod, "get_xai_oauth_auth_status", raising=False)
|
||||
monkeypatch.setattr(_auth_mod, "get_gemini_oauth_auth_status", lambda: {"logged_in": True})
|
||||
monkeypatch.delitem(sys.modules, "hermes_cli.doctor", raising=False)
|
||||
from hermes_cli.doctor import _has_healthy_oauth_fallback_for_apikey_provider
|
||||
assert _has_healthy_oauth_fallback_for_apikey_provider("gemini") is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ◆ Auth Providers — xAI OAuth display in run_doctor()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDoctorXaiOAuthStatus:
|
||||
"""The ◆ Auth Providers section must show xAI OAuth login state.
|
||||
|
||||
xAI OAuth is checked in a *separate* try/except block so that an import
|
||||
failure (or runtime exception) cannot silence the Nous / Codex / Gemini /
|
||||
MiniMax rows that were already printed above it.
|
||||
"""
|
||||
|
||||
def _run(self, monkeypatch, tmp_path, *, xai_auth_fn) -> str:
|
||||
"""Run doctor with a controlled xAI auth callable; return stdout."""
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
(home / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
|
||||
project = tmp_path / "project"
|
||||
project.mkdir(exist_ok=True)
|
||||
|
||||
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
|
||||
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
|
||||
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
|
||||
|
||||
fake_model_tools = types.SimpleNamespace(
|
||||
check_tool_availability=lambda *a, **kw: ([], []),
|
||||
TOOLSET_REQUIREMENTS={},
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
|
||||
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_gemini_oauth_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_minimax_oauth_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_xai_oauth_auth_status", xai_auth_fn)
|
||||
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
doctor_mod.run_doctor(Namespace(fix=False))
|
||||
return buf.getvalue()
|
||||
|
||||
def test_logged_in_shows_ok(self, monkeypatch, tmp_path):
|
||||
out = self._run(
|
||||
monkeypatch, tmp_path,
|
||||
xai_auth_fn=lambda: {"logged_in": True},
|
||||
)
|
||||
assert "xAI OAuth" in out
|
||||
assert "(logged in)" in out
|
||||
|
||||
def test_not_logged_in_shows_warn(self, monkeypatch, tmp_path):
|
||||
out = self._run(
|
||||
monkeypatch, tmp_path,
|
||||
xai_auth_fn=lambda: {"logged_in": False},
|
||||
)
|
||||
assert "xAI OAuth" in out
|
||||
assert "(not logged in)" in out
|
||||
|
||||
def test_error_shown_when_not_logged_in_and_error_present(self, monkeypatch, tmp_path):
|
||||
out = self._run(
|
||||
monkeypatch, tmp_path,
|
||||
xai_auth_fn=lambda: {"logged_in": False, "error": "refresh token expired"},
|
||||
)
|
||||
assert "xAI OAuth" in out
|
||||
assert "refresh token expired" in out
|
||||
|
||||
def test_no_error_line_when_error_key_absent(self, monkeypatch, tmp_path):
|
||||
out = self._run(
|
||||
monkeypatch, tmp_path,
|
||||
xai_auth_fn=lambda: {"logged_in": False},
|
||||
)
|
||||
assert "xAI OAuth" in out
|
||||
# The check_info line is only emitted when the "error" key is present.
|
||||
# Pick a token that would appear in no ordinary doctor output.
|
||||
assert "refresh token expired" not in out
|
||||
|
||||
def test_logged_in_does_not_emit_not_logged_in_on_xai_line(self, monkeypatch, tmp_path):
|
||||
out = self._run(
|
||||
monkeypatch, tmp_path,
|
||||
xai_auth_fn=lambda: {"logged_in": True},
|
||||
)
|
||||
assert "xAI OAuth" in out
|
||||
# The xAI OAuth line itself must say "(logged in)", not "(not logged in)".
|
||||
xai_line = next(l for l in out.splitlines() if "xAI OAuth" in l)
|
||||
assert "(logged in)" in xai_line
|
||||
assert "(not logged in)" not in xai_line
|
||||
|
||||
def test_import_failure_does_not_crash_doctor(self, monkeypatch, tmp_path):
|
||||
"""Doctor must not crash when get_xai_oauth_auth_status cannot be imported."""
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
(home / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
|
||||
project = tmp_path / "project"
|
||||
project.mkdir(exist_ok=True)
|
||||
|
||||
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
|
||||
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
|
||||
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
|
||||
|
||||
fake_model_tools = types.SimpleNamespace(
|
||||
check_tool_availability=lambda *a, **kw: ([], []),
|
||||
TOOLSET_REQUIREMENTS={},
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
|
||||
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_gemini_oauth_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_minimax_oauth_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.delattr(_auth_mod, "get_xai_oauth_auth_status", raising=False)
|
||||
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
doctor_mod.run_doctor(Namespace(fix=False))
|
||||
out = buf.getvalue()
|
||||
# The ◆ Auth Providers header must still appear — other providers unaffected.
|
||||
assert "Auth Providers" in out
|
||||
|
||||
def test_import_failure_does_not_affect_other_providers(self, monkeypatch, tmp_path):
|
||||
"""Nous / Codex / Gemini / MiniMax rows must survive an xAI import failure."""
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
(home / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
|
||||
project = tmp_path / "project"
|
||||
project.mkdir(exist_ok=True)
|
||||
|
||||
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
|
||||
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
|
||||
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
|
||||
|
||||
fake_model_tools = types.SimpleNamespace(
|
||||
check_tool_availability=lambda *a, **kw: ([], []),
|
||||
TOOLSET_REQUIREMENTS={},
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
|
||||
|
||||
from hermes_cli import auth as _auth_mod
|
||||
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {"logged_in": True})
|
||||
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_gemini_oauth_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.setattr(_auth_mod, "get_minimax_oauth_auth_status", lambda: {"logged_in": False})
|
||||
monkeypatch.delattr(_auth_mod, "get_xai_oauth_auth_status", raising=False)
|
||||
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
doctor_mod.run_doctor(Namespace(fix=False))
|
||||
out = buf.getvalue()
|
||||
assert "Nous Portal auth" in out
|
||||
assert "logged in" in out
|
||||
|
||||
def test_function_raises_does_not_crash_doctor(self, monkeypatch, tmp_path):
|
||||
"""A runtime exception from get_xai_oauth_auth_status must be swallowed."""
|
||||
def _raise():
|
||||
raise RuntimeError("simulated xAI status failure")
|
||||
|
||||
out = self._run(monkeypatch, tmp_path, xai_auth_fn=_raise)
|
||||
assert "Auth Providers" in out
|
||||
|
||||
def test_function_returns_none_does_not_crash_doctor(self, monkeypatch, tmp_path):
|
||||
"""None return is normalised to {} via `or {}` — must not AttributeError."""
|
||||
out = self._run(monkeypatch, tmp_path, xai_auth_fn=lambda: None)
|
||||
# None → {} → logged_in falsy → shows not-logged-in warn
|
||||
assert "xAI OAuth" in out
|
||||
assert "(not logged in)" in out
|
||||
|
||||
@@ -48,7 +48,7 @@ class TestInstallCuaDriverUpgrade:
|
||||
with patch("platform.system", return_value="Darwin"), \
|
||||
patch.object(tools_config.shutil, "which",
|
||||
side_effect=lambda n: "/usr/local/bin/" + n
|
||||
if n in ("cua-driver", "curl") else None), \
|
||||
if n in {"cua-driver", "curl"} else None), \
|
||||
patch.object(tools_config, "_run_cua_driver_installer",
|
||||
return_value=True) as runner, \
|
||||
patch("subprocess.run"):
|
||||
@@ -82,7 +82,7 @@ class TestInstallCuaDriverUpgrade:
|
||||
with patch("platform.system", return_value="Darwin"), \
|
||||
patch.object(tools_config.shutil, "which",
|
||||
side_effect=lambda n: "/usr/local/bin/" + n
|
||||
if n in ("cua-driver", "curl") else None), \
|
||||
if n in {"cua-driver", "curl"} else None), \
|
||||
patch.object(tools_config, "_run_cua_driver_installer") as runner, \
|
||||
patch("subprocess.run"):
|
||||
assert tools_config.install_cua_driver(upgrade=False) is True
|
||||
|
||||
@@ -1046,7 +1046,7 @@ def test_enforce_max_runtime_integrates_with_dispatch(kanban_home, monkeypatch):
|
||||
task = kb.get_task(conn, tid)
|
||||
# After timeout, task is back in 'ready' and will be re-spawned
|
||||
# by the same pass. That's the intended behaviour.
|
||||
assert task.status in ("ready", "running")
|
||||
assert task.status in {"ready", "running"}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
@@ -0,0 +1,242 @@
|
||||
"""Tests for the decomposer module + `hermes kanban decompose` CLI surface.
|
||||
|
||||
The auxiliary LLM client is mocked — no network calls. Tests exercise the
|
||||
prompt plumbing, response parsing, DB writes (via the real DB helper),
|
||||
and the assignee-fallback logic.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json as jsonlib
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import kanban as kanban_cli
|
||||
from hermes_cli import kanban_db as kb
|
||||
from hermes_cli import kanban_decompose as decomp
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def kanban_home(tmp_path, monkeypatch):
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
kb.init_db()
|
||||
return home
|
||||
|
||||
|
||||
def _fake_aux_response(content: str):
|
||||
resp = MagicMock()
|
||||
resp.choices = [MagicMock()]
|
||||
resp.choices[0].message.content = content
|
||||
return resp
|
||||
|
||||
|
||||
def _mock_client_returning(content: str):
|
||||
client = MagicMock()
|
||||
client.chat.completions.create = MagicMock(return_value=_fake_aux_response(content))
|
||||
return client
|
||||
|
||||
|
||||
def _patch_aux_client(content: str, *, model: str = "test-model"):
|
||||
client = _mock_client_returning(content)
|
||||
return patch(
|
||||
"agent.auxiliary_client.get_text_auxiliary_client",
|
||||
return_value=(client, model),
|
||||
)
|
||||
|
||||
|
||||
def _patch_extra_body():
|
||||
return patch(
|
||||
"agent.auxiliary_client.get_auxiliary_extra_body",
|
||||
return_value={},
|
||||
)
|
||||
|
||||
|
||||
def _patch_list_profiles(names: list[str]):
|
||||
"""Pretend the named profiles exist. The decomposer uses
|
||||
profiles_mod.list_profiles() to build the roster + valid-set, and
|
||||
profiles_mod.profile_exists() to resolve orchestrator/default."""
|
||||
from types import SimpleNamespace
|
||||
fake_profiles = [
|
||||
SimpleNamespace(
|
||||
name=n, is_default=(i == 0), description=f"desc for {n}",
|
||||
description_auto=False, model="m", provider="p", skill_count=1,
|
||||
)
|
||||
for i, n in enumerate(names)
|
||||
]
|
||||
return [
|
||||
patch("hermes_cli.profiles.list_profiles", return_value=fake_profiles),
|
||||
patch("hermes_cli.profiles.profile_exists", side_effect=lambda x: x in names),
|
||||
patch("hermes_cli.profiles.get_active_profile_name", return_value=names[0] if names else "default"),
|
||||
]
|
||||
|
||||
|
||||
def test_decompose_with_fanout_creates_children(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="ship a feature", triage=True)
|
||||
|
||||
llm_payload = jsonlib.dumps({
|
||||
"fanout": True,
|
||||
"rationale": "test split",
|
||||
"tasks": [
|
||||
{"title": "research", "body": "look it up", "assignee": "researcher", "parents": []},
|
||||
{"title": "build", "body": "code it", "assignee": "engineer", "parents": [0]},
|
||||
],
|
||||
})
|
||||
|
||||
patches = _patch_list_profiles(["orchestrator", "researcher", "engineer"])
|
||||
for p in patches:
|
||||
p.start()
|
||||
try:
|
||||
with _patch_aux_client(llm_payload), _patch_extra_body():
|
||||
outcome = decomp.decompose_task(tid, author="me")
|
||||
finally:
|
||||
for p in patches:
|
||||
p.stop()
|
||||
|
||||
assert outcome.ok, outcome.reason
|
||||
assert outcome.fanout is True
|
||||
assert outcome.child_ids and len(outcome.child_ids) == 2
|
||||
|
||||
with kb.connect() as conn:
|
||||
root = kb.get_task(conn, tid)
|
||||
c0 = kb.get_task(conn, outcome.child_ids[0])
|
||||
c1 = kb.get_task(conn, outcome.child_ids[1])
|
||||
assert root.status == "todo"
|
||||
assert c0.status == "ready"
|
||||
assert c1.status == "todo"
|
||||
assert c0.assignee == "researcher"
|
||||
assert c1.assignee == "engineer"
|
||||
|
||||
|
||||
def test_decompose_fanout_false_falls_back_to_specify(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="just one thing", triage=True)
|
||||
|
||||
llm_payload = jsonlib.dumps({
|
||||
"fanout": False,
|
||||
"rationale": "single unit",
|
||||
"title": "Tightened title",
|
||||
"body": "**Goal**\nDo the thing.",
|
||||
})
|
||||
|
||||
patches = _patch_list_profiles(["orchestrator"])
|
||||
for p in patches:
|
||||
p.start()
|
||||
try:
|
||||
with _patch_aux_client(llm_payload), _patch_extra_body():
|
||||
outcome = decomp.decompose_task(tid, author="me")
|
||||
finally:
|
||||
for p in patches:
|
||||
p.stop()
|
||||
|
||||
assert outcome.ok, outcome.reason
|
||||
assert outcome.fanout is False
|
||||
assert outcome.new_title == "Tightened title"
|
||||
with kb.connect() as conn:
|
||||
task = kb.get_task(conn, tid)
|
||||
# specify path with no parents -> recompute_ready flips to 'ready'
|
||||
assert task.status == "ready"
|
||||
assert task.title == "Tightened title"
|
||||
|
||||
|
||||
def test_decompose_unknown_assignee_falls_back_to_default(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="x", triage=True)
|
||||
|
||||
# Roster only has 'orchestrator' and 'fallback'; LLM picks 'made_up'.
|
||||
llm_payload = jsonlib.dumps({
|
||||
"fanout": True,
|
||||
"rationale": "test",
|
||||
"tasks": [
|
||||
{"title": "do X", "body": "", "assignee": "made_up", "parents": []},
|
||||
],
|
||||
})
|
||||
|
||||
patches = _patch_list_profiles(["orchestrator", "fallback"])
|
||||
for p in patches:
|
||||
p.start()
|
||||
try:
|
||||
with patch.dict(
|
||||
"os.environ", {}, clear=False,
|
||||
), _patch_aux_client(llm_payload), _patch_extra_body(), \
|
||||
patch(
|
||||
"hermes_cli.kanban_decompose._load_config",
|
||||
return_value={
|
||||
"kanban": {
|
||||
"orchestrator_profile": "orchestrator",
|
||||
"default_assignee": "fallback",
|
||||
}
|
||||
},
|
||||
):
|
||||
outcome = decomp.decompose_task(tid, author="me")
|
||||
finally:
|
||||
for p in patches:
|
||||
p.stop()
|
||||
|
||||
assert outcome.ok, outcome.reason
|
||||
assert outcome.child_ids and len(outcome.child_ids) == 1
|
||||
with kb.connect() as conn:
|
||||
child = kb.get_task(conn, outcome.child_ids[0])
|
||||
# 'made_up' wasn't in roster, so assignee rewritten to 'fallback'
|
||||
assert child.assignee == "fallback"
|
||||
|
||||
|
||||
def test_decompose_handles_malformed_llm_json(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="x", triage=True)
|
||||
|
||||
patches = _patch_list_profiles(["orchestrator"])
|
||||
for p in patches:
|
||||
p.start()
|
||||
try:
|
||||
with _patch_aux_client("not json at all, sorry"), _patch_extra_body():
|
||||
outcome = decomp.decompose_task(tid, author="me")
|
||||
finally:
|
||||
for p in patches:
|
||||
p.stop()
|
||||
|
||||
assert outcome.ok is False
|
||||
assert "malformed JSON" in outcome.reason
|
||||
|
||||
|
||||
def test_decompose_returns_false_when_task_not_triage(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="x") # ready, not triage
|
||||
|
||||
patches = _patch_list_profiles(["orchestrator"])
|
||||
for p in patches:
|
||||
p.start()
|
||||
try:
|
||||
outcome = decomp.decompose_task(tid, author="me")
|
||||
finally:
|
||||
for p in patches:
|
||||
p.stop()
|
||||
assert outcome.ok is False
|
||||
assert "not in triage" in outcome.reason
|
||||
|
||||
|
||||
def test_decompose_no_aux_client_configured(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="x", triage=True)
|
||||
|
||||
patches = _patch_list_profiles(["orchestrator"])
|
||||
for p in patches:
|
||||
p.start()
|
||||
try:
|
||||
with patch(
|
||||
"agent.auxiliary_client.get_text_auxiliary_client",
|
||||
return_value=(None, ""),
|
||||
):
|
||||
outcome = decomp.decompose_task(tid, author="me")
|
||||
finally:
|
||||
for p in patches:
|
||||
p.stop()
|
||||
|
||||
assert outcome.ok is False
|
||||
assert "no auxiliary client" in outcome.reason
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Tests for kb.decompose_triage_task — the DB-layer atomic fan-out
|
||||
from the triage column. LLM-free by design.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import kanban_db as kb
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def kanban_home(tmp_path, monkeypatch):
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
kb.init_db()
|
||||
return home
|
||||
|
||||
|
||||
def _create_triage(conn, title="rough idea", body=None, assignee=None, tenant=None):
|
||||
return kb.create_task(
|
||||
conn,
|
||||
title=title,
|
||||
body=body,
|
||||
assignee=assignee,
|
||||
tenant=tenant,
|
||||
triage=True,
|
||||
)
|
||||
|
||||
|
||||
def test_decompose_creates_children_and_promotes_root(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = _create_triage(conn, title="ship a feature")
|
||||
assert kb.get_task(conn, tid).status == "triage"
|
||||
|
||||
children = [
|
||||
{"title": "research", "body": "look at prior art", "assignee": "researcher", "parents": []},
|
||||
{"title": "build it", "body": "write code", "assignee": "engineer", "parents": [0]},
|
||||
]
|
||||
with kb.connect() as conn:
|
||||
child_ids = kb.decompose_triage_task(
|
||||
conn,
|
||||
tid,
|
||||
root_assignee="orchestrator",
|
||||
children=children,
|
||||
author="decomposer",
|
||||
)
|
||||
assert child_ids is not None
|
||||
assert len(child_ids) == 2
|
||||
|
||||
with kb.connect() as conn:
|
||||
root = kb.get_task(conn, tid)
|
||||
c0 = kb.get_task(conn, child_ids[0])
|
||||
c1 = kb.get_task(conn, child_ids[1])
|
||||
|
||||
# Root flipped to todo with orchestrator assignee, gated by children.
|
||||
assert root.status == "todo"
|
||||
assert root.assignee == "orchestrator"
|
||||
# First child has no internal parents → ready on recompute_ready.
|
||||
assert c0.status == "ready"
|
||||
assert c0.assignee == "researcher"
|
||||
# Second child has parents=[0] → stays in todo until c0 completes.
|
||||
assert c1.status == "todo"
|
||||
assert c1.assignee == "engineer"
|
||||
|
||||
|
||||
def test_decompose_returns_none_when_task_missing(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
result = kb.decompose_triage_task(
|
||||
conn,
|
||||
"nonexistent",
|
||||
root_assignee="orch",
|
||||
children=[{"title": "x"}],
|
||||
author="me",
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_decompose_returns_none_when_task_not_in_triage(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="already a real task") # not triage
|
||||
result = kb.decompose_triage_task(
|
||||
conn,
|
||||
tid,
|
||||
root_assignee="orch",
|
||||
children=[{"title": "x"}],
|
||||
author="me",
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_decompose_empty_children_returns_none(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = _create_triage(conn)
|
||||
result = kb.decompose_triage_task(
|
||||
conn,
|
||||
tid,
|
||||
root_assignee="orch",
|
||||
children=[],
|
||||
author="me",
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_decompose_rejects_self_parent(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = _create_triage(conn)
|
||||
with pytest.raises(ValueError, match="cannot list itself"):
|
||||
kb.decompose_triage_task(
|
||||
conn,
|
||||
tid,
|
||||
root_assignee="orch",
|
||||
children=[{"title": "x", "parents": [0]}],
|
||||
author="me",
|
||||
)
|
||||
|
||||
|
||||
def test_decompose_rejects_out_of_range_parent(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = _create_triage(conn)
|
||||
with pytest.raises(ValueError, match="not a valid index"):
|
||||
kb.decompose_triage_task(
|
||||
conn,
|
||||
tid,
|
||||
root_assignee="orch",
|
||||
children=[{"title": "x", "parents": [5]}],
|
||||
author="me",
|
||||
)
|
||||
|
||||
|
||||
def test_decompose_records_audit_comment_and_event(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = _create_triage(conn)
|
||||
child_ids = kb.decompose_triage_task(
|
||||
conn,
|
||||
tid,
|
||||
root_assignee="orch",
|
||||
children=[{"title": "task A", "assignee": "researcher"}],
|
||||
author="alice",
|
||||
)
|
||||
assert child_ids is not None
|
||||
|
||||
with kb.connect() as conn:
|
||||
comments = kb.list_comments(conn, tid)
|
||||
events = kb.list_events(conn, tid)
|
||||
|
||||
assert any("Decomposed into" in (c.body or "") for c in comments)
|
||||
assert any(ev.kind == "decomposed" for ev in events)
|
||||
@@ -43,9 +43,9 @@ def _run_memory_reset(target="all", yes=False, monkeypatch=None, confirm_input="
|
||||
|
||||
mem_dir = get_hermes_home() / "memories"
|
||||
files_to_reset = []
|
||||
if target in ("all", "memory"):
|
||||
if target in {"all", "memory"}:
|
||||
files_to_reset.append(("MEMORY.md", "agent notes"))
|
||||
if target in ("all", "user"):
|
||||
if target in {"all", "user"}:
|
||||
files_to_reset.append(("USER.md", "user profile"))
|
||||
|
||||
existing = [(f, desc) for f, desc in files_to_reset if (mem_dir / f).exists()]
|
||||
|
||||
@@ -252,7 +252,7 @@ class TestDetectProviderForModel:
|
||||
result = detect_provider_for_model("deepseek-chat", "openai-codex")
|
||||
assert result is not None
|
||||
# Provider is deepseek (direct) or openrouter (fallback) depending on creds
|
||||
assert result[0] in ("deepseek", "openrouter")
|
||||
assert result[0] in {"deepseek", "openrouter"}
|
||||
|
||||
def test_current_provider_model_returns_none(self):
|
||||
"""Models belonging to the current provider should not trigger a switch."""
|
||||
@@ -302,7 +302,7 @@ class TestDetectProviderForModel:
|
||||
with patch("hermes_cli.models.fetch_openrouter_models", return_value=LIVE_OPENROUTER_MODELS):
|
||||
result = detect_provider_for_model("claude-opus-4-6", "openai-codex")
|
||||
assert result is not None
|
||||
assert result[0] not in ("nous",) # nous has claude models but shouldn't be suggested
|
||||
assert result[0] not in {"nous",} # nous has claude models but shouldn't be suggested
|
||||
|
||||
|
||||
class TestIsNousFreeTier:
|
||||
|
||||
@@ -44,7 +44,7 @@ def test_opencode_go_appears_when_api_key_set():
|
||||
# opencode-go can appear as "built-in" (from PROVIDER_TO_MODELS_DEV when
|
||||
# models.dev is reachable) or "hermes" (from HERMES_OVERLAYS fallback when
|
||||
# the API is unavailable, e.g. in CI).
|
||||
assert opencode_go["source"] in ("built-in", "hermes")
|
||||
assert opencode_go["source"] in {"built-in", "hermes"}
|
||||
|
||||
|
||||
def test_opencode_go_not_appears_when_no_creds():
|
||||
|
||||
@@ -0,0 +1,168 @@
|
||||
"""Tests for the profile.yaml metadata layer (description + description_auto)
|
||||
and the profile_describer LLM module.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json as jsonlib
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import profiles as profiles_mod
|
||||
from hermes_cli import profile_describer as describer
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def profile_env(tmp_path, monkeypatch):
|
||||
"""Set up an isolated HERMES_HOME with a default profile dir."""
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
return home
|
||||
|
||||
|
||||
def test_read_profile_meta_empty_when_missing(profile_env):
|
||||
meta = profiles_mod.read_profile_meta(profile_env)
|
||||
assert meta == {"description": "", "description_auto": False}
|
||||
|
||||
|
||||
def test_write_and_read_profile_meta(profile_env):
|
||||
profiles_mod.write_profile_meta(
|
||||
profile_env,
|
||||
description="a useful researcher",
|
||||
description_auto=False,
|
||||
)
|
||||
meta = profiles_mod.read_profile_meta(profile_env)
|
||||
assert meta["description"] == "a useful researcher"
|
||||
assert meta["description_auto"] is False
|
||||
|
||||
|
||||
def test_write_profile_meta_preserves_other_fields(profile_env):
|
||||
# First write sets description_auto=True; second write only updates
|
||||
# description and leaves description_auto unchanged.
|
||||
profiles_mod.write_profile_meta(
|
||||
profile_env,
|
||||
description="auto-gen",
|
||||
description_auto=True,
|
||||
)
|
||||
profiles_mod.write_profile_meta(profile_env, description="edited by hand")
|
||||
meta = profiles_mod.read_profile_meta(profile_env)
|
||||
assert meta["description"] == "edited by hand"
|
||||
assert meta["description_auto"] is True
|
||||
|
||||
|
||||
def test_write_profile_meta_rejects_missing_dir(tmp_path):
|
||||
bogus = tmp_path / "does_not_exist"
|
||||
with pytest.raises(FileNotFoundError):
|
||||
profiles_mod.write_profile_meta(bogus, description="x")
|
||||
|
||||
|
||||
def test_read_profile_meta_tolerates_corrupt_yaml(profile_env):
|
||||
(profile_env / "profile.yaml").write_text("not: valid: yaml: [unclosed")
|
||||
meta = profiles_mod.read_profile_meta(profile_env)
|
||||
assert meta == {"description": "", "description_auto": False}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# profile_describer module
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _fake_aux_response(content: str):
|
||||
resp = MagicMock()
|
||||
resp.choices = [MagicMock()]
|
||||
resp.choices[0].message.content = content
|
||||
return resp
|
||||
|
||||
|
||||
def _patch_aux_client(content: str):
|
||||
client = MagicMock()
|
||||
client.chat.completions.create = MagicMock(return_value=_fake_aux_response(content))
|
||||
return patch(
|
||||
"agent.auxiliary_client.get_text_auxiliary_client",
|
||||
return_value=(client, "test-model"),
|
||||
)
|
||||
|
||||
|
||||
def test_describer_writes_description_with_auto_true(profile_env, monkeypatch):
|
||||
# Pretend "myprof" is a registered profile pointing at profile_env.
|
||||
monkeypatch.setattr(
|
||||
profiles_mod, "profile_exists", lambda n: n == "myprof",
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
profiles_mod, "normalize_profile_name", lambda n: n,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
profiles_mod, "get_profile_dir", lambda n: profile_env,
|
||||
)
|
||||
|
||||
payload = jsonlib.dumps({"description": "writes Python codebases"})
|
||||
with _patch_aux_client(payload), patch(
|
||||
"agent.auxiliary_client.get_auxiliary_extra_body", return_value={}
|
||||
):
|
||||
outcome = describer.describe_profile("myprof")
|
||||
|
||||
assert outcome.ok, outcome.reason
|
||||
assert outcome.description == "writes Python codebases"
|
||||
meta = profiles_mod.read_profile_meta(profile_env)
|
||||
assert meta["description"] == "writes Python codebases"
|
||||
assert meta["description_auto"] is True
|
||||
|
||||
|
||||
def test_describer_refuses_to_overwrite_user_authored(profile_env, monkeypatch):
|
||||
profiles_mod.write_profile_meta(
|
||||
profile_env, description="curated", description_auto=False,
|
||||
)
|
||||
monkeypatch.setattr(profiles_mod, "profile_exists", lambda n: n == "myprof")
|
||||
monkeypatch.setattr(profiles_mod, "normalize_profile_name", lambda n: n)
|
||||
monkeypatch.setattr(profiles_mod, "get_profile_dir", lambda n: profile_env)
|
||||
|
||||
outcome = describer.describe_profile("myprof")
|
||||
assert outcome.ok is False
|
||||
assert "already has a user-authored description" in outcome.reason
|
||||
# Description unchanged
|
||||
assert profiles_mod.read_profile_meta(profile_env)["description"] == "curated"
|
||||
|
||||
|
||||
def test_describer_overwrite_flag_replaces_user_authored(profile_env, monkeypatch):
|
||||
profiles_mod.write_profile_meta(
|
||||
profile_env, description="curated", description_auto=False,
|
||||
)
|
||||
monkeypatch.setattr(profiles_mod, "profile_exists", lambda n: n == "myprof")
|
||||
monkeypatch.setattr(profiles_mod, "normalize_profile_name", lambda n: n)
|
||||
monkeypatch.setattr(profiles_mod, "get_profile_dir", lambda n: profile_env)
|
||||
|
||||
payload = jsonlib.dumps({"description": "new auto-gen"})
|
||||
with _patch_aux_client(payload), patch(
|
||||
"agent.auxiliary_client.get_auxiliary_extra_body", return_value={}
|
||||
):
|
||||
outcome = describer.describe_profile("myprof", overwrite=True)
|
||||
assert outcome.ok, outcome.reason
|
||||
meta = profiles_mod.read_profile_meta(profile_env)
|
||||
assert meta["description"] == "new auto-gen"
|
||||
assert meta["description_auto"] is True
|
||||
|
||||
|
||||
def test_describer_handles_malformed_llm_response(profile_env, monkeypatch):
|
||||
monkeypatch.setattr(profiles_mod, "profile_exists", lambda n: n == "myprof")
|
||||
monkeypatch.setattr(profiles_mod, "normalize_profile_name", lambda n: n)
|
||||
monkeypatch.setattr(profiles_mod, "get_profile_dir", lambda n: profile_env)
|
||||
|
||||
# Non-JSON: describer falls back to taking the first paragraph as the description.
|
||||
with _patch_aux_client("Plain text description that sneaks in"), patch(
|
||||
"agent.auxiliary_client.get_auxiliary_extra_body", return_value={}
|
||||
):
|
||||
outcome = describer.describe_profile("myprof")
|
||||
assert outcome.ok
|
||||
assert "Plain text description" in (outcome.description or "")
|
||||
|
||||
|
||||
def test_describer_returns_false_when_profile_missing(profile_env, monkeypatch):
|
||||
monkeypatch.setattr(profiles_mod, "profile_exists", lambda n: False)
|
||||
monkeypatch.setattr(profiles_mod, "normalize_profile_name", lambda n: n)
|
||||
outcome = describer.describe_profile("ghost")
|
||||
assert outcome.ok is False
|
||||
assert "not found" in outcome.reason
|
||||
+175
-23
@@ -103,7 +103,7 @@ def test_nous_adapter_authenticated_with_refresh_token_only(tmp_path, monkeypatc
|
||||
assert NousPortalAdapter().is_authenticated()
|
||||
|
||||
|
||||
def test_nous_adapter_get_credential_refreshes_and_persists(tmp_path, monkeypatch):
|
||||
def test_nous_adapter_get_credential_uses_runtime_resolver(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
_write_auth_store(tmp_path, {
|
||||
"access_token": "access-tok",
|
||||
@@ -114,31 +114,82 @@ def test_nous_adapter_get_credential_refreshes_and_persists(tmp_path, monkeypatc
|
||||
})
|
||||
|
||||
refreshed_state = {
|
||||
"access_token": "access-tok",
|
||||
"refresh_token": "refresh-tok",
|
||||
"client_id": "hermes-cli",
|
||||
"portal_base_url": "https://portal.nousresearch.com",
|
||||
"inference_base_url": "https://inference-api.nousresearch.com/v1",
|
||||
"agent_key": "minted-bearer",
|
||||
"agent_key_expires_at": "2099-01-01T00:00:00Z",
|
||||
"api_key": "minted-bearer",
|
||||
"base_url": "https://inference-api.nousresearch.com/v1",
|
||||
"expires_at": "2099-01-01T00:00:00Z",
|
||||
}
|
||||
|
||||
with patch(
|
||||
"hermes_cli.proxy.adapters.nous_portal.refresh_nous_oauth_from_state",
|
||||
"hermes_cli.proxy.adapters.nous_portal.resolve_nous_runtime_credentials",
|
||||
return_value=refreshed_state,
|
||||
) as mock_refresh:
|
||||
) as mock_resolve:
|
||||
adapter = NousPortalAdapter()
|
||||
cred = adapter.get_credential()
|
||||
|
||||
mock_refresh.assert_called_once()
|
||||
mock_resolve.assert_called_once()
|
||||
assert cred.bearer == "minted-bearer"
|
||||
assert cred.base_url == "https://inference-api.nousresearch.com/v1"
|
||||
assert cred.expires_at == "2099-01-01T00:00:00Z"
|
||||
assert cred.token_type == "Bearer"
|
||||
|
||||
# Verify state was persisted back
|
||||
stored = json.loads((tmp_path / "auth.json").read_text())
|
||||
assert stored["providers"]["nous"]["agent_key"] == "minted-bearer"
|
||||
|
||||
def test_nous_adapter_retry_credential_forces_legacy_mint(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
_write_auth_store(tmp_path, {
|
||||
"access_token": "jwt-access",
|
||||
"refresh_token": "refresh-tok",
|
||||
"client_id": "hermes-cli",
|
||||
"portal_base_url": "https://portal.nousresearch.com",
|
||||
"inference_base_url": "https://inference-api.nousresearch.com/v1",
|
||||
"agent_key": "jwt-access",
|
||||
})
|
||||
|
||||
refreshed_state = {
|
||||
"api_key": "legacy-bearer",
|
||||
"base_url": "https://inference-api.nousresearch.com/v1",
|
||||
"expires_at": "2099-01-01T00:00:00Z",
|
||||
}
|
||||
|
||||
with patch(
|
||||
"hermes_cli.proxy.adapters.nous_portal.resolve_nous_runtime_credentials",
|
||||
return_value=refreshed_state,
|
||||
) as mock_resolve:
|
||||
adapter = NousPortalAdapter()
|
||||
cred = adapter.get_retry_credential(
|
||||
failed_credential=UpstreamCredential(
|
||||
bearer="header.jwt.signature",
|
||||
base_url="https://inference-api.nousresearch.com/v1",
|
||||
),
|
||||
status_code=401,
|
||||
)
|
||||
|
||||
assert cred is not None
|
||||
assert cred.bearer == "legacy-bearer"
|
||||
assert mock_resolve.call_args.kwargs["inference_auth_mode"] == "legacy"
|
||||
|
||||
|
||||
def test_nous_adapter_retry_credential_skips_opaque_bearer(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
_write_auth_store(tmp_path, {
|
||||
"access_token": "jwt-access",
|
||||
"refresh_token": "refresh-tok",
|
||||
"agent_key": "opaque-bearer",
|
||||
})
|
||||
|
||||
with patch(
|
||||
"hermes_cli.proxy.adapters.nous_portal.resolve_nous_runtime_credentials",
|
||||
) as mock_resolve:
|
||||
adapter = NousPortalAdapter()
|
||||
cred = adapter.get_retry_credential(
|
||||
failed_credential=UpstreamCredential(
|
||||
bearer="opaque-bearer",
|
||||
base_url="https://inference-api.nousresearch.com/v1",
|
||||
),
|
||||
status_code=401,
|
||||
)
|
||||
|
||||
assert cred is None
|
||||
mock_resolve.assert_not_called()
|
||||
|
||||
|
||||
def test_nous_adapter_get_credential_raises_when_not_logged_in(tmp_path, monkeypatch):
|
||||
@@ -156,7 +207,7 @@ def test_nous_adapter_get_credential_raises_on_refresh_failure(tmp_path, monkeyp
|
||||
})
|
||||
|
||||
with patch(
|
||||
"hermes_cli.proxy.adapters.nous_portal.refresh_nous_oauth_from_state",
|
||||
"hermes_cli.proxy.adapters.nous_portal.resolve_nous_runtime_credentials",
|
||||
side_effect=RuntimeError("Refresh session has been revoked"),
|
||||
):
|
||||
adapter = NousPortalAdapter()
|
||||
@@ -164,6 +215,40 @@ def test_nous_adapter_get_credential_raises_on_refresh_failure(tmp_path, monkeyp
|
||||
adapter.get_credential()
|
||||
|
||||
|
||||
def test_nous_adapter_quarantines_terminal_refresh_failure(tmp_path, monkeypatch):
|
||||
from hermes_cli.auth import AuthError
|
||||
from agent.credential_pool import load_pool
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
_write_auth_store(tmp_path, {
|
||||
"access_token": "access-tok",
|
||||
"refresh_token": "refresh-tok",
|
||||
"agent_key": "stale-agent-key",
|
||||
})
|
||||
assert load_pool("nous").select() is not None
|
||||
|
||||
with patch(
|
||||
"hermes_cli.proxy.adapters.nous_portal.resolve_nous_runtime_credentials",
|
||||
side_effect=AuthError(
|
||||
"Refresh session has been revoked",
|
||||
provider="nous",
|
||||
code="invalid_grant",
|
||||
relogin_required=True,
|
||||
),
|
||||
):
|
||||
adapter = NousPortalAdapter()
|
||||
with pytest.raises(RuntimeError, match="Refresh session has been revoked"):
|
||||
adapter.get_credential()
|
||||
|
||||
stored = json.loads((tmp_path / "auth.json").read_text())
|
||||
nous_state = stored["providers"]["nous"]
|
||||
assert not nous_state.get("refresh_token")
|
||||
assert not nous_state.get("access_token")
|
||||
assert not nous_state.get("agent_key")
|
||||
assert nous_state["last_auth_error"]["code"] == "invalid_grant"
|
||||
assert stored.get("credential_pool", {}).get("nous") == []
|
||||
|
||||
|
||||
def test_nous_adapter_get_credential_raises_when_no_agent_key_returned(tmp_path, monkeypatch):
|
||||
"""If the refresh helper succeeds but produces no agent_key, we surface a clear error."""
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
@@ -173,7 +258,7 @@ def test_nous_adapter_get_credential_raises_when_no_agent_key_returned(tmp_path,
|
||||
})
|
||||
|
||||
with patch(
|
||||
"hermes_cli.proxy.adapters.nous_portal.refresh_nous_oauth_from_state",
|
||||
"hermes_cli.proxy.adapters.nous_portal.resolve_nous_runtime_credentials",
|
||||
return_value={"access_token": "a", "refresh_token": "r"},
|
||||
):
|
||||
adapter = NousPortalAdapter()
|
||||
@@ -194,7 +279,7 @@ def test_nous_adapter_concurrent_refresh_serialized(tmp_path, monkeypatch):
|
||||
counter = [0]
|
||||
counter_lock = threading.Lock()
|
||||
|
||||
def serializing_refresh(state, **kwargs):
|
||||
def serializing_refresh(**kwargs):
|
||||
# If another thread is already inside refresh, the lock is broken.
|
||||
if in_flight.is_set():
|
||||
overlap_detected.set()
|
||||
@@ -208,10 +293,9 @@ def test_nous_adapter_concurrent_refresh_serialized(tmp_path, monkeypatch):
|
||||
counter[0] += 1
|
||||
idx = counter[0]
|
||||
return {
|
||||
**state,
|
||||
"agent_key": f"key-{idx}",
|
||||
"agent_key_expires_at": "2099-01-01T00:00:00Z",
|
||||
"inference_base_url": "https://inference-api.nousresearch.com/v1",
|
||||
"api_key": f"key-{idx}",
|
||||
"expires_at": "2099-01-01T00:00:00Z",
|
||||
"base_url": "https://inference-api.nousresearch.com/v1",
|
||||
}
|
||||
finally:
|
||||
in_flight.clear()
|
||||
@@ -227,7 +311,7 @@ def test_nous_adapter_concurrent_refresh_serialized(tmp_path, monkeypatch):
|
||||
errors.append(exc)
|
||||
|
||||
with patch(
|
||||
"hermes_cli.proxy.adapters.nous_portal.refresh_nous_oauth_from_state",
|
||||
"hermes_cli.proxy.adapters.nous_portal.resolve_nous_runtime_credentials",
|
||||
side_effect=serializing_refresh,
|
||||
):
|
||||
threads = [threading.Thread(target=worker) for _ in range(3)]
|
||||
@@ -260,12 +344,15 @@ class FakeAdapter(UpstreamAdapter):
|
||||
"""A test adapter that returns a fixed credential without touching disk."""
|
||||
|
||||
def __init__(self, base_url: str, bearer: str = "test-bearer",
|
||||
allowed=None, raise_on_credential=False):
|
||||
allowed=None, raise_on_credential=False,
|
||||
retry_bearer: str | None = None):
|
||||
self._base_url = base_url
|
||||
self._bearer = bearer
|
||||
self._allowed = frozenset(allowed or ["/chat/completions"])
|
||||
self._raise = raise_on_credential
|
||||
self._retry_bearer = retry_bearer
|
||||
self.calls = 0
|
||||
self.retry_calls = 0
|
||||
|
||||
@property
|
||||
def name(self): return "fake"
|
||||
@@ -287,6 +374,17 @@ class FakeAdapter(UpstreamAdapter):
|
||||
expires_at="2099-01-01T00:00:00Z",
|
||||
)
|
||||
|
||||
def get_retry_credential(self, *, failed_credential, status_code):
|
||||
_ = failed_credential
|
||||
self.retry_calls += 1
|
||||
if status_code != 401 or not self._retry_bearer:
|
||||
return None
|
||||
return UpstreamCredential(
|
||||
bearer=self._retry_bearer,
|
||||
base_url=self._base_url,
|
||||
expires_at="2099-01-01T00:00:00Z",
|
||||
)
|
||||
|
||||
|
||||
async def _start_runner(app: "web.Application"):
|
||||
"""Spin up an aiohttp app on an ephemeral localhost port. Returns (runner, base_url)."""
|
||||
@@ -327,6 +425,25 @@ def _build_fake_upstream(captured: Dict[str, Any]) -> "web.Application":
|
||||
return app
|
||||
|
||||
|
||||
def _build_retrying_fake_upstream(captured: Dict[str, Any]) -> "web.Application":
|
||||
async def maybe_unauthorized(request):
|
||||
body = await request.read()
|
||||
auth = request.headers.get("Authorization")
|
||||
captured["requests"].append({
|
||||
"method": request.method,
|
||||
"path": request.path,
|
||||
"auth": auth,
|
||||
"body": body.decode("utf-8") if body else "",
|
||||
})
|
||||
if auth == "Bearer jwt-bearer":
|
||||
return web.json_response({"error": "bad token"}, status=401)
|
||||
return web.json_response({"ok": True})
|
||||
|
||||
app = web.Application()
|
||||
app.router.add_route("*", "/v1/chat/completions", maybe_unauthorized)
|
||||
return app
|
||||
|
||||
|
||||
def test_server_forwards_chat_completions():
|
||||
async def run():
|
||||
captured: Dict[str, Any] = {"requests": []}
|
||||
@@ -357,6 +474,41 @@ def test_server_forwards_chat_completions():
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_server_retries_once_with_adapter_retry_credential_on_401():
|
||||
async def run():
|
||||
captured: Dict[str, Any] = {"requests": []}
|
||||
upstream_runner, upstream_base = await _start_runner(
|
||||
_build_retrying_fake_upstream(captured)
|
||||
)
|
||||
adapter = FakeAdapter(
|
||||
f"{upstream_base}/v1",
|
||||
bearer="jwt-bearer",
|
||||
retry_bearer="legacy-bearer",
|
||||
)
|
||||
proxy_runner, proxy_base = await _start_runner(create_app(adapter))
|
||||
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(
|
||||
f"{proxy_base}/v1/chat/completions",
|
||||
json={"model": "Hermes-4-70B"},
|
||||
) as resp:
|
||||
assert resp.status == 200
|
||||
data = await resp.json()
|
||||
assert data["ok"] is True
|
||||
|
||||
assert adapter.retry_calls == 1
|
||||
assert [req["auth"] for req in captured["requests"]] == [
|
||||
"Bearer jwt-bearer",
|
||||
"Bearer legacy-bearer",
|
||||
]
|
||||
finally:
|
||||
await proxy_runner.cleanup()
|
||||
await upstream_runner.cleanup()
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_server_rejects_disallowed_path():
|
||||
async def run():
|
||||
adapter = FakeAdapter("http://unused.example/v1", allowed=["/chat/completions"])
|
||||
|
||||
@@ -29,6 +29,7 @@ def test_show_status_termux_gateway_section_skips_systemctl(monkeypatch, capsys,
|
||||
monkeypatch.setattr(status_mod, "provider_label", lambda provider: "OpenAI Codex", raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_nous_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_codex_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda exclude_pids=None: [], raising=False)
|
||||
|
||||
def _unexpected_systemctl(*args, **kwargs):
|
||||
@@ -70,6 +71,7 @@ def test_show_status_reports_nous_auth_error(monkeypatch, capsys, tmp_path):
|
||||
)
|
||||
monkeypatch.setattr(auth_mod, "get_codex_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_qwen_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda exclude_pids=None: [], raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
@@ -96,6 +98,7 @@ def test_show_status_reports_vercel_backend_contract(monkeypatch, capsys, tmp_pa
|
||||
monkeypatch.setattr(auth_mod, "get_nous_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_codex_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_qwen_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda exclude_pids=None: [], raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
@@ -109,3 +112,223 @@ def test_show_status_reports_vercel_backend_contract(monkeypatch, capsys, tmp_pa
|
||||
assert "oidc-token" not in output
|
||||
assert "snapshot filesystem" in output
|
||||
assert "live processes do not survive" in output
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers shared by xAI OAuth status tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _base_xai_mocks(monkeypatch, tmp_path):
|
||||
"""Set up the minimal environment for show_status, returning status_mod."""
|
||||
from hermes_cli import status as status_mod
|
||||
import hermes_cli.auth as auth_mod
|
||||
import hermes_cli.gateway as gateway_mod
|
||||
|
||||
monkeypatch.setattr(status_mod, "get_env_path", lambda: tmp_path / ".env", raising=False)
|
||||
monkeypatch.setattr(status_mod, "get_hermes_home", lambda: tmp_path, raising=False)
|
||||
monkeypatch.setattr(status_mod, "load_config", lambda: {"model": "gpt-5.4"}, raising=False)
|
||||
monkeypatch.setattr(status_mod, "resolve_requested_provider", lambda requested=None: "openai-codex", raising=False)
|
||||
monkeypatch.setattr(status_mod, "resolve_provider", lambda requested=None, **kwargs: "openai-codex", raising=False)
|
||||
monkeypatch.setattr(status_mod, "provider_label", lambda provider: "OpenAI Codex", raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_nous_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_codex_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_qwen_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "get_minimax_oauth_auth_status", lambda: {}, raising=False)
|
||||
monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda exclude_pids=None: [], raising=False)
|
||||
return status_mod
|
||||
|
||||
|
||||
class TestShowStatusXaiOAuth:
|
||||
"""xAI OAuth row in hermes status."""
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Logged-in branch
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_logged_in_shows_check_mark_and_label(self, monkeypatch, capsys, tmp_path):
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {"logged_in": True, "auth_store": "/a/auth.json"},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "xAI OAuth" in out
|
||||
# The logged-in label must appear; the "not logged in" label must not
|
||||
assert "✓" in out or "logged in" in out
|
||||
assert "not logged in" not in out.split("xAI OAuth", 1)[1].split("\n")[0]
|
||||
|
||||
def test_logged_in_shows_auth_store(self, monkeypatch, capsys, tmp_path):
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {"logged_in": True, "auth_store": "/home/u/.hermes/auth.json"},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "Auth file: /home/u/.hermes/auth.json" in out
|
||||
|
||||
def test_logged_in_shows_last_refresh(self, monkeypatch, capsys, tmp_path):
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {
|
||||
"logged_in": True,
|
||||
"auth_store": "/a/auth.json",
|
||||
"last_refresh": "2026-05-17T10:00:00+00:00",
|
||||
},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "Refreshed:" in out
|
||||
|
||||
def test_logged_in_does_not_show_error_line(self, monkeypatch, capsys, tmp_path):
|
||||
"""Error field must be suppressed when logged_in is True."""
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {
|
||||
"logged_in": True,
|
||||
"auth_store": "/a/auth.json",
|
||||
"error": "stale-error-must-not-appear",
|
||||
},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
xai_section = out.split("xAI OAuth", 1)[1]
|
||||
assert "stale-error-must-not-appear" not in xai_section
|
||||
|
||||
def test_no_auth_store_line_when_field_absent(self, monkeypatch, capsys, tmp_path):
|
||||
"""Auth file line must not appear when auth_store is missing."""
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {"logged_in": True},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
xai_section = out.split("xAI OAuth", 1)[1].split("◆", 1)[0]
|
||||
assert "Auth file:" not in xai_section
|
||||
|
||||
def test_no_refreshed_line_when_last_refresh_absent(self, monkeypatch, capsys, tmp_path):
|
||||
"""Refreshed line must not appear when last_refresh is not present."""
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {"logged_in": True, "auth_store": "/a/auth.json"},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
xai_section = out.split("xAI OAuth", 1)[1].split("◆", 1)[0]
|
||||
assert "Refreshed:" not in xai_section
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Not-logged-in branch
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_not_logged_in_shows_login_command(self, monkeypatch, capsys, tmp_path):
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {"logged_in": False, "error": "no credentials"},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "not logged in (run: hermes auth add xai-oauth)" in out
|
||||
|
||||
def test_not_logged_in_shows_error(self, monkeypatch, capsys, tmp_path):
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {"logged_in": False, "error": "Token has expired"},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "Error: Token has expired" in out
|
||||
|
||||
def test_not_logged_in_omits_error_line_when_error_absent(self, monkeypatch, capsys, tmp_path):
|
||||
"""No Error: line when not logged in but error key is missing."""
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: {"logged_in": False},
|
||||
raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
xai_section = out.split("xAI OAuth", 1)[1].split("◆", 1)[0]
|
||||
assert "Error:" not in xai_section
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Resilience: import failure and runtime exception
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_import_failure_does_not_crash_show_status(self, monkeypatch, capsys, tmp_path):
|
||||
"""show_status must complete even when get_xai_oauth_auth_status cannot be imported."""
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.delattr(auth_mod, "get_xai_oauth_auth_status", raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "◆ Auth Providers" in out
|
||||
|
||||
def test_import_failure_does_not_break_other_oauth_providers(self, monkeypatch, capsys, tmp_path):
|
||||
"""Nous/Codex/MiniMax rows must still appear when xAI import fails."""
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_nous_auth_status",
|
||||
lambda: {"logged_in": True}, raising=False)
|
||||
monkeypatch.delattr(auth_mod, "get_xai_oauth_auth_status", raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "Nous Portal" in out
|
||||
assert "MiniMax OAuth" in out
|
||||
|
||||
def test_status_function_exception_does_not_crash(self, monkeypatch, capsys, tmp_path):
|
||||
"""show_status must not propagate an exception raised by get_xai_oauth_auth_status."""
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
|
||||
def _raises():
|
||||
raise RuntimeError("backend unreachable")
|
||||
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status", _raises, raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "◆ Auth Providers" in out
|
||||
|
||||
def test_status_function_returns_none_does_not_crash(self, monkeypatch, capsys, tmp_path):
|
||||
"""get_xai_oauth_auth_status returning None must be handled gracefully."""
|
||||
import hermes_cli.auth as auth_mod
|
||||
status_mod = _base_xai_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status",
|
||||
lambda: None, raising=False)
|
||||
|
||||
status_mod.show_status(SimpleNamespace(all=False, deep=False))
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "xAI OAuth" in out
|
||||
assert "not logged in (run: hermes auth add xai-oauth)" in out
|
||||
|
||||
@@ -125,6 +125,62 @@ def test_get_platform_tools_homeassistant_toolset_off_for_cron_when_hass_token_m
|
||||
assert "homeassistant" not in cron_enabled
|
||||
|
||||
|
||||
def test_get_platform_tools_x_search_auto_enabled_when_xai_oauth_present(monkeypatch):
|
||||
"""x_search toolset auto-enables across platforms when xAI Grok OAuth
|
||||
tokens are present, mirroring the HASS_TOKEN → homeassistant rule.
|
||||
|
||||
The user already authenticated via SuperGrok OAuth; they shouldn't have
|
||||
to also click through `hermes tools` → X (Twitter) Search to flip the
|
||||
toolset on. Tool's check_fn still gates schema registration if creds
|
||||
later go missing.
|
||||
"""
|
||||
monkeypatch.delenv("XAI_API_KEY", raising=False)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.tools_config._xai_credentials_present", lambda: True
|
||||
)
|
||||
|
||||
for plat in ("cli", "cron", "telegram"):
|
||||
enabled = _get_platform_tools({}, plat)
|
||||
assert "x_search" in enabled, f"x_search missing for {plat}"
|
||||
|
||||
|
||||
def test_get_platform_tools_x_search_auto_enabled_when_xai_api_key_present(monkeypatch):
|
||||
"""x_search toolset auto-enables when XAI_API_KEY is set, even without
|
||||
OAuth tokens — the API-key path is a supported credential source."""
|
||||
monkeypatch.setenv("XAI_API_KEY", "fake-xai-key")
|
||||
|
||||
cli_enabled = _get_platform_tools({}, "cli")
|
||||
assert "x_search" in cli_enabled
|
||||
|
||||
|
||||
def test_get_platform_tools_x_search_off_when_no_xai_credentials(monkeypatch):
|
||||
"""Without any xAI credentials, x_search stays off — preserves the
|
||||
"don't ship the schema to users who can't use it" default."""
|
||||
monkeypatch.delenv("XAI_API_KEY", raising=False)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.tools_config._xai_credentials_present", lambda: False
|
||||
)
|
||||
|
||||
cli_enabled = _get_platform_tools({}, "cli")
|
||||
assert "x_search" not in cli_enabled
|
||||
|
||||
|
||||
def test_get_platform_tools_x_search_respects_explicit_config(monkeypatch):
|
||||
"""Once the user has saved an explicit toolset list via `hermes tools`,
|
||||
that list is authoritative — x_search auto-enable does NOT fire even
|
||||
when xAI creds exist. The saved list represents deliberate choices."""
|
||||
monkeypatch.delenv("XAI_API_KEY", raising=False)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.tools_config._xai_credentials_present", lambda: True
|
||||
)
|
||||
|
||||
# User explicitly opted into spotify but not x_search via `hermes tools`.
|
||||
config = {"platform_toolsets": {"cli": ["hermes-cli", "spotify"]}}
|
||||
enabled = _get_platform_tools(config, "cli")
|
||||
assert "x_search" not in enabled
|
||||
assert "spotify" in enabled
|
||||
|
||||
|
||||
def test_get_platform_tools_expands_composite_when_mixed_with_configurable():
|
||||
"""``[hermes-cli, spotify]`` (composite + configurable) must keep the full
|
||||
``hermes-cli`` toolset alongside the explicit Spotify opt-in. The
|
||||
@@ -989,3 +1045,27 @@ def test_reconfigure_browser_provider_overwrites_stale_use_gateway():
|
||||
provider = {"name": "Browserbase", "browser_provider": "browserbase", "env_vars": []}
|
||||
_reconfigure_provider(provider, config)
|
||||
assert config["browser"]["use_gateway"] is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider_name,post_setup_key", [
|
||||
("Camofox", "camofox"),
|
||||
])
|
||||
def test_reconfigure_provider_runs_post_setup_for_env_var_providers(
|
||||
monkeypatch, provider_name, post_setup_key
|
||||
):
|
||||
"""_reconfigure_provider() must call _run_post_setup() for providers that have
|
||||
both env_vars and post_setup — parity with _configure_provider() line 2286."""
|
||||
called = []
|
||||
monkeypatch.setattr("hermes_cli.tools_config._run_post_setup", lambda key: called.append(key))
|
||||
monkeypatch.setattr("hermes_cli.tools_config.get_env_value", lambda k: None)
|
||||
monkeypatch.setattr("hermes_cli.tools_config._prompt", lambda *a, **kw: "")
|
||||
monkeypatch.setattr("hermes_cli.tools_config.save_env_value", lambda k, v: None)
|
||||
|
||||
provider = next(
|
||||
p
|
||||
for p in TOOL_CATEGORIES["browser"]["providers"]
|
||||
if p["name"] == provider_name
|
||||
)
|
||||
_reconfigure_provider(provider, {})
|
||||
|
||||
assert called == [post_setup_key]
|
||||
|
||||
@@ -237,7 +237,7 @@ class TestKillStaleDashboardPosix:
|
||||
sent.append((pid, sig))
|
||||
# Simulate stubborn process: probe (sig 0) always succeeds,
|
||||
# SIGTERM does nothing, SIGKILL is where it "dies".
|
||||
if sig in (_signal.SIGTERM, 0, _signal.SIGKILL):
|
||||
if sig in {_signal.SIGTERM, 0, _signal.SIGKILL}:
|
||||
return
|
||||
# Any other signal — also fine.
|
||||
|
||||
|
||||
@@ -19,11 +19,12 @@ The fix:
|
||||
|
||||
These tests pin the corrected behavior.
|
||||
"""
|
||||
import asyncio
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
import httpx
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from hermes_cli.web_server import _SESSION_TOKEN, app
|
||||
@@ -32,6 +33,32 @@ client = TestClient(app)
|
||||
HEADERS = {"X-Hermes-Session-Token": _SESSION_TOKEN}
|
||||
|
||||
|
||||
def _fake_nous_device_data():
|
||||
return {
|
||||
"device_code": "device-code",
|
||||
"user_code": "NOUS-1234",
|
||||
"verification_uri": "https://portal.nousresearch.com/device",
|
||||
"verification_uri_complete": (
|
||||
"https://portal.nousresearch.com/device?user_code=NOUS-1234"
|
||||
),
|
||||
"expires_in": 600,
|
||||
"interval": 5,
|
||||
}
|
||||
|
||||
|
||||
def _invoke_scope_refusal():
|
||||
request = httpx.Request("POST", "https://portal.nousresearch.com/oauth/device/code")
|
||||
response = httpx.Response(
|
||||
400,
|
||||
json={
|
||||
"error": "invalid_scope",
|
||||
"error_description": "unsupported scope inference:invoke",
|
||||
},
|
||||
request=request,
|
||||
)
|
||||
return httpx.HTTPStatusError("invalid scope", request=request, response=response)
|
||||
|
||||
|
||||
def test_minimax_login_does_not_launch_anthropic_flow():
|
||||
"""Click 'Login' on MiniMax → MUST NOT return claude.ai auth_url."""
|
||||
fake_user_code_resp = {
|
||||
@@ -48,6 +75,9 @@ def test_minimax_login_does_not_launch_anthropic_flow():
|
||||
), patch(
|
||||
"hermes_cli.auth._minimax_pkce_pair",
|
||||
return_value=("verifier-stub", "challenge-stub", "stub-state"),
|
||||
), patch(
|
||||
"hermes_cli.web_server._minimax_poller",
|
||||
return_value=None,
|
||||
):
|
||||
resp = client.post(
|
||||
"/api/providers/oauth/minimax-oauth/start",
|
||||
@@ -69,6 +99,113 @@ def test_minimax_login_does_not_launch_anthropic_flow():
|
||||
assert body["expires_in"] == 600
|
||||
|
||||
|
||||
def test_nous_dashboard_device_flow_honors_legacy_scope_override(monkeypatch):
|
||||
from hermes_cli import auth as auth_mod
|
||||
from hermes_cli import web_server as ws
|
||||
|
||||
requested_scopes = []
|
||||
|
||||
def fake_request_device_code(**kwargs):
|
||||
requested_scopes.append(kwargs["scope"])
|
||||
return _fake_nous_device_data()
|
||||
|
||||
monkeypatch.setenv(auth_mod.NOUS_LEGACY_SESSION_KEYS_ENV, "true")
|
||||
monkeypatch.setattr(auth_mod, "_request_device_code", fake_request_device_code)
|
||||
monkeypatch.setattr(ws, "_nous_poller", lambda sid: None)
|
||||
|
||||
result = asyncio.run(ws._start_device_code_flow("nous"))
|
||||
try:
|
||||
assert requested_scopes == [auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE]
|
||||
assert result["flow"] == "device_code"
|
||||
assert result["user_code"] == "NOUS-1234"
|
||||
assert (
|
||||
ws._oauth_sessions[result["session_id"]]["scope"]
|
||||
== auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE
|
||||
)
|
||||
finally:
|
||||
ws._oauth_sessions.pop(result["session_id"], None)
|
||||
|
||||
|
||||
def test_nous_dashboard_device_flow_retries_legacy_scope_on_invoke_refusal(monkeypatch):
|
||||
from hermes_cli import auth as auth_mod
|
||||
from hermes_cli import web_server as ws
|
||||
|
||||
requested_scopes = []
|
||||
|
||||
def fake_request_device_code(**kwargs):
|
||||
requested_scopes.append(kwargs["scope"])
|
||||
if len(requested_scopes) == 1:
|
||||
raise _invoke_scope_refusal()
|
||||
return _fake_nous_device_data()
|
||||
|
||||
monkeypatch.delenv(auth_mod.NOUS_LEGACY_SESSION_KEYS_ENV, raising=False)
|
||||
monkeypatch.setattr(auth_mod, "_request_device_code", fake_request_device_code)
|
||||
monkeypatch.setattr(ws, "_nous_poller", lambda sid: None)
|
||||
|
||||
result = asyncio.run(ws._start_device_code_flow("nous"))
|
||||
try:
|
||||
assert requested_scopes == [
|
||||
auth_mod.DEFAULT_NOUS_SCOPE,
|
||||
auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
|
||||
]
|
||||
assert (
|
||||
ws._oauth_sessions[result["session_id"]]["scope"]
|
||||
== auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE
|
||||
)
|
||||
finally:
|
||||
ws._oauth_sessions.pop(result["session_id"], None)
|
||||
|
||||
|
||||
def test_nous_dashboard_poller_preserves_effective_scope_when_token_omits_scope(monkeypatch):
|
||||
from hermes_cli import auth as auth_mod
|
||||
from hermes_cli import web_server as ws
|
||||
|
||||
session_id = "nous-effective-scope-test"
|
||||
ws._oauth_sessions[session_id] = {
|
||||
"session_id": session_id,
|
||||
"provider": "nous",
|
||||
"flow": "device_code",
|
||||
"created_at": time.time(),
|
||||
"status": "pending",
|
||||
"error_message": None,
|
||||
"portal_base_url": "https://portal.nousresearch.com",
|
||||
"client_id": "hermes-cli",
|
||||
"device_code": "device-code",
|
||||
"interval": 5,
|
||||
"expires_at": time.time() + 600,
|
||||
"scope": auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
|
||||
}
|
||||
captured_state = {}
|
||||
|
||||
def fake_refresh_nous_oauth_from_state(state, **kwargs):
|
||||
captured_state.update(state)
|
||||
return {**state, "agent_key": "legacy-agent-key"}
|
||||
|
||||
monkeypatch.setattr(
|
||||
auth_mod,
|
||||
"_poll_for_token",
|
||||
lambda **kwargs: {
|
||||
"access_token": "access-token",
|
||||
"refresh_token": "refresh-token",
|
||||
"expires_in": 3600,
|
||||
"token_type": "Bearer",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
auth_mod,
|
||||
"refresh_nous_oauth_from_state",
|
||||
fake_refresh_nous_oauth_from_state,
|
||||
)
|
||||
monkeypatch.setattr(auth_mod, "persist_nous_credentials", lambda state: None)
|
||||
|
||||
try:
|
||||
ws._nous_poller(session_id)
|
||||
assert captured_state["scope"] == auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE
|
||||
assert ws._oauth_sessions[session_id]["status"] == "approved"
|
||||
finally:
|
||||
ws._oauth_sessions.pop(session_id, None)
|
||||
|
||||
|
||||
def test_minimax_dashboard_poller_accepts_absolute_ms_expired_in():
|
||||
"""Dashboard MiniMax completion must accept unix-ms token expiry values."""
|
||||
from hermes_cli import web_server as ws
|
||||
|
||||
@@ -449,7 +449,7 @@ class TestWebServerEndpoints:
|
||||
resp = self.client.get("/api/auth/session-token")
|
||||
# The endpoint is gone — the catch-all SPA route serves index.html
|
||||
# or the middleware returns 401 for unauthenticated /api/ paths.
|
||||
assert resp.status_code in (200, 404)
|
||||
assert resp.status_code in {200, 404}
|
||||
# Either way, it must NOT return the token as JSON
|
||||
try:
|
||||
data = resp.json()
|
||||
@@ -476,7 +476,7 @@ class TestWebServerEndpoints:
|
||||
# %2e%2e = ..
|
||||
resp = self.client.get("/%2e%2e/%2e%2e/etc/passwd")
|
||||
# Should return 200 with index.html (SPA fallback), not the actual file
|
||||
assert resp.status_code in (200, 404)
|
||||
assert resp.status_code in {200, 404}
|
||||
if resp.status_code == 200:
|
||||
# Should be the SPA fallback, not the system file
|
||||
assert "root:" not in resp.text
|
||||
@@ -484,7 +484,7 @@ class TestWebServerEndpoints:
|
||||
def test_path_traversal_dotdot_blocked(self):
|
||||
"""Direct .. path traversal via encoded sequences."""
|
||||
resp = self.client.get("/%2e%2e/hermes_cli/web_server.py")
|
||||
assert resp.status_code in (200, 404)
|
||||
assert resp.status_code in {200, 404}
|
||||
if resp.status_code == 200:
|
||||
assert "FastAPI" not in resp.text # Should not serve the actual source
|
||||
|
||||
@@ -678,7 +678,7 @@ class TestConfigRoundTrip:
|
||||
if val is None:
|
||||
continue # not set in user config — fine
|
||||
expected = entry["type"]
|
||||
if expected in ("string", "select") and not isinstance(val, str):
|
||||
if expected in {"string", "select"} and not isinstance(val, str):
|
||||
mismatches.append(f"{key}: expected str, got {type(val).__name__}")
|
||||
elif expected == "number" and not isinstance(val, (int, float)):
|
||||
mismatches.append(f"{key}: expected number, got {type(val).__name__}")
|
||||
@@ -1175,7 +1175,7 @@ class TestNewEndpoints:
|
||||
"""GET /api/auth/session-token no longer exists."""
|
||||
resp = self.client.get("/api/auth/session-token")
|
||||
# Should not return a JSON token object
|
||||
assert resp.status_code in (200, 404)
|
||||
assert resp.status_code in {200, 404}
|
||||
try:
|
||||
data = resp.json()
|
||||
assert "token" not in data
|
||||
|
||||
@@ -0,0 +1,359 @@
|
||||
"""Regression coverage for xAI OAuth PKCE token exchange (issue #26990).
|
||||
|
||||
Issue [#26990] reported that ``hermes auth add xai-oauth`` succeeds at the
|
||||
browser-side authorize step but fails at the token endpoint with
|
||||
``code_challenge is required`` — the symptom of an OAuth server that
|
||||
re-validates PKCE at the token step instead of relying purely on
|
||||
state captured during the authorize redirect.
|
||||
|
||||
The fix in ``hermes_cli/auth.py`` extracts the token POST into
|
||||
:func:`_xai_oauth_exchange_code_for_tokens` and:
|
||||
|
||||
* Sends ``code_verifier`` (RFC 7636 §4.5 requirement).
|
||||
* **Also** echoes ``code_challenge`` and ``code_challenge_method``
|
||||
in the request body as defense-in-depth — strictly compliant
|
||||
servers ignore extras at the token endpoint, but xAI's server
|
||||
needs them.
|
||||
* Refuses to fire the POST locally when ``code_verifier`` is empty
|
||||
(avoids leaking the auth code to a server that can't redeem it).
|
||||
* Surfaces the HTTP status code prominently in the error message so
|
||||
users / maintainers can tell a 400 (bad request) from a 403
|
||||
(entitlement denied) at a glance.
|
||||
|
||||
These tests pin all three behaviors so the fix can't silently regress.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List
|
||||
from urllib.parse import parse_qs
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from hermes_cli.auth import (
|
||||
AuthError,
|
||||
XAI_OAUTH_CLIENT_ID,
|
||||
_xai_oauth_exchange_code_for_tokens,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# httpx.post recorder
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _PostRecorder:
|
||||
"""Capture every ``httpx.post`` call without touching the network."""
|
||||
|
||||
def __init__(self, response: httpx.Response) -> None:
|
||||
self.response = response
|
||||
self.calls: List[Dict[str, Any]] = []
|
||||
|
||||
def __call__(self, url, *, headers=None, data=None, timeout=None, **kw):
|
||||
self.calls.append(
|
||||
{"url": url, "headers": headers or {}, "data": data or {},
|
||||
"timeout": timeout, "extra": kw}
|
||||
)
|
||||
return self.response
|
||||
|
||||
|
||||
def _ok_response(payload: dict) -> httpx.Response:
|
||||
return httpx.Response(200, json=payload)
|
||||
|
||||
|
||||
def _err_response(status: int, body: str) -> httpx.Response:
|
||||
return httpx.Response(status, text=body)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def post_recorder(monkeypatch):
|
||||
"""Default: 200 response with a full xAI token payload."""
|
||||
recorder = _PostRecorder(
|
||||
_ok_response(
|
||||
{
|
||||
"access_token": "AT-fresh",
|
||||
"refresh_token": "RT-fresh",
|
||||
"id_token": "ID",
|
||||
"expires_in": 3600,
|
||||
"token_type": "Bearer",
|
||||
}
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr("hermes_cli.auth.httpx.post", recorder)
|
||||
return recorder
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Core contract: which fields go on the wire?
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_token_exchange_includes_code_verifier(post_recorder):
|
||||
"""RFC 7636 §4.5 — ``code_verifier`` MUST be sent."""
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="theVerifier_43_to_128_chars_____________________",
|
||||
code_challenge="aBcDeF",
|
||||
)
|
||||
sent = post_recorder.calls[-1]["data"]
|
||||
assert sent["code_verifier"] == "theVerifier_43_to_128_chars_____________________"
|
||||
|
||||
|
||||
def test_token_exchange_also_echoes_code_challenge_for_xai(post_recorder):
|
||||
"""Defense-in-depth for #26990 — xAI re-validates the challenge
|
||||
at the token endpoint, not just at authorize. Without this echo
|
||||
we get ``code_challenge is required`` even though we send a valid
|
||||
``code_verifier``."""
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="aBcDeF",
|
||||
)
|
||||
sent = post_recorder.calls[-1]["data"]
|
||||
assert sent["code_challenge"] == "aBcDeF"
|
||||
assert sent["code_challenge_method"] == "S256"
|
||||
|
||||
|
||||
def test_token_exchange_uses_correct_grant_and_client(post_recorder):
|
||||
"""Lock the static fields too — a future refactor must not flip
|
||||
these to ``client_credentials`` or drop ``client_id``."""
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
)
|
||||
sent = post_recorder.calls[-1]["data"]
|
||||
assert sent["grant_type"] == "authorization_code"
|
||||
assert sent["code"] == "AUTHCODE"
|
||||
assert sent["redirect_uri"] == "http://127.0.0.1:56121/callback"
|
||||
assert sent["client_id"] == XAI_OAUTH_CLIENT_ID
|
||||
|
||||
|
||||
def test_token_exchange_uses_form_urlencoded_content_type(post_recorder):
|
||||
"""xAI's token endpoint expects ``application/x-www-form-urlencoded``."""
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
)
|
||||
headers = post_recorder.calls[-1]["headers"]
|
||||
assert headers["Content-Type"] == "application/x-www-form-urlencoded"
|
||||
assert headers["Accept"] == "application/json"
|
||||
|
||||
|
||||
def test_token_exchange_targets_the_supplied_endpoint(post_recorder):
|
||||
"""Some test fixtures sniff the discovered token endpoint dynamically.
|
||||
We must POST to the URL the caller passed, not a hard-coded constant."""
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/some/other/token/path",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
)
|
||||
assert post_recorder.calls[-1]["url"] == "https://auth.x.ai/some/other/token/path"
|
||||
|
||||
|
||||
def test_token_exchange_passes_timeout_through(post_recorder):
|
||||
"""Operators on slow networks pass a higher ``timeout_seconds``;
|
||||
the helper must forward it (and bump the floor to 20s)."""
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
timeout_seconds=45.0,
|
||||
)
|
||||
assert post_recorder.calls[-1]["timeout"] == 45.0
|
||||
|
||||
|
||||
def test_token_exchange_floor_timeout_is_20s(post_recorder):
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
timeout_seconds=2.0,
|
||||
)
|
||||
assert post_recorder.calls[-1]["timeout"] == 20.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sanity guard: refuse to POST with an empty code_verifier
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_empty_code_verifier_raises_without_posting(post_recorder):
|
||||
"""If ``code_verifier`` is somehow lost upstream, we must refuse to
|
||||
send the request — leaking an authorization code to xAI without a
|
||||
verifier is worse than failing locally with an actionable error."""
|
||||
with pytest.raises(AuthError) as exc_info:
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="",
|
||||
code_challenge="c" * 43,
|
||||
)
|
||||
assert exc_info.value.code == "xai_pkce_verifier_missing"
|
||||
assert "26990" in str(exc_info.value)
|
||||
# And critically: nothing was sent.
|
||||
assert post_recorder.calls == []
|
||||
|
||||
|
||||
def test_missing_code_challenge_omits_echo_but_still_sends_verifier(post_recorder):
|
||||
"""``code_challenge`` is defensive — if a caller doesn't have it
|
||||
handy, we must still send the standards-compliant request rather
|
||||
than refusing. This keeps RFC-compliant servers happy."""
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="",
|
||||
)
|
||||
sent = post_recorder.calls[-1]["data"]
|
||||
assert sent["code_verifier"] == "v" * 64
|
||||
assert "code_challenge" not in sent
|
||||
assert "code_challenge_method" not in sent
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Error surfacing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_non_200_response_surfaces_status_and_body(monkeypatch):
|
||||
"""When xAI returns a 4xx, the operator needs both the HTTP status
|
||||
code (to tell 400 from 401 from 403 at a glance) and the response
|
||||
body (the actual server-side reason)."""
|
||||
recorder = _PostRecorder(
|
||||
_err_response(400, '{"error":"invalid_grant","error_description":"code_challenge is required"}')
|
||||
)
|
||||
monkeypatch.setattr("hermes_cli.auth.httpx.post", recorder)
|
||||
with pytest.raises(AuthError) as exc_info:
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
)
|
||||
msg = str(exc_info.value)
|
||||
assert "HTTP 400" in msg, (
|
||||
"Status code must be in the error so callers can disambiguate "
|
||||
"tier-denied (403) from bad-request (400) without inspecting "
|
||||
"exc.code."
|
||||
)
|
||||
assert "code_challenge is required" in msg
|
||||
assert exc_info.value.code == "xai_token_exchange_failed"
|
||||
|
||||
|
||||
def test_transport_error_wraps_as_auth_error(monkeypatch):
|
||||
"""A connection failure must come back as ``AuthError`` so the
|
||||
surrounding ``format_auth_error`` UI mapping fires correctly."""
|
||||
|
||||
def _boom(*args, **kwargs):
|
||||
raise httpx.ConnectError("dns failure")
|
||||
|
||||
monkeypatch.setattr("hermes_cli.auth.httpx.post", _boom)
|
||||
with pytest.raises(AuthError) as exc_info:
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
)
|
||||
assert exc_info.value.code == "xai_token_exchange_failed"
|
||||
assert "dns failure" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_non_dict_payload_raises_invalid_json(monkeypatch):
|
||||
"""xAI returning ``[]`` or a string at 200 is a server bug — fail
|
||||
with a precise error rather than crashing later in token storage."""
|
||||
recorder = _PostRecorder(_ok_response([1, 2, 3])) # type: ignore[arg-type]
|
||||
monkeypatch.setattr("hermes_cli.auth.httpx.post", recorder)
|
||||
with pytest.raises(AuthError) as exc_info:
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
)
|
||||
assert exc_info.value.code == "xai_token_exchange_invalid"
|
||||
|
||||
|
||||
def test_success_returns_full_payload_dict(post_recorder):
|
||||
"""200 happy path: the parsed JSON dict comes back verbatim so the
|
||||
caller can pluck ``access_token`` / ``refresh_token`` etc."""
|
||||
out = _xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="v" * 64,
|
||||
code_challenge="c" * 43,
|
||||
)
|
||||
assert out["access_token"] == "AT-fresh"
|
||||
assert out["refresh_token"] == "RT-fresh"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Wire-format guard: httpx must serialise ``data`` as form-urlencoded
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_wire_format_is_form_urlencoded_with_all_pkce_fields(monkeypatch):
|
||||
"""End-to-end check on the actual bytes httpx puts on the wire.
|
||||
If anyone ever swaps ``data=`` for ``json=`` or refactors the dict,
|
||||
xAI will start rejecting again — this catches it locally."""
|
||||
|
||||
captured: Dict[str, Any] = {}
|
||||
|
||||
class _Transport(httpx.BaseTransport):
|
||||
def handle_request(self, request):
|
||||
captured["body"] = bytes(request.read())
|
||||
captured["content_type"] = request.headers.get("content-type", "")
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={"access_token": "AT", "refresh_token": "RT",
|
||||
"id_token": "", "expires_in": 60, "token_type": "Bearer"},
|
||||
)
|
||||
|
||||
real_post = httpx.post
|
||||
|
||||
def _post(*args, **kwargs):
|
||||
with httpx.Client(transport=_Transport()) as c:
|
||||
return c.post(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr("hermes_cli.auth.httpx.post", _post)
|
||||
|
||||
_xai_oauth_exchange_code_for_tokens(
|
||||
token_endpoint="https://auth.x.ai/oauth2/token",
|
||||
code="AUTHCODE",
|
||||
redirect_uri="http://127.0.0.1:56121/callback",
|
||||
code_verifier="theVerifier_43+",
|
||||
code_challenge="theChallenge_43+",
|
||||
)
|
||||
|
||||
assert "application/x-www-form-urlencoded" in captured["content_type"]
|
||||
parsed = parse_qs(captured["body"].decode())
|
||||
assert parsed["grant_type"] == ["authorization_code"]
|
||||
assert parsed["code"] == ["AUTHCODE"]
|
||||
assert parsed["redirect_uri"] == ["http://127.0.0.1:56121/callback"]
|
||||
assert parsed["client_id"] == [XAI_OAUTH_CLIENT_ID]
|
||||
assert parsed["code_verifier"] == ["theVerifier_43+"]
|
||||
assert parsed["code_challenge"] == ["theChallenge_43+"]
|
||||
assert parsed["code_challenge_method"] == ["S256"]
|
||||
Reference in New Issue
Block a user