import asyncio
import time
from unittest.mock import AsyncMock, patch

import httpx
import openai
import pytest
from httpx import Request

import free_claude_code.providers.rate_limit as rate_limit_module
from free_claude_code.providers.failure_policy import (
    retryable_upstream_status,
    retryable_upstream_transport_error,
)
from free_claude_code.providers.rate_limit import (
    DEFAULT_UPSTREAM_MAX_RETRIES,
    UPSTREAM_TRANSIENT_TOTAL_ATTEMPTS,
    ProviderRateLimiter,
)


def test_upstream_transient_retry_total_attempts_is_five() -> None:
    assert UPSTREAM_TRANSIENT_TOTAL_ATTEMPTS == 5
    assert DEFAULT_UPSTREAM_MAX_RETRIES == 4


def _statusless_api_error(message: str, body: object | None) -> openai.APIError:
    return openai.APIError(message, request=Request("POST", "http://x"), body=body)


def test_retryable_upstream_status_reads_statusless_api_error_body_status() -> None:
    exc = _statusless_api_error(
        "stream embedded error",
        {"error": {"message": "internal failure", "code": 500}},
    )

    assert retryable_upstream_status(exc) == 500


def test_retryable_upstream_status_reads_statusless_resource_exhausted_text() -> None:
    exc = _statusless_api_error(
        "ResourceExhausted: limit reached while generating response",
        {"error": {"message": "ResourceExhausted: limit reached"}},
    )

    assert retryable_upstream_status(exc) == 503


def test_retryable_upstream_transport_error_classifies_connection_failures() -> None:
    request = Request("POST", "http://x")
    assert retryable_upstream_transport_error(
        openai.APIConnectionError(request=request)
    )
    assert retryable_upstream_transport_error(httpx.ConnectError("connect failed"))
    assert retryable_upstream_transport_error(httpx.ReadError("read failed"))
    assert retryable_upstream_transport_error(httpx.WriteError("write failed"))
    assert retryable_upstream_transport_error(
        httpx.RemoteProtocolError("server disconnected")
    )
    assert retryable_upstream_transport_error(TimeoutError("timed out"))


def test_retryable_upstream_transport_error_rejects_request_errors() -> None:
    from httpx import Response

    request = Request("POST", "http://x")
    assert not retryable_upstream_transport_error(
        openai.BadRequestError(
            "bad request",
            response=Response(400, request=request),
            body={},
        )
    )
    assert not retryable_upstream_transport_error(
        httpx.HTTPStatusError(
            "bad request",
            request=request,
            response=Response(400, request=request),
        )
    )


class TestProviderRateLimiter:
    """Tests for providers.rate_limit.ProviderRateLimiter."""

    @pytest.mark.asyncio
    async def test_proactive_throttling(self):
        """
        Test proactive throttling.
        Logic ported from verify_provider_limiter.py
        """
        # Re-init with tight limits: 1 request per 0.25 second
        limiter = ProviderRateLimiter(rate_limit=1, rate_window=0.25)

        start_time = time.monotonic()

        async def call_limiter():
            await limiter.wait_if_blocked()
            return time.monotonic()

        # 5 requests.
        # R0 -> 0s
        # R1 -> 0.25s
        # R2 -> 0.50s
        # R3 -> 0.75s
        # R4 -> 1.00s
        results = [await call_limiter() for _ in range(5)]

        total_time = time.monotonic() - start_time

        assert len(results) == 5
        # Should take at least ~1.0s
        assert total_time >= 0.9, f"Throttling failed, took too fast: {total_time:.2f}s"

    @pytest.mark.asyncio
    async def test_reactive_blocking(self):
        """
        Test reactive blocking when a deadline is extended.
        Logic ported from verify_provider_limiter.py
        """
        limiter = ProviderRateLimiter()

        start_time = time.monotonic()

        # Manually block for 1.5s
        block_time = 1.5
        limiter.extend_reactive_block(block_time)

        assert limiter.is_blocked()

        async def call_limiter():
            return await limiter.wait_if_blocked()

        # Run 2 calls concurrently
        # They should both wait for the block time
        results = await asyncio.gather(call_limiter(), call_limiter())

        total_time = time.monotonic() - start_time

        # Both should report having waited reactively
        assert all(results) is True
        assert total_time >= block_time - 0.1, (
            f"Reactive block failed, took {total_time:.2f}s"
        )

    @pytest.mark.asyncio
    async def test_reactive_block_at_proactive_commit_retries_without_reserving(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)
        proactive_attempts = 0
        proactive_reservations = 0
        reactive_checks = 0

        async def acquire_if_allowed(_allowed) -> bool:
            nonlocal proactive_attempts, proactive_reservations
            proactive_attempts += 1
            if proactive_attempts == 1:
                return False
            proactive_reservations += 1
            return True

        async def wait_reactively() -> bool:
            nonlocal reactive_checks
            reactive_checks += 1
            return reactive_checks == 2

        monkeypatch.setattr(
            limiter._proactive_limiter,
            "acquire_if",
            acquire_if_allowed,
        )
        monkeypatch.setattr(
            limiter,
            "_wait_for_reactive_block",
            wait_reactively,
        )

        assert await limiter.wait_if_blocked() is True
        assert proactive_attempts == 2
        assert proactive_reservations == 1
        assert reactive_checks == 2

    @pytest.mark.asyncio
    async def test_reactive_wait_rechecks_an_extended_deadline(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        now = 0.0
        sleep_delays: list[float] = []
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        async def advance_time(delay: float) -> None:
            nonlocal now
            sleep_delays.append(delay)
            if len(sleep_delays) == 1:
                now = 5.0
                limiter.extend_reactive_block(10.0)
                now = 10.0
                return
            now += delay

        monkeypatch.setattr(rate_limit_module.time, "monotonic", lambda: now)
        monkeypatch.setattr(rate_limit_module.asyncio, "sleep", advance_time)
        monkeypatch.setattr(
            limiter._proactive_limiter,
            "acquire_if",
            AsyncMock(return_value=True),
        )
        limiter.extend_reactive_block(10.0)

        assert await limiter.wait_if_blocked() is True
        assert sleep_delays == [10.0, 5.0]
        assert limiter.is_blocked() is False

    @pytest.mark.asyncio
    async def test_reactive_wait_propagates_cancellation(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)
        sleep_started = asyncio.Event()

        async def wait_forever(_delay: float) -> None:
            sleep_started.set()
            await asyncio.Event().wait()

        monkeypatch.setattr(rate_limit_module.asyncio, "sleep", wait_forever)
        limiter.extend_reactive_block(60.0)
        waiter = asyncio.create_task(limiter.wait_if_blocked())
        await sleep_started.wait()

        waiter.cancel()

        with pytest.raises(asyncio.CancelledError):
            await waiter
        assert limiter.is_blocked() is True

    @pytest.mark.asyncio
    async def test_proactive_wait_propagates_cancellation(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)
        acquire_started = asyncio.Event()

        async def wait_forever(_allowed) -> bool:
            acquire_started.set()
            await asyncio.Event().wait()
            return True

        monkeypatch.setattr(limiter._proactive_limiter, "acquire_if", wait_forever)
        waiter = asyncio.create_task(limiter.wait_if_blocked())
        await acquire_started.wait()

        waiter.cancel()

        with pytest.raises(asyncio.CancelledError):
            await waiter

    def test_shorter_reactive_backoff_does_not_shorten_existing_deadline(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        now = 100.0
        monkeypatch.setattr(rate_limit_module.time, "monotonic", lambda: now)
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        limiter.extend_reactive_block(10.0)
        now = 102.0
        limiter.extend_reactive_block(1.0)

        assert limiter.remaining_wait() == 8.0

    @pytest.mark.parametrize("seconds", [0.0, -1.0])
    def test_reactive_backoff_duration_must_be_positive(self, seconds: float) -> None:
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        with pytest.raises(ValueError, match="reactive block duration must be > 0"):
            limiter.extend_reactive_block(seconds)

    @pytest.mark.asyncio
    async def test_remaining_wait_when_not_blocked(self):
        """remaining_wait() should return 0 when not blocked."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)
        assert limiter.remaining_wait() == 0

    @pytest.mark.asyncio
    async def test_remaining_wait_decreases(self):
        """remaining_wait() should decrease over time."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)
        limiter.extend_reactive_block(2.0)

        wait1 = limiter.remaining_wait()
        assert wait1 > 1.5

        await asyncio.sleep(0.5)
        wait2 = limiter.remaining_wait()
        assert wait2 < wait1

    @pytest.mark.asyncio
    async def test_is_blocked_false_initially(self):
        """is_blocked() should be False for a fresh limiter."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)
        assert limiter.is_blocked() is False

    @pytest.mark.asyncio
    async def test_high_rate_limit_no_throttling(self):
        """Very high rate limit should not cause throttling."""
        limiter = ProviderRateLimiter(rate_limit=10000, rate_window=60)

        start = time.monotonic()
        for _ in range(20):
            await limiter.wait_if_blocked()
        duration = time.monotonic() - start

        # 20 requests with 10000 limit should be near-instant
        assert duration < 1.0, f"High rate limit caused throttling: {duration:.2f}s"

    @pytest.mark.asyncio
    async def test_instances_are_independent(self):
        """Each provider limiter owns independent reactive state."""
        limiter1 = ProviderRateLimiter(rate_limit=10, rate_window=1)
        limiter2 = ProviderRateLimiter(rate_limit=10, rate_window=1)

        limiter1.extend_reactive_block(1)

        assert limiter1 is not limiter2
        assert limiter1.is_blocked() is True
        assert limiter2.is_blocked() is False

    @pytest.mark.asyncio
    async def test_wait_if_blocked_returns_false_when_not_blocked(self):
        """wait_if_blocked should return False when not reactively blocked."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)
        result = await limiter.wait_if_blocked()
        assert result is False

    @pytest.mark.asyncio
    async def test_proactive_strict_rolling_window(self):
        """
        Proactive limiter should enforce a strict rolling window:
        for any i, t[i+rate_limit] - t[i] >= rate_window (within tolerance).
        """
        rate_limit = 2
        rate_window = 0.5
        limiter = ProviderRateLimiter(rate_limit=rate_limit, rate_window=rate_window)

        acquired: list[float] = []

        async def acquire():
            await limiter.wait_if_blocked()
            acquired.append(time.monotonic())

        # Trigger concurrency; without strict rolling windows, this can burst.
        await asyncio.gather(*(acquire() for _ in range(5)))

        acquired.sort()
        assert len(acquired) == 5

        tolerance = 0.05
        for i in range(len(acquired) - rate_limit):
            assert acquired[i + rate_limit] - acquired[i] >= rate_window - tolerance, (
                f"Rolling window violated at i={i}: "
                f"dt={acquired[i + rate_limit] - acquired[i]:.3f}s"
            )

    @pytest.mark.asyncio
    async def test_init_rate_limit_zero_raises(self):
        """rate_limit <= 0 raises ValueError."""
        with pytest.raises(ValueError, match="rate_limit must be > 0"):
            ProviderRateLimiter(rate_limit=0, rate_window=60)

    @pytest.mark.asyncio
    async def test_init_rate_window_zero_raises(self):
        """rate_window <= 0 raises ValueError."""
        with pytest.raises(ValueError, match="rate_window must be > 0"):
            ProviderRateLimiter(rate_limit=10, rate_window=0)

    @pytest.mark.asyncio
    async def test_execute_with_retry_exhaust_retries_raises(self):
        """When all 429 retries exhausted, last exception is raised."""
        import openai
        from httpx import Request, Response

        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        def make_429():
            return openai.RateLimitError(
                "rate limited",
                response=Response(429, request=Request("POST", "http://x")),
                body={},
            )

        async def fail():
            raise make_429()

        with pytest.raises(openai.RateLimitError):
            await limiter.execute_with_retry(
                fail, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
            )

    @pytest.mark.asyncio
    async def test_execute_with_retry_succeeds_on_retry(self):
        """429 then success returns result."""
        import openai
        from httpx import Request, Response

        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        def make_429():
            return openai.RateLimitError(
                "rate limited",
                response=Response(429, request=Request("POST", "http://x")),
                body={},
            )

        call_count = 0

        async def fail_then_ok():
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise make_429()
            return "ok"

        result = await limiter.execute_with_retry(
            fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
        )
        assert result == "ok"
        assert call_count == 2

    @pytest.mark.asyncio
    async def test_execute_with_retry_succeeds_on_httpx_429(self):
        """HTTP 429 as httpx.HTTPStatusError then success returns result."""
        import httpx
        from httpx import Request, Response

        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        call_count = 0

        async def fail_then_ok():
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                r = Response(429, request=Request("POST", "http://x"), text="slow")
                raise httpx.HTTPStatusError(
                    "Too Many Requests", request=r.request, response=r
                )
            return "ok"

        result = await limiter.execute_with_retry(
            fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
        )
        assert result == "ok"
        assert call_count == 2

    @pytest.mark.parametrize("status_code", [500, 502, 503, 504])
    @pytest.mark.asyncio
    async def test_execute_with_retry_succeeds_on_openai_internal_server_error_5xx(
        self, status_code
    ):
        """5xx as openai.InternalServerError then success."""
        import openai
        from httpx import Request, Response

        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        def make_upstream_error():
            return openai.InternalServerError(
                "unavailable",
                response=Response(status_code, request=Request("POST", "http://x")),
                body={},
            )

        call_count = 0

        async def fail_then_ok():
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise make_upstream_error()
            return "ok"

        result = await limiter.execute_with_retry(
            fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
        )
        assert result == "ok"
        assert call_count == 2

    @pytest.mark.parametrize("status_code", [500, 502, 503, 504])
    @pytest.mark.asyncio
    async def test_execute_with_retry_succeeds_on_httpx_5xx(self, status_code):
        """HTTP 5xx as httpx.HTTPStatusError then success."""
        import httpx
        from httpx import Request, Response

        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        call_count = 0

        async def fail_then_ok():
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                r = Response(
                    status_code, request=Request("POST", "http://x"), text="error"
                )
                raise httpx.HTTPStatusError(
                    "Server Error", request=r.request, response=r
                )
            return "ok"

        result = await limiter.execute_with_retry(
            fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
        )
        assert result == "ok"
        assert call_count == 2

    @pytest.mark.asyncio
    async def test_execute_with_retry_succeeds_on_statusless_transient_api_error(self):
        """Status-less SDK APIError transient markers participate in backoff retry."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        call_count = 0

        async def fail_then_ok():
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise _statusless_api_error(
                    "ResourceExhausted: limit reached while generating response",
                    {"error": {"message": "ResourceExhausted: limit reached"}},
                )
            return "ok"

        result = await limiter.execute_with_retry(
            fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
        )

        assert result == "ok"
        assert call_count == 2

    @pytest.mark.asyncio
    async def test_execute_with_retry_succeeds_on_openai_connection_retry(self):
        """Pre-response OpenAI SDK connection errors retry then succeed."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        call_count = 0
        request = Request("POST", "http://x")

        async def fail_then_ok():
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise openai.APIConnectionError(request=request)
            return "ok"

        with (
            patch.object(limiter, "extend_reactive_block") as extend_block,
            patch("asyncio.sleep", new_callable=AsyncMock) as sleep,
        ):
            result = await limiter.execute_with_retry(
                fail_then_ok,
                max_retries=2,
                base_delay=0.01,
                max_delay=0.1,
                jitter=0,
            )

        assert result == "ok"
        assert call_count == 2
        extend_block.assert_not_called()
        sleep.assert_awaited_once()

    @pytest.mark.asyncio
    async def test_execute_with_retry_succeeds_on_httpx_transport_retry(self):
        """Pre-response HTTPX transport errors retry then succeed."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        call_count = 0

        async def fail_then_ok():
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise httpx.RemoteProtocolError("server disconnected")
            return "ok"

        with (
            patch.object(limiter, "extend_reactive_block") as extend_block,
            patch("asyncio.sleep", new_callable=AsyncMock) as sleep,
        ):
            result = await limiter.execute_with_retry(
                fail_then_ok,
                max_retries=2,
                base_delay=0.01,
                max_delay=0.1,
                jitter=0,
            )

        assert result == "ok"
        assert call_count == 2
        extend_block.assert_not_called()
        sleep.assert_awaited_once()

    @pytest.mark.asyncio
    async def test_execute_with_retry_exhaust_transport_error_attempts(self):
        """Transport retries exhaust after the shared 5 total attempts."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        call_count = 0

        async def always_fail():
            nonlocal call_count
            call_count += 1
            raise httpx.ConnectError("connect failed")

        with (
            patch.object(limiter, "extend_reactive_block") as extend_block,
            patch("asyncio.sleep", new_callable=AsyncMock) as sleep,
            pytest.raises(httpx.ConnectError),
        ):
            await limiter.execute_with_retry(
                always_fail,
                base_delay=0.01,
                max_delay=0.1,
                jitter=0,
            )

        assert call_count == UPSTREAM_TRANSIENT_TOTAL_ATTEMPTS
        extend_block.assert_not_called()
        assert sleep.await_count == DEFAULT_UPSTREAM_MAX_RETRIES

    @pytest.mark.parametrize("status_code", [500, 502, 503, 504])
    @pytest.mark.asyncio
    async def test_execute_with_retry_exhaust_openai_5xx_raises(self, status_code):
        """When all 5xx retries exhausted (OpenAI SDK), last InternalServerError is raised."""
        import openai
        from httpx import Request, Response

        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        exc = openai.InternalServerError(
            "unavailable",
            response=Response(status_code, request=Request("POST", "http://x")),
            body={},
        )

        async def always_fail():
            raise exc

        with pytest.raises(openai.InternalServerError):
            await limiter.execute_with_retry(
                always_fail, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0
            )

    @pytest.mark.asyncio
    async def test_execute_with_retry_httpx_400_raises_immediately(self):
        """Non-retryable 4xx is not wrapped by execute_with_retry loop."""
        import httpx
        from httpx import Request, Response

        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60)

        call_count = 0

        async def bad_request():
            nonlocal call_count
            call_count += 1
            r = Response(400, request=Request("POST", "http://x"), text="bad request")
            raise httpx.HTTPStatusError("Bad Request", request=r.request, response=r)

        with pytest.raises(httpx.HTTPStatusError):
            await limiter.execute_with_retry(
                bad_request,
                max_retries=2,
                base_delay=0.01,
                max_delay=0.1,
                jitter=0,
            )

        assert call_count == 1

    @pytest.mark.asyncio
    async def test_max_concurrency_zero_raises(self):
        """max_concurrency <= 0 raises ValueError."""
        with pytest.raises(ValueError, match="max_concurrency must be > 0"):
            ProviderRateLimiter(rate_limit=10, rate_window=60, max_concurrency=0)

    @pytest.mark.asyncio
    async def test_concurrency_slot_limits_simultaneous_streams(self):
        """At most max_concurrency streams can hold a slot simultaneously."""
        max_concurrency = 2
        limiter = ProviderRateLimiter(
            rate_limit=100, rate_window=60, max_concurrency=max_concurrency
        )

        peak_concurrent = 0
        current_concurrent = 0
        lock = asyncio.Lock()

        async def stream_task(hold_time: float) -> None:
            nonlocal peak_concurrent, current_concurrent
            async with limiter.concurrency_slot():
                async with lock:
                    current_concurrent += 1
                    if current_concurrent > peak_concurrent:
                        peak_concurrent = current_concurrent
                await asyncio.sleep(hold_time)
                async with lock:
                    current_concurrent -= 1

        # Launch 5 tasks that each hold the slot; only 2 can be active at once
        await asyncio.gather(*(stream_task(0.05) for _ in range(5)))

        assert peak_concurrent <= max_concurrency, (
            f"Concurrency exceeded: peak={peak_concurrent}, max={max_concurrency}"
        )

    @pytest.mark.asyncio
    async def test_concurrency_slot_releases_on_exception(self):
        """Slot is released even when the body raises an exception."""
        limiter = ProviderRateLimiter(rate_limit=100, rate_window=60, max_concurrency=1)
        assert limiter._concurrency_sem is not None

        with pytest.raises(RuntimeError):
            async with limiter.concurrency_slot():
                raise RuntimeError("boom")

        # Semaphore value should be restored (1 available again)
        assert limiter._concurrency_sem._value == 1

    @pytest.mark.asyncio
    async def test_constructor_sets_max_concurrency(self):
        """Constructor applies max_concurrency to an independent limiter."""
        limiter = ProviderRateLimiter(rate_limit=10, rate_window=60, max_concurrency=3)
        assert limiter._concurrency_sem is not None
        assert limiter._concurrency_sem._value == 3

    @pytest.mark.asyncio
    async def test_provider_owned_instances_are_isolated(self):
        """Independent provider limiters do not share reactive block state."""
        nim = ProviderRateLimiter(rate_limit=10, rate_window=60)
        openrouter = ProviderRateLimiter(rate_limit=20, rate_window=30)

        assert nim is not openrouter
        nim.extend_reactive_block(1.0)

        assert nim.is_blocked() is True
        assert openrouter.is_blocked() is False