"""Tests for the automatic fallback router module. Tests quota error detection, fallback chain resolution, and auto-fallback logic. """ import os import pytest from unittest.mock import MagicMock, patch from agent.fallback_router import ( is_quota_error, get_default_fallback_chain, should_auto_fallback, log_fallback_event, get_auto_fallback_chain, is_fallback_available, filter_available_fallbacks, QUOTA_STATUS_CODES, DEFAULT_FALLBACK_CHAINS, ) class TestIsQuotaError: """Tests for quota error detection.""" def test_none_error_returns_false(self): assert is_quota_error(None) is False def test_rate_limit_status_code_429(self): error = MagicMock() error.status_code = 429 error.__str__ = lambda self: "Rate limit exceeded" assert is_quota_error(error) is True def test_payment_required_status_code_402(self): error = MagicMock() error.status_code = 402 error.__str__ = lambda self: "Payment required" assert is_quota_error(error) is True def test_forbidden_status_code_403(self): error = MagicMock() error.status_code = 403 error.__str__ = lambda self: "Forbidden" assert is_quota_error(error) is True def test_anthropic_quota_patterns(self): patterns = [ "Rate limit exceeded", "quota exceeded", "insufficient quota", "capacity exceeded", "over capacity", "billing threshold reached", "credit balance too low", ] for pattern in patterns: error = Exception(pattern) assert is_quota_error(error, provider="anthropic") is True, f"Failed for: {pattern}" def test_anthropic_error_type_detection(self): class RateLimitError(Exception): pass error = RateLimitError("Too many requests") assert is_quota_error(error) is True def test_non_quota_error(self): error = Exception("Some random error") assert is_quota_error(error) is False def test_context_length_error_not_quota(self): error = Exception("Context length exceeded") assert is_quota_error(error) is False def test_provider_specific_patterns(self): # Test openrouter patterns error = Exception("Insufficient credits") assert is_quota_error(error, provider="openrouter") is True # Test kimi patterns error = Exception("Insufficient balance") assert is_quota_error(error, provider="kimi-coding") is True class TestGetDefaultFallbackChain: """Tests for default fallback chain retrieval.""" def test_anthropic_fallback_chain(self): chain = get_default_fallback_chain("anthropic") assert len(chain) >= 1 assert chain[0]["provider"] == "kimi-coding" assert chain[0]["model"] == "kimi-k2.5" def test_openrouter_fallback_chain(self): chain = get_default_fallback_chain("openrouter") assert len(chain) >= 1 assert any(fb["provider"] == "kimi-coding" for fb in chain) def test_unknown_provider_returns_empty(self): chain = get_default_fallback_chain("unknown_provider") assert chain == [] def test_exclude_provider(self): chain = get_default_fallback_chain("anthropic", exclude_provider="kimi-coding") assert all(fb["provider"] != "kimi-coding" for fb in chain) class TestShouldAutoFallback: """Tests for auto-fallback decision logic.""" def test_auto_fallback_enabled_by_default(self): with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "true"}): assert should_auto_fallback("anthropic") is True def test_auto_fallback_disabled_via_env(self): with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "false"}): assert should_auto_fallback("anthropic") is False def test_auto_fallback_disabled_via_override(self): assert should_auto_fallback("anthropic", auto_fallback_enabled=False) is False def test_quota_error_triggers_fallback(self): error = Exception("Rate limit exceeded") assert should_auto_fallback("unknown_provider", error=error) is True def test_non_quota_error_no_fallback(self): error = Exception("Some random error") # Unknown provider with non-quota error should not fallback assert should_auto_fallback("unknown_provider", error=error) is False def test_anthropic_eager_fallback(self): # Anthropic falls back eagerly even without error assert should_auto_fallback("anthropic") is True class TestLogFallbackEvent: """Tests for fallback event logging.""" def test_log_fallback_event(self): with patch("agent.fallback_router.logger") as mock_logger: log_fallback_event( from_provider="anthropic", to_provider="kimi-coding", to_model="kimi-k2.5", reason="quota_exceeded", ) mock_logger.info.assert_called_once() # Check the arguments passed to logger.info call_args = mock_logger.info.call_args[0] # First arg is format string, remaining are the values assert len(call_args) >= 4 assert "anthropic" in call_args # Provider names are in the args assert "kimi-coding" in call_args def test_log_fallback_event_with_error(self): error = Exception("Rate limit exceeded") with patch("agent.fallback_router.logger") as mock_logger: log_fallback_event( from_provider="anthropic", to_provider="kimi-coding", to_model="kimi-k2.5", reason="quota_exceeded", error=error, ) mock_logger.info.assert_called_once() mock_logger.debug.assert_called_once() class TestGetAutoFallbackChain: """Tests for automatic fallback chain resolution.""" def test_user_chain_takes_precedence(self): user_chain = [{"provider": "zai", "model": "glm-5"}] chain = get_auto_fallback_chain("anthropic", user_fallback_chain=user_chain) assert chain == user_chain def test_default_chain_when_no_user_chain(self): chain = get_auto_fallback_chain("anthropic") assert chain == DEFAULT_FALLBACK_CHAINS["anthropic"] class TestIsFallbackAvailable: """Tests for fallback availability checking.""" def test_anthropic_available_with_key(self): with patch.dict(os.environ, {"ANTHROPIC_API_KEY": "test-key"}): config = {"provider": "anthropic", "model": "claude-3"} assert is_fallback_available(config) is True def test_anthropic_unavailable_without_key(self): with patch.dict(os.environ, {}, clear=True): config = {"provider": "anthropic", "model": "claude-3"} assert is_fallback_available(config) is False def test_kimi_available_with_key(self): with patch.dict(os.environ, {"KIMI_API_KEY": "test-key"}): config = {"provider": "kimi-coding", "model": "kimi-k2.5"} assert is_fallback_available(config) is True def test_kimi_available_with_token(self): with patch.dict(os.environ, {"KIMI_API_TOKEN": "test-token"}): config = {"provider": "kimi-coding", "model": "kimi-k2.5"} assert is_fallback_available(config) is True def test_invalid_config_returns_false(self): assert is_fallback_available({}) is False assert is_fallback_available({"provider": ""}) is False class TestFilterAvailableFallbacks: """Tests for filtering available fallbacks.""" def test_filters_unavailable_providers(self): with patch.dict(os.environ, {"KIMI_API_KEY": "test-key"}): chain = [ {"provider": "kimi-coding", "model": "kimi-k2.5"}, {"provider": "anthropic", "model": "claude-3"}, # No key ] available = filter_available_fallbacks(chain) assert len(available) == 1 assert available[0]["provider"] == "kimi-coding" def test_returns_empty_when_none_available(self): with patch.dict(os.environ, {}, clear=True): chain = [ {"provider": "anthropic", "model": "claude-3"}, {"provider": "kimi-coding", "model": "kimi-k2.5"}, ] available = filter_available_fallbacks(chain) assert available == [] def test_preserves_order(self): with patch.dict(os.environ, {"KIMI_API_KEY": "test", "ANTHROPIC_API_KEY": "test"}): chain = [ {"provider": "kimi-coding", "model": "kimi-k2.5"}, {"provider": "anthropic", "model": "claude-3"}, ] available = filter_available_fallbacks(chain) assert len(available) == 2 assert available[0]["provider"] == "kimi-coding" assert available[1]["provider"] == "anthropic" class TestIntegration: """Integration tests for the fallback router.""" def test_full_fallback_flow_for_anthropic_quota(self): """Test the complete fallback flow when Anthropic quota is exceeded.""" # Simulate Anthropic quota error error = Exception("Rate limit exceeded: quota exceeded for model claude-3") # Verify error detection assert is_quota_error(error, provider="anthropic") is True # Verify auto-fallback is enabled assert should_auto_fallback("anthropic", error=error) is True # Get fallback chain chain = get_auto_fallback_chain("anthropic") assert len(chain) > 0 # Verify kimi-coding is first fallback assert chain[0]["provider"] == "kimi-coding" def test_fallback_availability_checking(self): """Test that fallback availability is properly checked.""" with patch.dict(os.environ, {"KIMI_API_KEY": "test-key"}): # Get default chain for anthropic chain = get_default_fallback_chain("anthropic") # Filter to available available = filter_available_fallbacks(chain) # Should have kimi-coding available assert any(fb["provider"] == "kimi-coding" for fb in available) class TestFallbackChainIntegration: """Integration tests for the complete fallback chain: anthropic -> kimi-coding -> openrouter.""" def test_complete_fallback_chain_structure(self): """Test that the complete fallback chain has correct structure.""" chain = get_default_fallback_chain("anthropic") # Should have at least 2 fallbacks: kimi-coding and openrouter assert len(chain) >= 2, f"Expected at least 2 fallbacks, got {len(chain)}" # First fallback should be kimi-coding assert chain[0]["provider"] == "kimi-coding" assert chain[0]["model"] == "kimi-k2.5" # Second fallback should be openrouter assert chain[1]["provider"] == "openrouter" assert "claude" in chain[1]["model"].lower() def test_fallback_chain_resolution_order(self): """Test that fallback chain respects the defined order.""" with patch.dict(os.environ, { "KIMI_API_KEY": "test-kimi-key", "OPENROUTER_API_KEY": "test-openrouter-key", }): chain = get_default_fallback_chain("anthropic") available = filter_available_fallbacks(chain) # Both providers should be available assert len(available) >= 2 # Order should be preserved: kimi-coding first, then openrouter assert available[0]["provider"] == "kimi-coding" assert available[1]["provider"] == "openrouter" def test_fallback_chain_skips_unavailable_providers(self): """Test that chain skips providers without credentials.""" with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}, clear=True): chain = get_default_fallback_chain("anthropic") available = filter_available_fallbacks(chain) # kimi-coding not available (no key), openrouter should be first assert len(available) >= 1 assert available[0]["provider"] == "openrouter" # kimi-coding should not be in available list assert not any(fb["provider"] == "kimi-coding" for fb in available) def test_fallback_chain_exhaustion(self): """Test behavior when all fallbacks are exhausted.""" with patch.dict(os.environ, {}, clear=True): chain = get_default_fallback_chain("anthropic") available = filter_available_fallbacks(chain) # No providers available assert available == [] def test_kimi_coding_fallback_chain(self): """Test that kimi-coding has its own fallback chain to openrouter.""" chain = get_default_fallback_chain("kimi-coding") assert len(chain) >= 1 # First fallback should be openrouter assert chain[0]["provider"] == "openrouter" def test_openrouter_fallback_chain(self): """Test that openrouter has its own fallback chain.""" chain = get_default_fallback_chain("openrouter") assert len(chain) >= 1 # Should include kimi-coding as fallback assert any(fb["provider"] == "kimi-coding" for fb in chain) class TestQuotaErrorDetection: """Comprehensive tests for quota error detection across providers.""" def test_anthropic_429_status_code(self): """Test 429 status code detection for Anthropic.""" error = MagicMock() error.status_code = 429 error.__str__ = lambda self: "Rate limit exceeded" assert is_quota_error(error, provider="anthropic") is True def test_anthropic_402_payment_required(self): """Test 402 payment required detection for Anthropic.""" error = MagicMock() error.status_code = 402 error.__str__ = lambda self: "Payment required" assert is_quota_error(error, provider="anthropic") is True def test_anthropic_403_forbidden_quota(self): """Test 403 forbidden detection for Anthropic quota.""" error = MagicMock() error.status_code = 403 error.__str__ = lambda self: "Forbidden" assert is_quota_error(error, provider="anthropic") is True def test_openrouter_quota_patterns(self): """Test OpenRouter-specific quota error patterns.""" patterns = [ "Rate limit exceeded", "Insufficient credits", "No endpoints available", "All providers failed", "Over capacity", ] for pattern in patterns: error = Exception(pattern) assert is_quota_error(error, provider="openrouter") is True, f"Failed for: {pattern}" def test_kimi_quota_patterns(self): """Test kimi-coding-specific quota error patterns.""" patterns = [ "Rate limit exceeded", "Insufficient balance", "Quota exceeded", ] for pattern in patterns: error = Exception(pattern) assert is_quota_error(error, provider="kimi-coding") is True, f"Failed for: {pattern}" def test_generic_quota_patterns(self): """Test generic quota patterns work across all providers.""" generic_patterns = [ "rate limit exceeded", "quota exceeded", "too many requests", "capacity exceeded", "temporarily unavailable", "resource exhausted", "insufficient credits", ] for pattern in generic_patterns: error = Exception(pattern) assert is_quota_error(error) is True, f"Failed for generic pattern: {pattern}" def test_non_quota_errors_not_detected(self): """Test that non-quota errors are not incorrectly detected.""" non_quota_errors = [ "Context length exceeded", "Invalid API key", "Model not found", "Network timeout", "Connection refused", "JSON decode error", ] for pattern in non_quota_errors: error = Exception(pattern) assert is_quota_error(error) is False, f"Incorrectly detected as quota: {pattern}" def test_error_type_detection(self): """Test that specific exception types are detected as quota errors.""" class RateLimitError(Exception): pass class QuotaExceededError(Exception): pass class TooManyRequests(Exception): pass for exc_class in [RateLimitError, QuotaExceededError, TooManyRequests]: error = exc_class("Some message") assert is_quota_error(error) is True, f"Failed for {exc_class.__name__}" class TestFallbackLogging: """Tests for fallback event logging.""" def test_fallback_event_logged_with_all_params(self): """Test that fallback events log all required parameters.""" with patch("agent.fallback_router.logger") as mock_logger: log_fallback_event( from_provider="anthropic", to_provider="kimi-coding", to_model="kimi-k2.5", reason="quota_exceeded", ) # Verify info was called mock_logger.info.assert_called_once() # Verify the log message format and arguments call_args = mock_logger.info.call_args log_format = call_args[0][0] log_args = call_args[0][1:] # Remaining positional args # Check format string contains placeholders assert "%s" in log_format # Check actual values are in the arguments assert "anthropic" in log_args assert "kimi-coding" in log_args assert "kimi-k2.5" in log_args def test_fallback_event_with_error_logs_debug(self): """Test that fallback events with errors also log debug info.""" error = Exception("Rate limit exceeded") with patch("agent.fallback_router.logger") as mock_logger: log_fallback_event( from_provider="anthropic", to_provider="kimi-coding", to_model="kimi-k2.5", reason="rate_limit", error=error, ) # Both info and debug should be called mock_logger.info.assert_called_once() mock_logger.debug.assert_called_once() def test_fallback_chain_resolution_logged(self): """Test logging during full chain resolution.""" with patch("agent.fallback_router.logger") as mock_logger: # Simulate getting chain chain = get_auto_fallback_chain("anthropic") # Log each fallback step for i, fallback in enumerate(chain): log_fallback_event( from_provider="anthropic" if i == 0 else chain[i-1]["provider"], to_provider=fallback["provider"], to_model=fallback["model"], reason="chain_resolution", ) # Should have logged for each fallback assert mock_logger.info.call_count == len(chain) class TestFallbackAvailability: """Tests for fallback availability checking with credentials.""" def test_anthropic_available_with_api_key(self): """Test Anthropic is available when ANTHROPIC_API_KEY is set.""" with patch.dict(os.environ, {"ANTHROPIC_API_KEY": "test-key"}): config = {"provider": "anthropic", "model": "claude-3"} assert is_fallback_available(config) is True def test_anthropic_available_with_token(self): """Test Anthropic is available when ANTHROPIC_TOKEN is set.""" with patch.dict(os.environ, {"ANTHROPIC_TOKEN": "test-token"}): config = {"provider": "anthropic", "model": "claude-3"} assert is_fallback_available(config) is True def test_kimi_available_with_api_key(self): """Test kimi-coding is available when KIMI_API_KEY is set.""" with patch.dict(os.environ, {"KIMI_API_KEY": "test-key"}): config = {"provider": "kimi-coding", "model": "kimi-k2.5"} assert is_fallback_available(config) is True def test_kimi_available_with_api_token(self): """Test kimi-coding is available when KIMI_API_TOKEN is set.""" with patch.dict(os.environ, {"KIMI_API_TOKEN": "test-token"}): config = {"provider": "kimi-coding", "model": "kimi-k2.5"} assert is_fallback_available(config) is True def test_openrouter_available_with_key(self): """Test openrouter is available when OPENROUTER_API_KEY is set.""" with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}): config = {"provider": "openrouter", "model": "claude-sonnet-4"} assert is_fallback_available(config) is True def test_zai_available(self): """Test zai is available when ZAI_API_KEY is set.""" with patch.dict(os.environ, {"ZAI_API_KEY": "test-key"}): config = {"provider": "zai", "model": "glm-5"} assert is_fallback_available(config) is True def test_unconfigured_provider_not_available(self): """Test that providers without credentials are not available.""" with patch.dict(os.environ, {}, clear=True): providers = [ {"provider": "anthropic", "model": "claude-3"}, {"provider": "kimi-coding", "model": "kimi-k2.5"}, {"provider": "openrouter", "model": "claude-sonnet-4"}, {"provider": "zai", "model": "glm-5"}, ] for config in providers: assert is_fallback_available(config) is False, f"{config['provider']} should not be available" def test_invalid_config_not_available(self): """Test that invalid configs are not available.""" assert is_fallback_available({}) is False assert is_fallback_available({"provider": ""}) is False assert is_fallback_available({"model": "some-model"}) is False class TestAutoFallbackDecision: """Tests for automatic fallback decision logic.""" def test_anthropic_eager_fallback_no_error(self): """Test Anthropic falls back eagerly even without an error.""" with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "true"}): assert should_auto_fallback("anthropic") is True def test_quota_error_triggers_fallback_any_provider(self): """Test that quota errors trigger fallback for any provider.""" error = Exception("Rate limit exceeded") with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "true"}): # Even unknown providers should fallback on quota errors assert should_auto_fallback("unknown_provider", error=error) is True def test_non_quota_error_no_fallback_unknown_provider(self): """Test that non-quota errors don't trigger fallback for unknown providers.""" error = Exception("Some random error") with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "true"}): assert should_auto_fallback("unknown_provider", error=error) is False def test_auto_fallback_disabled_via_env(self): """Test auto-fallback can be disabled via environment variable.""" with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "false"}): assert should_auto_fallback("anthropic") is False with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "0"}): assert should_auto_fallback("anthropic") is False with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "off"}): assert should_auto_fallback("anthropic") is False def test_auto_fallback_disabled_via_param(self): """Test auto-fallback can be disabled via parameter.""" assert should_auto_fallback("anthropic", auto_fallback_enabled=False) is False def test_auto_fallback_enabled_variations(self): """Test various truthy values for HERMES_AUTO_FALLBACK.""" truthy_values = ["true", "1", "yes", "on"] for value in truthy_values: with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": value}): assert should_auto_fallback("anthropic") is True, f"Failed for {value}" class TestEndToEndFallbackChain: """End-to-end tests simulating real fallback scenarios.""" def test_anthropic_to_kimi_fallback_scenario(self): """Simulate complete fallback: Anthropic quota -> kimi-coding.""" # Step 1: Anthropic encounters a quota error anthropic_error = Exception("Rate limit exceeded: quota exceeded for model claude-3-5-sonnet") # Step 2: Verify it's detected as a quota error assert is_quota_error(anthropic_error, provider="anthropic") is True # Step 3: Check if auto-fallback should trigger with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "true"}): assert should_auto_fallback("anthropic", error=anthropic_error) is True # Step 4: Get fallback chain chain = get_auto_fallback_chain("anthropic") assert len(chain) > 0 # Step 5: Simulate kimi-coding being available with patch.dict(os.environ, {"KIMI_API_KEY": "test-kimi-key"}): available = filter_available_fallbacks(chain) assert len(available) > 0 assert available[0]["provider"] == "kimi-coding" # Step 6: Log the fallback event with patch("agent.fallback_router.logger") as mock_logger: log_fallback_event( from_provider="anthropic", to_provider=available[0]["provider"], to_model=available[0]["model"], reason="quota_exceeded", error=anthropic_error, ) mock_logger.info.assert_called_once() def test_full_chain_exhaustion_scenario(self): """Simulate scenario where entire fallback chain is exhausted.""" # Simulate Anthropic error error = Exception("Rate limit exceeded") with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "true"}): # Get chain chain = get_auto_fallback_chain("anthropic") # Simulate no providers available with patch.dict(os.environ, {}, clear=True): available = filter_available_fallbacks(chain) assert available == [] # Fallback should not be possible assert len(available) == 0 def test_chain_continues_on_provider_failure(self): """Test that chain continues when a fallback provider fails.""" with patch.dict(os.environ, {"HERMES_AUTO_FALLBACK": "true"}): chain = get_auto_fallback_chain("anthropic") # Simulate only openrouter available (kimi-coding not configured) with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}, clear=True): available = filter_available_fallbacks(chain) # Should have openrouter as available (skipping kimi-coding) assert len(available) >= 1 assert available[0]["provider"] == "openrouter" if __name__ == "__main__": pytest.main([__file__, "-v"])