mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-29 23:41:35 +08:00
Compare commits
1 Commits
fix/plugin
...
rm/add-por
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
857496b343 |
@@ -466,6 +466,23 @@ class AuthError(RuntimeError):
|
||||
self.relogin_required = relogin_required
|
||||
|
||||
|
||||
def _nous_billing_url(portal_base_url: Any = None) -> str:
|
||||
"""Return the Nous Portal billing URL, preferring the active portal origin."""
|
||||
base_url = (
|
||||
_optional_base_url(portal_base_url)
|
||||
or _optional_base_url(os.getenv("HERMES_PORTAL_BASE_URL"))
|
||||
or _optional_base_url(os.getenv("NOUS_PORTAL_BASE_URL"))
|
||||
)
|
||||
if not base_url:
|
||||
try:
|
||||
auth_state = get_provider_auth_state("nous") or {}
|
||||
except Exception:
|
||||
auth_state = {}
|
||||
if isinstance(auth_state, dict):
|
||||
base_url = _optional_base_url(auth_state.get("portal_base_url"))
|
||||
return f"{(base_url or DEFAULT_NOUS_PORTAL_URL).rstrip('/')}/billing"
|
||||
|
||||
|
||||
def format_auth_error(error: Exception) -> str:
|
||||
"""Map auth failures to concise user-facing guidance."""
|
||||
if not isinstance(error, AuthError):
|
||||
@@ -477,13 +494,13 @@ def format_auth_error(error: Exception) -> str:
|
||||
if error.code == "subscription_required":
|
||||
return (
|
||||
"No active paid subscription found on Nous Portal. "
|
||||
"Please purchase/activate a subscription, then retry."
|
||||
f"Please purchase/activate a subscription at {_nous_billing_url()}, then retry."
|
||||
)
|
||||
|
||||
if error.code == "insufficient_credits":
|
||||
return (
|
||||
"Subscription credits are exhausted. "
|
||||
"Top up/renew credits in Nous Portal, then retry."
|
||||
f"Top up/renew credits at {_nous_billing_url()}, then retry."
|
||||
)
|
||||
|
||||
if error.code == "temporarily_unavailable":
|
||||
@@ -2802,15 +2819,20 @@ def _nous_device_code_login(
|
||||
force_mint=True,
|
||||
)
|
||||
except AuthError as exc:
|
||||
if exc.code == "subscription_required":
|
||||
portal_url = auth_state.get(
|
||||
"portal_base_url", DEFAULT_NOUS_PORTAL_URL
|
||||
).rstrip("/")
|
||||
if exc.code in {"subscription_required", "insufficient_credits"}:
|
||||
billing_url = _nous_billing_url(auth_state.get("portal_base_url"))
|
||||
print()
|
||||
print("Your Nous Portal account does not have an active subscription.")
|
||||
print(f" Subscribe here: {portal_url}/billing")
|
||||
if exc.code == "subscription_required":
|
||||
print("Your Nous Portal account does not have an active subscription.")
|
||||
print(f" Subscribe here: {billing_url}")
|
||||
print()
|
||||
print("After subscribing, run `hermes model` again to finish setup.")
|
||||
else:
|
||||
print("Your Nous Portal credits are exhausted.")
|
||||
print(f" Top up credits here: {billing_url}")
|
||||
print()
|
||||
print("After topping up, run `hermes model` again to finish setup.")
|
||||
print()
|
||||
print("After subscribing, run `hermes model` again to finish setup.")
|
||||
raise SystemExit(1)
|
||||
raise
|
||||
|
||||
|
||||
@@ -7,7 +7,13 @@ from pathlib import Path
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from hermes_cli.auth import AuthError, get_provider_auth_state, resolve_nous_runtime_credentials
|
||||
from hermes_cli.auth import (
|
||||
AuthError,
|
||||
_nous_device_code_login,
|
||||
format_auth_error,
|
||||
get_provider_auth_state,
|
||||
resolve_nous_runtime_credentials,
|
||||
)
|
||||
|
||||
|
||||
def _setup_nous_auth(
|
||||
@@ -54,6 +60,69 @@ def _mint_payload(api_key: str = "agent-key") -> dict:
|
||||
}
|
||||
|
||||
|
||||
def test_format_auth_error_includes_billing_link_for_subscription_required(tmp_path, monkeypatch):
|
||||
hermes_home = tmp_path / "hermes"
|
||||
_setup_nous_auth(hermes_home)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
message = format_auth_error(
|
||||
AuthError("subscription required", provider="nous", code="subscription_required")
|
||||
)
|
||||
|
||||
assert "https://portal.example.com/billing" in message
|
||||
|
||||
|
||||
def test_format_auth_error_includes_billing_link_for_insufficient_credits(tmp_path, monkeypatch):
|
||||
hermes_home = tmp_path / "hermes"
|
||||
_setup_nous_auth(hermes_home)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
message = format_auth_error(
|
||||
AuthError("credits exhausted", provider="nous", code="insufficient_credits")
|
||||
)
|
||||
|
||||
assert "https://portal.example.com/billing" in message
|
||||
|
||||
|
||||
def test_nous_device_code_login_shows_billing_link_when_credits_exhausted(monkeypatch, capsys):
|
||||
monkeypatch.setattr("hermes_cli.auth._is_remote_session", lambda: False)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.auth._request_device_code",
|
||||
lambda **kwargs: {
|
||||
"verification_uri_complete": "https://portal.example.com/verify",
|
||||
"user_code": "ABC-123",
|
||||
"expires_in": 600,
|
||||
"interval": 1,
|
||||
"device_code": "device-code",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.auth._poll_for_token",
|
||||
lambda **kwargs: {
|
||||
"access_token": "access-token",
|
||||
"refresh_token": "refresh-token",
|
||||
"expires_in": 3600,
|
||||
"token_type": "Bearer",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.auth.refresh_nous_oauth_from_state",
|
||||
lambda *args, **kwargs: (_ for _ in ()).throw(
|
||||
AuthError("credits exhausted", provider="nous", code="insufficient_credits")
|
||||
),
|
||||
)
|
||||
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
_nous_device_code_login(
|
||||
portal_base_url="https://portal.example.com",
|
||||
open_browser=False,
|
||||
)
|
||||
|
||||
assert exc.value.code == 1
|
||||
out = capsys.readouterr().out
|
||||
assert "Top up credits here: https://portal.example.com/billing" in out
|
||||
|
||||
|
||||
def test_refresh_token_persisted_when_mint_returns_insufficient_credits(tmp_path, monkeypatch):
|
||||
hermes_home = tmp_path / "hermes"
|
||||
_setup_nous_auth(hermes_home, refresh_token="refresh-old")
|
||||
|
||||
Reference in New Issue
Block a user