From 370ac6b4c19df35291dfadfcbf14da5a2730739f Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sat, 19 Sep 2026 18:43:38 -0700 Subject: [PATCH] test(auth): trim salvaged seam/registry tests to invariants; cover pool hook and metadata round-trip Two invariants per fix: the oauth plugin owns every hermes auth action through the real argparse surface (and a handler-less oauth profile fails loud while openrouter never consults the seam); the registry is complete after a mid-discovery auth import (subprocess) and a post-discovery register_provider() is mirrored; plugin pool rows refresh through refresh_credential (hook-less rows are returned unchanged) and unknown extra keys survive from_dict -> to_dict -> from_dict. --- .../agent/test_credential_pool_plugin_seam.py | 66 ++++++ tests/hermes_cli/test_provider_auth_seam.py | 189 ++++-------------- .../test_auth_registry_mid_discovery.py | 68 ------- 3 files changed, 102 insertions(+), 221 deletions(-) create mode 100644 tests/agent/test_credential_pool_plugin_seam.py diff --git a/tests/agent/test_credential_pool_plugin_seam.py b/tests/agent/test_credential_pool_plugin_seam.py new file mode 100644 index 0000000000..ccac5eb780 --- /dev/null +++ b/tests/agent/test_credential_pool_plugin_seam.py @@ -0,0 +1,66 @@ +"""Plugin providers in the credential pool (#116408): opaque metadata round-trips and refresh goes +through the profile's ``refresh_credential`` hook — eligibility derives from the hook, never a name set.""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest + +import providers +from providers.base import ProviderProfile + +from agent.credential_pool import AUTH_TYPE_OAUTH, CredentialPool, PooledCredential +from hermes_cli.auth_plugin_providers import is_refreshable_oauth_provider + + +def _entry(**over): + base = dict(provider="example-oauth", id="abc123", label="acme", auth_type=AUTH_TYPE_OAUTH, priority=0, + source="manual:example_device", access_token="tok-1", refresh_token="rt-1", + extra={"tenant": "acme", "region": "eu"}) + return PooledCredential(**{**base, **over}) + + +def test_plugin_metadata_survives_load_save_load(): + payload = _entry().to_dict() + again = PooledCredential.from_dict("example-oauth", payload).to_dict() + assert again["tenant"] == "acme" and again["region"] == "eu" + assert PooledCredential.from_dict("example-oauth", again).extra == {"tenant": "acme", "region": "eu"} + # Core-known extra keys keep their attribute surface; unknown ones stay opaque payload. + assert PooledCredential.from_dict("nous", {"access_token": "t", "org_id": "o1"}).org_id == "o1" + + +@pytest.fixture +def plugin_profiles(): + seen = [] + + def refresh_credential(entry): + seen.append(entry.refresh_token) + return {"access_token": "tok-2", "refresh_token": "rt-2"} + + providers.register_provider(ProviderProfile(name="example-oauth", auth_type="oauth_external", + base_url="https://example.invalid/v1", + refresh_credential=refresh_credential)) + providers.register_provider(ProviderProfile(name="example-oauth-nohook", auth_type="oauth_external", + base_url="https://example.invalid/v1")) + yield seen + for name in ("example-oauth", "example-oauth-nohook"): + providers._REGISTRY.pop(name, None) + providers._PROVIDER_LIST_CACHE = None + + +def test_pool_refresh_dispatches_to_profile_hook(plugin_profiles, monkeypatch): + assert is_refreshable_oauth_provider("example-oauth") is True + assert is_refreshable_oauth_provider("example-oauth-nohook") is False + assert is_refreshable_oauth_provider("anthropic") is True # built-ins unchanged + + entry = _entry() + pool = CredentialPool("example-oauth", [entry]) + monkeypatch.setattr(pool, "_persist", lambda *a, **k: None) + refreshed = pool._refresh_entry_impl(entry, force=True) + assert plugin_profiles == ["rt-1"] + assert (refreshed.access_token, refreshed.refresh_token, refreshed.extra) == ("tok-2", "rt-2", entry.extra) + + # Without the hook the pool must not pretend it refreshed anything. + nohook = replace(entry, provider="example-oauth-nohook") + assert CredentialPool("example-oauth-nohook", [nohook])._refresh_entry_impl(nohook, force=True) is nohook diff --git a/tests/hermes_cli/test_provider_auth_seam.py b/tests/hermes_cli/test_provider_auth_seam.py index 9a134a5e3e..04fb4c5de8 100644 --- a/tests/hermes_cli/test_provider_auth_seam.py +++ b/tests/hermes_cli/test_provider_auth_seam.py @@ -1,10 +1,8 @@ -"""`hermes auth ` prefers a model-provider plugin's auth handler. +"""OAuth-shaped model-provider plugins are first-class in `hermes auth` (#116408). -The seam: ``ProviderProfile.auth_handler`` (an optional callable on the profile a -``kind: model-provider`` plugin registers). These tests drive the real -``hermes auth`` subcommand surface through the real argparse definitions, so they -fail if the dispatch is dropped, reordered after the built-in paths, or stops -passing the parsed arguments through. +The seam: ``ProviderProfile.auth_handler``. These tests drive the real ``hermes auth`` argparse +surface against a plugin discovered from an isolated HERMES_HOME, so they fail if the dispatch is +dropped, reordered after the built-in paths, or the registry stops admitting non-api-key profiles. """ from __future__ import annotations @@ -13,7 +11,6 @@ import argparse import json import sys from pathlib import Path -from types import SimpleNamespace import pytest @@ -48,7 +45,8 @@ def handler(action, args): return True -register_provider(ProviderProfile(name="__NAME__", auth_handler=handler)) +register_provider(ProviderProfile(name="__NAME__", auth_type="oauth_external", + base_url="https://example.invalid/v1", auth_handler=handler)) ''' @@ -72,9 +70,7 @@ def install_provider(tmp_path, monkeypatch): """Write a model-provider plugin into an isolated HERMES_HOME and discover it.""" installed: list[str] = [] - def _install(name: str = "fake-auth", *, with_handler: bool = True, - async_handler: bool = False, raises: bool = False, - reinstall: bool = False) -> Path: + def _install(name: str = "fake-auth", *, with_handler: bool = True) -> Path: """Write (or rewrite) the fixture plugin and re-run discovery.""" plugin_dir = tmp_path / "hermes" / "plugins" / "model-providers" / name plugin_dir.mkdir(parents=True, exist_ok=True) @@ -84,10 +80,6 @@ def install_provider(tmp_path, monkeypatch): source = _PLUGIN_SOURCE.replace("__NAME__", name) if not with_handler: source = source.replace(", auth_handler=handler", "") - if async_handler: - source = source.replace("def handler(action, args):", "async def handler(action, args):") - if raises: - source = source.replace(" return True\n", ' raise ValueError("device flow exploded")\n') (plugin_dir / "__init__.py").write_text(source, encoding="utf-8") monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) @@ -126,158 +118,49 @@ def _log(tmp_path: Path) -> list[dict]: return [json.loads(line) for line in log.read_text(encoding="utf-8").splitlines()] -def test_add_dispatches_to_provider_handler_with_arguments(tmp_path, install_provider): - """`hermes auth add ` reaches the plugin handler, args included.""" +@pytest.mark.parametrize( + ("argv", "expected"), + [ + (["add", "fake-auth", "--label", "work", "--api-key", "sk-fixture"], + {"action": "add", "provider": "fake-auth", "label": "work", "target": None, "api_key": "sk-fixture"}), + (["status", "fake-auth"], + {"action": "status", "provider": "fake-auth", "label": None, "target": None, "api_key": None}), + (["logout", "fake-auth"], + {"action": "logout", "provider": "fake-auth", "label": None, "target": None, "api_key": None}), + (["refresh", "fake-auth", "acct-2"], + {"action": "refresh", "provider": "fake-auth", "label": None, "target": "acct-2", "api_key": None}), + ], +) +def test_oauth_plugin_owns_every_auth_action(tmp_path, install_provider, capsys, argv, expected): + """An oauth_external plugin registers and `hermes auth ` reaches its handler, args included, + before any built-in path (nothing printed, nothing written to the pool).""" install_provider() + import hermes_cli.auth as auth_mod from hermes_cli.auth_commands import auth_command - args = _parse_auth_args(["add", "fake-auth", "--label", "work", "--api-key", "sk-fixture"]) - assert args.auth_action == "add" and args.provider == "fake-auth" + assert auth_mod.resolve_provider("fake-auth") == "fake-auth" + assert auth_mod.PROVIDER_REGISTRY["fake-auth"].auth_type == "oauth_external" - auth_command(args) + auth_command(_parse_auth_args(argv)) - assert _log(tmp_path) == [{ - "action": "add", "provider": "fake-auth", "label": "work", - "target": None, "api_key": "sk-fixture"}] - # The built-in path never ran: nothing was written to the credential pool. + assert _log(tmp_path) == [expected] + assert capsys.readouterr().out == "" assert not (tmp_path / "hermes" / "auth.json").exists() -@pytest.mark.parametrize( - ("argv", "action"), - [ - (["status", "fake-auth"], "status"), - (["logout", "fake-auth"], "logout"), - (["refresh", "fake-auth", "acct-2"], "refresh"), - ], -) -def test_status_logout_refresh_dispatch_to_same_handler(tmp_path, install_provider, capsys, argv, action): - install_provider() - - from hermes_cli.auth_commands import auth_command - - args = _parse_auth_args(argv) - auth_command(args) - - recorded = _log(tmp_path) - assert [r["action"] for r in recorded] == [action] - # Core prints only when it owns the action; a dispatched action prints nothing here. - assert capsys.readouterr().out == "" - - -def test_declining_handler_falls_back_to_the_builtin_path(tmp_path, install_provider, monkeypatch): - """A handler may decline per action — core then behaves as it always did.""" - install_provider() - monkeypatch.setenv("FAKE_AUTH_DECLINE", "add") - - from hermes_cli.auth_commands import auth_command - - with pytest.raises(SystemExit) as excinfo: - auth_command(_parse_auth_args(["add", "fake-auth"])) - - # Offered first, declined by the handler, then the built-in unknown-provider exit... - assert [r["action"] for r in _log(tmp_path)] == ["add"] - # ...which now names the plugin instead of pretending the provider is unknown. - message = str(excinfo.value) - assert message.startswith("Unknown provider: fake-auth") - assert "does not provide auth handling" in message - - -def test_builtin_provider_without_handler_is_unchanged(tmp_path, install_provider): - """A provider with no handler keeps the exact built-in credential-pool path.""" - install_provider() - - from hermes_cli.auth_commands import auth_command - - auth_command(_parse_auth_args(["add", "openrouter", "--api-key", "sk-or-fixture", "--label", "personal"])) - - assert _log(tmp_path) == [] # no handler was ever consulted - pool = json.loads((tmp_path / "hermes" / "auth.json").read_text(encoding="utf-8"))["credential_pool"] - entry = next(e for e in pool["openrouter"] if e["access_token"] == "sk-or-fixture") - assert entry["label"] == "personal" - - -def test_provider_without_handler_still_reports_unknown_provider(tmp_path, install_provider): +def test_oauth_plugin_without_handler_fails_loud_and_builtins_are_untouched(tmp_path, install_provider): + """No handler on an oauth-shaped profile = a clear error naming the missing hook (never a silent + api-key prompt or "Unknown provider"); a built-in provider never consults the seam.""" install_provider("handlerless", with_handler=False) from hermes_cli.auth_commands import auth_command with pytest.raises(SystemExit) as excinfo: auth_command(_parse_auth_args(["add", "handlerless"])) + assert "ships no auth_handler" in str(excinfo.value) and "oauth_external" in str(excinfo.value) - message = str(excinfo.value) - assert message.startswith("Unknown provider: handlerless") - assert "does not provide auth handling" in message + auth_command(_parse_auth_args(["add", "openrouter", "--api-key", "sk-or-fixture", "--label", "personal"])) assert _log(tmp_path) == [] - - -def test_unregistered_provider_lookup_failure_falls_through(tmp_path, install_provider): - """Registry lookup failure (no profile at all) must never raise or dispatch.""" - install_provider() - - from hermes_cli.auth_commands import _dispatch_provider_auth, _provider_auth_handler, auth_status_command - - assert _provider_auth_handler("not-a-registered-provider") == (None, None) - assert _dispatch_provider_auth("add", SimpleNamespace(provider="not-a-registered-provider"), - "not-a-registered-provider") is False - - auth_status_command(SimpleNamespace(provider="not-a-registered-provider")) - assert _log(tmp_path) == [] - - -def test_duplicate_registration_last_writer_wins(tmp_path, install_provider): - """Two profiles under one name (a user plugin overriding a bundled one) resolve - to the newest handler — the documented override semantics of register_provider.""" - install_provider() - from hermes_cli.auth_commands import _provider_auth_handler - - first, _ = _provider_auth_handler("fake-auth") - assert callable(first.auth_handler) - - install_provider(reinstall=True) # second registration for the same name - - second, handler = _provider_auth_handler("fake-auth") - assert handler is not None and handler is not first.auth_handler - assert second is not first - - -def test_async_handler_is_awaited(tmp_path, install_provider): - install_provider(async_handler=True) - - from hermes_cli.auth_commands import auth_command - - auth_command(_parse_auth_args(["add", "fake-auth"])) - - assert [r["action"] for r in _log(tmp_path)] == ["add"] - - -def test_handler_failure_becomes_a_readable_exit(tmp_path, install_provider): - install_provider(raises=True) - - from hermes_cli.auth_commands import auth_command - - with pytest.raises(SystemExit) as excinfo: - auth_command(_parse_auth_args(["add", "fake-auth"])) - - message = str(excinfo.value) - assert "fake-auth auth handler failed for `add`" in message - assert "ValueError: device flow exploded" in message - - -def test_every_action_reaches_the_seam(tmp_path, install_provider): - """Guard against a future action being added to the core enum without dispatch.""" - install_provider() - - from hermes_cli import auth_commands - - for action, command in ( - ("add", auth_commands.auth_add_command), - ("status", auth_commands.auth_status_command), - ("logout", auth_commands.auth_logout_command), - ("refresh", auth_commands.auth_refresh_command), - ): - assert auth_commands._dispatch_provider_auth(action, SimpleNamespace(provider="fake-auth"), - "fake-auth"), action - - assert sorted(r["action"] for r in _log(tmp_path)) == ["add", "logout", "refresh", "status"] + pool = json.loads((tmp_path / "hermes" / "auth.json").read_text(encoding="utf-8"))["credential_pool"] + assert next(e for e in pool["openrouter"] if e["access_token"] == "sk-or-fixture")["label"] == "personal" diff --git a/tests/providers/test_auth_registry_mid_discovery.py b/tests/providers/test_auth_registry_mid_discovery.py index bca66d5dde..1f876ca5a0 100644 --- a/tests/providers/test_auth_registry_mid_discovery.py +++ b/tests/providers/test_auth_registry_mid_discovery.py @@ -154,56 +154,6 @@ def _isolated_registries(): providers._discovering = False -def test_mid_discovery_auth_import_reconciled( - _isolated_registries, monkeypatch, tmp_path -): - """Mid-discovery auth snapshot + late provider -> REGISTRY finally complete.""" - - def fake_entry_point_step(): - # Reproduce a mid-discovery `import hermes_cli.auth`: its module - # top-level snapshots list_providers() exactly like this, and at this - # point nothing has been registered yet, so the snapshot is partial. - auth_mod.sync_plugin_provider_registry() - providers.register_provider( - ProviderProfile( - name=EARLY, - display_name="Early", - base_url="https://early.example/v1", - env_vars=("PROBE_102123_EARLY_KEY",), - ) - ) - # Registration during discovery must NOT sync eagerly (that would be - # one sync per plugin); the hazard is still live at this point. - assert LATE not in auth_mod.PROVIDER_REGISTRY - - late_plugin = tmp_path / "zzz_probe_102123" - late_plugin.mkdir() - (late_plugin / "__init__.py").write_text( - "from providers import register_provider\n" - "from providers.base import ProviderProfile\n" - "register_provider(ProviderProfile(\n" - f" name={LATE!r}, display_name='Late',\n" - " base_url='https://late.example/v1',\n" - f" env_vars=('PROBE_102123_LATE_KEY',), aliases=({LATE_ALIAS!r},)))\n", - encoding="utf-8", - ) - sys.modules.pop("plugins.model_providers.zzz_probe_102123", None) - monkeypatch.setattr( - providers, "_discover_entry_point_providers", fake_entry_point_step - ) - monkeypatch.setattr(providers, "_BUNDLED_PLUGINS_DIR", tmp_path) - monkeypatch.setattr(providers, "_user_plugins_dir", lambda: None) - monkeypatch.setattr(providers, "_installed_plugins_dir", lambda: None) - - providers._discover_providers() - - assert providers.get_provider_profile(LATE) is not None - assert LATE in auth_mod.PROVIDER_REGISTRY - assert LATE_ALIAS in auth_mod.PROVIDER_REGISTRY - assert EARLY in auth_mod.PROVIDER_REGISTRY - assert providers._discovering is False - - def test_post_discovery_registration_is_mirrored(_isolated_registries, monkeypatch, tmp_path): """A register_provider() call after discovery finished reaches the auth registry at once.""" monkeypatch.setattr(providers, "_discover_entry_point_providers", lambda: None) @@ -223,21 +173,3 @@ def test_post_discovery_registration_is_mirrored(_isolated_registries, monkeypat ) assert LATE in auth_mod.PROVIDER_REGISTRY assert auth_mod.PROVIDER_REGISTRY[LATE_ALIAS] is auth_mod.PROVIDER_REGISTRY[LATE] - - -def test_sync_idempotent_and_no_clobber(_isolated_registries): - """Repeated syncs add nothing new and never replace existing entries.""" - snapshot = dict(auth_mod.PROVIDER_REGISTRY) - auth_mod.sync_plugin_provider_registry() - for key, config in snapshot.items(): - assert auth_mod.PROVIDER_REGISTRY[key] is config - after = dict(auth_mod.PROVIDER_REGISTRY) - assert auth_mod.sync_plugin_provider_registry() == 0 - assert dict(auth_mod.PROVIDER_REGISTRY) == after - - -def test_sync_noop_when_auth_never_imported(monkeypatch): - """The providers-side hook must not import hermes_cli.auth by itself.""" - monkeypatch.delitem(sys.modules, "hermes_cli.auth", raising=False) - providers._sync_auth_registry() - assert "hermes_cli.auth" not in sys.modules