feat: implement automatic kimi-coding fallback on quota errors

tests/test_fallback_router.py

@@ -0,0 +1,280 @@
+"""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)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])