fix(tests): drive the Nous terminal-refresh test through the real resolver

The test built the AuthError by hand and called _recover_failed_refresh directly, so it pinned the
terminal set in auth.py but never the production raise sites in auth_nous.py (swapping auth_nous.py
to main stayed green). It now runs _refresh_entry_impl against a temp HERMES_HOME for both
'needs a login' shapes — no Portal login, and an expired JWT with no refresh token — so the raise
site's code is what reaches the classifier.
This commit is contained in:
teknium1
2026-09-18 03:34:23 -07:00
committed by Teknium
parent 5dbeaf2366
commit 6689bddeb7

View File

@@ -9,8 +9,11 @@ later refresh attempt.
"""
from __future__ import annotations
import base64
import json
import logging
import threading
import time
import pytest
@@ -117,23 +120,39 @@ def test_surviving_manual_entry_is_marked_dead_after_terminal_refresh(monkeypatc
assert [e.id for e in pool._entries] == ["e1"] # manual rows are never dropped by the quarantine
assert pool._entries[0].last_status == STATUS_DEAD
def test_nous_login_missing_refresh_failure_is_terminal(monkeypatch, caplog):
"""The resolver's "not logged in" raise (``nous_auth_missing``, relogin_required) is terminal:
retrying cannot succeed, so the row leaves rotation with a WARNING naming the fix and the
reason recorded, instead of an hour-long bench with null error fields (#113718)."""
from hermes_cli.auth_constants import _nous_err
def _expired_invoke_jwt() -> str:
def _part(payload: dict) -> str:
return base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=")
return f"{_part({'alg': 'none'})}.{_part({'sub': 'u', 'scope': 'inference:invoke', 'exp': int(time.time()) - 60})}.sig"
@pytest.mark.parametrize(
("nous_state", "expected_code"),
[
(None, "nous_auth_missing"),
({"client_id": "hermes-cli", "scope": "inference:invoke", "access_token": _expired_invoke_jwt(),
"refresh_token": "", "expires_at": "2026-02-01T00:00:00+00:00"}, "nous_auth_missing_refresh_token"),
],
ids=["not_logged_in", "expired_jwt_without_refresh_token"],
)
def test_nous_login_missing_refresh_failure_is_terminal(tmp_path, monkeypatch, caplog, nous_state, expected_code):
"""A "needs a login" raise from the real resolver — no Portal login at all, or an unusable
access token with no refresh token to redeem — is terminal: retrying cannot succeed, so the row
leaves rotation with a WARNING naming the fix and the reason recorded, instead of an hour-long
bench with null error fields (#113718). Driven through ``_refresh_entry_impl`` against a temp
HERMES_HOME so the production raise site's code is what reaches the classifier."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
providers = {"nous": nous_state} if nous_state else {}
(tmp_path / "auth.json").write_text(json.dumps({"version": 1, "providers": providers}), encoding="utf-8")
pool = _pool("nous")
entry = _entry("nous", source="manual:device_code")
pool._entries = [entry]
cleared: list = []
monkeypatch.setattr(pool, "_sync_nous_entry_from_auth_store", lambda e: e)
monkeypatch.setattr(pool, "_clear_terminal_nous_state", lambda e, exc: cleared.append(e.id))
monkeypatch.setattr(pool, "_quarantine_sources", lambda e, sources: None)
monkeypatch.setattr(pool, "_quarantine_sources", lambda e, sources: None) # keep the row to inspect it
exc = _nous_err("Hermes is not logged into Nous Portal.", "nous_auth_missing", relogin=True)
with caplog.at_level(logging.INFO, logger=cp.logger.name):
result = pool._recover_failed_refresh(entry, exc)
result = pool._refresh_entry_impl(entry, force=True)
assert result is None and cleared == ["e1"]
warnings = [r for r in caplog.records if r.levelno == logging.WARNING and "terminally invalid" in r.getMessage()]
@@ -141,7 +160,7 @@ def test_nous_login_missing_refresh_failure_is_terminal(monkeypatch, caplog):
assert "hermes auth add nous" in warnings[0].getMessage()
row = pool._entries[0]
assert row.last_status == STATUS_DEAD
assert (row.last_error_reason, row.last_error_message) == ("nous_auth_missing", str(exc))
assert row.last_error_reason == expected_code and row.last_error_message # no more null error fields
def test_nous_transient_error_still_benched(monkeypatch, caplog):