learningcircuit--local-deep-research/tests/rate_limiting/test_llm_rate_limiting.py

"""
Tests for LLM-specific rate limit detection.

Tests cover:
- Rate limit error detection from HTTP 429
- Rate limit error detection from message patterns
- Retry-after extraction from headers and messages
"""

from unittest.mock import Mock


class TestLLMRateLimitDetection:
    """Tests for LLM rate limit error detection."""

    def test_is_llm_rate_limit_error_http_429(self):
        """Detect rate limit from HTTP 429 status code."""
        from local_deep_research.web_search_engines.rate_limiting.llm.detection import (
            is_llm_rate_limit_error,
        )

        # Create mock error with 429 response
        error = Mock()
        error.response = Mock()
        error.response.status_code = 429
        error.__str__ = lambda self: "HTTP 429 Too Many Requests"
        error.__class__.__name__ = "HTTPError"
        error.__class__.__module__ = "requests"

        result = is_llm_rate_limit_error(error)

        assert result is True

    def test_is_llm_rate_limit_error_message_patterns(self):
        """Detect rate limit from error message patterns."""
        from local_deep_research.web_search_engines.rate_limiting.llm.detection import (
            is_llm_rate_limit_error,
        )

        # Test various rate limit message patterns
        patterns = [
            "Rate limit exceeded",
            "rate_limit_error",
            "Too many requests",
            "Quota exceeded",
            "Resource has been exhausted",
            "Please try again later",
            "429 error occurred",
        ]

        for pattern in patterns:
            error = Exception(pattern)
            result = is_llm_rate_limit_error(error)
            assert result is True, f"Failed to detect pattern: {pattern}"

    def test_is_llm_rate_limit_error_not_rate_limit(self):
        """Non-rate-limit errors should return False."""
        from local_deep_research.web_search_engines.rate_limiting.llm.detection import (
            is_llm_rate_limit_error,
        )

        # Regular errors
        error = Exception("Connection timeout")
        assert is_llm_rate_limit_error(error) is False

        error = Exception("Invalid API key")
        assert is_llm_rate_limit_error(error) is False

        error = Exception("Server error 500")
        assert is_llm_rate_limit_error(error) is False

    def test_extract_retry_after_header(self):
        """Extract retry time from Retry-After header."""
        from local_deep_research.web_search_engines.rate_limiting.llm.detection import (
            extract_retry_after,
        )

        # Create mock error with Retry-After header
        error = Mock()
        error.response = Mock()
        error.response.headers = {"Retry-After": "30"}
        error.__str__ = lambda self: "Rate limit error"

        result = extract_retry_after(error)

        assert result == 30.0

    def test_extract_retry_after_message(self):
        """Extract retry time from error message."""
        from local_deep_research.web_search_engines.rate_limiting.llm.detection import (
            extract_retry_after,
        )

        # Error with retry time in message
        error = Exception("Rate limited. Please try again in 45 seconds")

        result = extract_retry_after(error)

        assert result == 45.0

        # Another pattern
        error = Exception("Wait 60 seconds before retrying")
        result = extract_retry_after(error)
        assert result == 60.0

    def test_extract_retry_after_not_found(self):
        """Return 0 when no retry time is specified."""
        from local_deep_research.web_search_engines.rate_limiting.llm.detection import (
            extract_retry_after,
        )

        error = Exception("Rate limit exceeded")

        result = extract_retry_after(error)

        assert result == 0