Files
hermes-agent/tests/agent/test_bedrock_integration.py
teknium1 b3d4f67b20 test: purge low-value tests, lane py01 (345 removed)
Change-detectors, tautologies, source-reading tests, redundant duplicates,
mock-echo tests and dead/unrunnable tests. Per-test rationale in the lane
ledger (category + reason for every removal).
2026-09-23 03:15:26 -07:00

510 lines
22 KiB
Python

"""Integration tests for the AWS Bedrock provider wiring.
Verifies that the Bedrock provider is correctly registered in the
provider registry, model catalog, and runtime resolution pipeline.
These tests do NOT require AWS credentials or boto3 — all AWS calls
are mocked.
Note: Tests that import ``hermes_cli.auth`` or ``hermes_cli.runtime_provider``
require Python 3.10+ due to ``str | None`` type syntax in the import chain.
"""
from unittest.mock import MagicMock, patch
import pytest
_BOTO_PREFIXES = ("botocore", "boto3")
@pytest.fixture(autouse=True)
def _boto_sys_modules_hygiene():
"""Snapshot/restore boto* sys.modules around every test.
Tests here plant fake botocore/boto3 modules; a fake that leaks (or a
real submodule first-imported inside a stub window) poisons later
imports of the real ``botocore.exceptions`` with
``No module named 'botocore.vendored'`` (PR #92617 CI flake). This
fixture makes stub windows airtight regardless of test ordering.
"""
import sys as _sys
saved = {
name: mod
for name, mod in _sys.modules.items()
if name.split(".", 1)[0] in _BOTO_PREFIXES
}
yield
for name in [n for n in _sys.modules if n.split(".", 1)[0] in _BOTO_PREFIXES]:
_sys.modules.pop(name, None)
_sys.modules.update(saved)
class TestModelCatalog:
"""Verify Bedrock has a static model fallback list."""
def test_bedrock_has_curated_models(self):
from hermes_cli.models import _PROVIDER_MODELS
models = _PROVIDER_MODELS.get("bedrock", [])
assert len(models) > 0
class TestResolveProvider:
"""Verify resolve_provider() handles bedrock correctly."""
def test_explicit_bedrock_resolves(self, monkeypatch):
"""When user explicitly requests 'bedrock', it should resolve."""
# bedrock is in the registry, so resolve_provider should return it
from hermes_cli.auth import resolve_provider
result = resolve_provider("bedrock")
assert result == "bedrock"
def test_aws_alias_resolves_to_bedrock(self):
from hermes_cli.auth import resolve_provider
result = resolve_provider("aws")
assert result == "bedrock"
def test_auto_detect_with_aws_credentials(self, monkeypatch):
"""When AWS credentials are present and no other provider is configured,
auto-detect should find bedrock."""
from hermes_cli.auth import resolve_provider
# Clear all other provider env vars
for var in ["OPENAI_API_KEY", "OPENROUTER_API_KEY", "ANTHROPIC_API_KEY",
"ANTHROPIC_TOKEN", "GOOGLE_API_KEY", "DEEPSEEK_API_KEY"]:
monkeypatch.delenv(var, raising=False)
# Set AWS credentials
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
# The Nous free tier counts as a configured provider and sits above the Bedrock chain
# (NS-829); this test's contract is the chain itself, so switch the free tier off.
monkeypatch.setattr("hermes_cli.anon_auth.guest_enabled", lambda: False)
# Mock the auth store to have no active provider
with patch("hermes_cli.auth._load_auth_store", return_value={}):
result = resolve_provider("auto")
assert result == "bedrock"
class TestRuntimeProvider:
"""Verify resolve_runtime_provider() handles bedrock correctly."""
def test_bedrock_runtime_no_credentials_raises_on_auto_detect(self, monkeypatch):
"""When bedrock is auto-detected (not explicitly requested) and no
credentials are found, runtime resolution should raise AuthError."""
from hermes_cli.runtime_provider import resolve_runtime_provider
from hermes_cli.auth import AuthError
# Clear all AWS env vars
for var in ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_PROFILE",
"AWS_BEARER_TOKEN_BEDROCK", "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI",
"AWS_WEB_IDENTITY_TOKEN_FILE"]:
monkeypatch.delenv(var, raising=False)
# Mock both the provider resolution and boto3's credential chain
mock_session = MagicMock()
mock_session.get_credentials.return_value = None
with patch("hermes_cli.runtime_provider.resolve_provider", return_value="bedrock"), \
patch("hermes_cli.runtime_provider._get_model_config", return_value={"provider": "bedrock"}), \
patch("hermes_cli.runtime_provider.resolve_requested_provider", return_value="auto"), \
patch.dict("sys.modules", {"botocore": MagicMock(), "botocore.session": MagicMock()}):
import botocore.session as _bs
_bs.get_session = MagicMock(return_value=mock_session)
with pytest.raises(AuthError, match="No AWS credentials"):
resolve_runtime_provider(requested="auto")
def test_bedrock_runtime_explicit_skips_credential_check(self, monkeypatch):
"""When user explicitly requests bedrock, trust boto3's credential chain
even if env-var detection finds nothing (covers IMDS, SSO, etc.)."""
from hermes_cli.runtime_provider import resolve_runtime_provider
# No AWS env vars set — but explicit bedrock request should not raise
for var in ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_PROFILE",
"AWS_BEARER_TOKEN_BEDROCK"]:
monkeypatch.delenv(var, raising=False)
with patch("hermes_cli.runtime_provider.resolve_provider", return_value="bedrock"), \
patch("hermes_cli.runtime_provider._get_model_config", return_value={"provider": "bedrock"}):
result = resolve_runtime_provider(requested="bedrock")
assert result["provider"] == "bedrock"
assert result["api_mode"] == "bedrock_converse"
def test_bedrock_openai_models_route_to_mantle_responses(self, monkeypatch):
"""Bedrock's OpenAI models (GPT-5.5 / GPT-5.6 family) are not Converse
models — they only answer on the Mantle /openai/v1 Responses surface.
Every allowlisted ID must route there, with the aws-sdk IAM sentinel."""
from agent.bedrock_adapter import BEDROCK_OPENAI_RESPONSES_MODEL_IDS
from hermes_cli.runtime_provider import resolve_runtime_provider
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
monkeypatch.setenv("AWS_REGION", "us-east-2")
assert BEDROCK_OPENAI_RESPONSES_MODEL_IDS
for model_id in BEDROCK_OPENAI_RESPONSES_MODEL_IDS:
with patch("hermes_cli.runtime_provider.resolve_provider", return_value="bedrock"), \
patch("hermes_cli.runtime_provider._get_model_config", return_value={
"provider": "bedrock",
"default": model_id,
}):
result = resolve_runtime_provider(requested="bedrock")
assert result["api_mode"] == "codex_responses", model_id
assert result["model"] == model_id
assert result["base_url"] == "https://bedrock-mantle.us-east-2.api.aws/openai/v1"
assert result["api_key"] == "aws-sdk"
assert result["bedrock_openai"] is True, model_id
def test_bedrock_openai_models_do_not_fall_back_to_default_context(self):
"""The Mantle OpenAI models have a larger window than the generic
default; make sure they never fall back to it."""
from agent.bedrock_adapter import (
BEDROCK_DEFAULT_CONTEXT_LENGTH,
BEDROCK_OPENAI_RESPONSES_MODEL_IDS,
get_bedrock_context_length,
)
for model_id in BEDROCK_OPENAI_RESPONSES_MODEL_IDS:
assert get_bedrock_context_length(model_id) > BEDROCK_DEFAULT_CONTEXT_LENGTH, model_id
# ---------------------------------------------------------------------------
# providers.py integration
# ---------------------------------------------------------------------------
class TestProvidersModule:
"""Verify bedrock is wired into hermes_cli/providers.py."""
def test_determine_api_mode_from_bedrock_url(self):
from hermes_cli.providers import determine_api_mode
assert determine_api_mode(
"unknown", "https://bedrock-runtime.us-east-1.amazonaws.com"
) == "bedrock_converse"
# ---------------------------------------------------------------------------
# Error classifier integration
# ---------------------------------------------------------------------------
class TestErrorClassifierBedrock:
"""Verify Bedrock error patterns are in the global error classifier."""
def test_bedrock_throttling_classified_as_rate_limit(self):
from agent.error_classifier import FailoverReason, classify_api_error
err = Exception(
"An error occurred (ThrottlingException) when calling the ConverseStream "
"operation: Too many requests, please wait before trying again."
)
assert classify_api_error(err, provider="bedrock").reason == FailoverReason.rate_limit
def test_bedrock_input_too_long_classified_as_context_overflow(self):
from agent.error_classifier import FailoverReason, classify_api_error
err = Exception(
"An error occurred (ValidationException) when calling the ConverseStream "
"operation: Input is too long for requested model."
)
assert classify_api_error(err, provider="bedrock").reason == FailoverReason.context_overflow
# ---------------------------------------------------------------------------
# pyproject.toml bedrock extra
# ---------------------------------------------------------------------------
class TestPackaging:
"""Verify Bedrock remains a declared lazy optional dependency."""
@staticmethod
def _optional_dependencies():
import tomllib
from pathlib import Path
content = (Path(__file__).parent.parent.parent / "pyproject.toml").read_text()
return tomllib.loads(content)["project"]["optional-dependencies"]
def test_bedrock_extra_exists(self):
extras = self._optional_dependencies()
assert "bedrock" in extras
assert any(dep.startswith("boto3==") for dep in extras["bedrock"])
def test_bedrock_is_not_eager_installed_by_all_extra(self):
extras = self._optional_dependencies()
assert "hermes-agent[bedrock]" not in extras["all"]
# ---------------------------------------------------------------------------
# Model ID dot preservation — regression for #11976
# ---------------------------------------------------------------------------
# AWS Bedrock inference-profile model IDs embed structural dots:
#
# global.anthropic.claude-opus-4-7
# us.anthropic.claude-sonnet-4-5-20250929-v1:0
# apac.anthropic.claude-haiku-4-5
#
# ``agent.anthropic_message_convert.normalize_model_name`` converts dots to hyphens
# unless the caller opts in via ``preserve_dots=True``. Before this fix,
# ``AIAgent._anthropic_preserve_dots`` returned False for the ``bedrock``
# provider, so Claude-on-Bedrock requests went out with
# ``global-anthropic-claude-opus-4-7`` (all dots mangled to hyphens) and
# Bedrock rejected them with:
#
# HTTP 400: The provided model identifier is invalid.
#
# The fix adds ``bedrock`` to the preserve-dots provider allowlist and
# ``bedrock-runtime.`` to the base-URL heuristic, mirroring the shape of
# the opencode-go fix for #5211 (commit f77be22c), which extended this
# same allowlist.
class TestBedrockPreserveDotsFlag:
"""``AIAgent._anthropic_preserve_dots`` must return True on Bedrock so
inference-profile IDs survive the normalize step intact."""
def test_bedrock_provider_preserves_dots(self):
from types import SimpleNamespace
agent = SimpleNamespace(provider="bedrock", base_url="")
from run_agent import AIAgent
assert AIAgent._anthropic_preserve_dots(agent) is True
def test_non_bedrock_aws_url_does_not_preserve_dots(self):
"""Unrelated AWS endpoints (e.g. ``s3.us-east-1.amazonaws.com``)
must not accidentally activate the dot-preservation heuristic —
the heuristic is scoped to the ``bedrock-runtime.`` substring
specifically."""
from types import SimpleNamespace
agent = SimpleNamespace(
provider="custom",
base_url="https://s3.us-east-1.amazonaws.com",
)
from run_agent import AIAgent
assert AIAgent._anthropic_preserve_dots(agent) is False
class TestBedrockModelNameNormalization:
"""End-to-end: ``normalize_model_name`` + the preserve-dots flag
reproduce the exact production request shape for each Bedrock model
family, confirming the fix resolves the reporter's HTTP 400."""
def test_bedrock_prefix_preserved_without_preserve_dots(self):
"""Bedrock inference profile IDs are auto-detected by prefix and
always returned unmangled -- ``preserve_dots`` is irrelevant for
these IDs because the dots are namespace separators, not version
separators. Regression for #12295."""
from agent.anthropic_message_convert import normalize_model_name
assert normalize_model_name(
"global.anthropic.claude-opus-4-7", preserve_dots=False
) == "global.anthropic.claude-opus-4-7"
class TestBedrockBuildAnthropicKwargsEndToEnd:
"""Integration: calling ``build_anthropic_kwargs`` with a Bedrock-
shaped model ID and ``preserve_dots=True`` produces the unmangled
model string in the outgoing kwargs — the exact body sent to the
``bedrock-runtime.`` endpoint. This is the integration-level
regression for the reporter's HTTP 400."""
def test_bedrock_model_preserved_without_preserve_dots(self):
"""Bedrock inference profile IDs survive ``build_anthropic_kwargs``
even without ``preserve_dots=True`` -- the prefix auto-detection
in ``normalize_model_name`` is the load-bearing piece.
Regression for #12295."""
from agent.anthropic_adapter import build_anthropic_kwargs
kwargs = build_anthropic_kwargs(
model="global.anthropic.claude-opus-4-7",
messages=[{"role": "user", "content": "hi"}],
tools=None,
max_tokens=1024,
reasoning_config=None,
preserve_dots=False,
)
assert kwargs["model"] == "global.anthropic.claude-opus-4-7"
class TestBedrockModelIdDetection:
"""Tests for ``_is_bedrock_model_id`` and the auto-detection that
makes ``normalize_model_name`` preserve dots for Bedrock IDs
regardless of ``preserve_dots``. Regression for #12295."""
def test_bare_bedrock_id_preserved_without_flag(self):
"""The primary bug from #12295: ``anthropic.claude-opus-4-7``
sent to bedrock-mantle via auxiliary clients that don't pass
``preserve_dots=True``."""
from agent.anthropic_message_convert import normalize_model_name
assert normalize_model_name(
"anthropic.claude-opus-4-7", preserve_dots=False
) == "anthropic.claude-opus-4-7"
# ---------------------------------------------------------------------------
# auxiliary_client Bedrock resolution — fix for #13919
# ---------------------------------------------------------------------------
# Before the fix, resolve_provider_client("bedrock", ...) fell through to the
# "unhandled auth_type" warning and returned (None, None), breaking all
# auxiliary tasks (compression, memory, summarization) for Bedrock users.
class TestAuxiliaryClientBedrockResolution:
"""Verify resolve_provider_client handles Bedrock's aws_sdk auth type."""
def test_bedrock_returns_client_with_credentials(self, monkeypatch):
"""With valid AWS credentials, Bedrock should return a usable client."""
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
monkeypatch.setenv("AWS_REGION", "us-west-2")
mock_anthropic_bedrock = MagicMock()
with patch("agent.anthropic_adapter.build_anthropic_bedrock_client",
return_value=mock_anthropic_bedrock):
from agent.auxiliary_client import resolve_provider_client, AnthropicAuxiliaryClient
client, model = resolve_provider_client("bedrock", None)
assert client is not None, (
"resolve_provider_client('bedrock') returned None — "
"aws_sdk auth type is not handled"
)
assert isinstance(client, AnthropicAuxiliaryClient)
assert model is not None
assert client.api_key == "aws-sdk"
assert "us-west-2" in client.base_url
def test_bedrock_returns_none_without_credentials(self, monkeypatch):
"""Without AWS credentials, Bedrock should return (None, None) gracefully."""
with patch("agent.bedrock_adapter.has_aws_credentials", return_value=False):
from agent.auxiliary_client import resolve_provider_client
client, model = resolve_provider_client("bedrock", None)
assert client is None
assert model is None
def test_bedrock_shim_uncapped_when_caller_omits_max_tokens(self, monkeypatch):
"""No caller max_tokens → the shim passes None through and the wire
request carries no inferenceConfig.maxTokens, so Bedrock uses the
model's maximum allowed output (#10809 on the Bedrock wire).
Guards against the shim's old hardcoded ``else 4096`` fallback, which
kept aux vision descriptions capped after the vision call sites
dropped their own caps."""
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIO...MPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
from agent.auxiliary_client import BedrockAuxiliaryClient
client = BedrockAuxiliaryClient("us-east-1", "openai.gpt-oss-20b-1:0")
boto3_client = MagicMock()
with patch("agent.bedrock_adapter._get_bedrock_runtime_client",
return_value=boto3_client), \
patch("agent.bedrock_adapter.normalize_converse_response"):
# Aux vision-style call: no max_tokens key at all.
client.chat.completions.create(
model="openai.gpt-oss-20b-1:0",
messages=[{"role": "user", "content": "describe"}],
temperature=0.1,
)
wire_kwargs = boto3_client.converse.call_args.kwargs
assert "maxTokens" not in wire_kwargs.get("inferenceConfig", {})
# An explicit caller cap still lands on the wire unchanged.
client.chat.completions.create(
model="openai.gpt-oss-20b-1:0",
messages=[{"role": "user", "content": "describe"}],
max_tokens=1234,
)
wire_kwargs = boto3_client.converse.call_args.kwargs
assert wire_kwargs["inferenceConfig"]["maxTokens"] == 1234
def test_bedrock_mantle_config_region_beats_env_region(self, monkeypatch):
"""bedrock.region in config.yaml must win over AWS_REGION for auxiliary
Mantle calls — the same priority the main runtime resolver uses (#65076
review: aux resolution previously derived its region env-first and
could leave the primary runtime's configured region)."""
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
monkeypatch.setenv("AWS_REGION", "eu-central-1")
monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "test-bearer")
captured = {}
class _FakeOpenAI:
def __init__(self, **kwargs):
captured.update(kwargs)
self.api_key = kwargs.get("api_key")
self.base_url = kwargs.get("base_url")
def close(self):
pass
with patch("hermes_cli.config.load_config_readonly",
return_value={"bedrock": {"region": "us-west-2"}}), \
patch("agent.auxiliary_client.OpenAI", _FakeOpenAI):
from agent.auxiliary_client import resolve_provider_client
client, model = resolve_provider_client("bedrock", "openai.gpt-5.6-sol")
assert client is not None
assert model == "openai.gpt-5.6-sol"
assert "us-west-2" in captured.get("base_url", ""), (
"Mantle auxiliary base_url ignored config.yaml bedrock.region"
)
def test_bedrock_openai_aux_uses_responses_client(self, monkeypatch):
"""Auxiliary tasks on Bedrock GPT models use the Mantle Responses
path (SigV4 http client + aws-sdk sentinel), not the Anthropic shim."""
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
monkeypatch.setenv("AWS_REGION", "us-east-2")
with patch("agent.auxiliary_client.OpenAI", return_value=MagicMock()) as mock_openai, \
patch("agent.bedrock_adapter.build_bedrock_openai_http_client", return_value=MagicMock()):
from agent.auxiliary_client import resolve_provider_client, CodexAuxiliaryClient
client, model = resolve_provider_client("bedrock", "openai.gpt-5.5")
assert model == "openai.gpt-5.5"
assert isinstance(client, CodexAuxiliaryClient)
kwargs = mock_openai.call_args.kwargs
assert kwargs["api_key"] == "aws-sdk"
assert kwargs["base_url"] == "https://bedrock-mantle.us-east-2.api.aws/openai/v1"
assert "http_client" in kwargs