"""codspeed benchmarks for HTTP client."""

import asyncio
import ssl
from collections.abc import Awaitable, Callable, Iterator
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, TypedDict

import pytest
from pytest_aiohttp import AiohttpClient, AiohttpServer
from yarl import URL

from aiohttp import hdrs, request, web
from aiohttp.test_utils import TestClient, TestServer

if TYPE_CHECKING:
    from pytest_codspeed import BenchmarkFixture
else:
    pytest_codspeed = pytest.importorskip("pytest_codspeed")
    BenchmarkFixture = pytest_codspeed.BenchmarkFixture


@pytest.fixture
def aiohttp_client_sync(
    event_loop: asyncio.AbstractEventLoop,
) -> Iterator[
    Callable[[web.Application], Awaitable[TestClient[web.Request, web.Application]]]
]:
    clients = []

    async def go(
        app: web.Application,
        *,
        server_kwargs: dict[str, Any] | None = None,
    ) -> TestClient[web.Request, web.Application]:
        server = TestServer(app)
        client = TestClient(server)
        await server.start_server(**(server_kwargs or {}))
        clients.append(client)
        return client

    yield go

    while clients:
        event_loop.run_until_complete(clients.pop().close())


class _ConnArgs(TypedDict, total=False):
    ssl: ssl.SSLContext


@dataclass(frozen=True)
class ConnectionType:
    s_kwargs: _ConnArgs
    c_kwargs: _ConnArgs


@pytest.fixture(params=("tcp", "ssl"), ids=("tcp", "ssl"))
def conn_type(
    request: pytest.FixtureRequest,
    ssl_ctx: ssl.SSLContext,
    client_ssl_ctx: ssl.SSLContext,
) -> ConnectionType:
    if request.param == "ssl":
        return ConnectionType(
            s_kwargs={"ssl": ssl_ctx},
            c_kwargs={"ssl": client_ssl_ctx},
        )
    return ConnectionType(s_kwargs={}, c_kwargs={})


@pytest.fixture
def aiohttp_server_sync(
    event_loop: asyncio.AbstractEventLoop,
) -> Iterator[AiohttpServer]:
    # TODO: Remove this fixture when async benchmarks are supported.
    servers = []

    async def go(
        app: web.Application,
        *,
        host: str = "127.0.0.1",
        port: int | None = None,
        **kwargs: Any,
    ) -> TestServer:
        server = TestServer(app, host=host, port=port)
        await server.start_server(**kwargs)
        servers.append(server)
        return server

    yield go

    while servers:
        event_loop.run_until_complete(servers.pop().close())


def test_one_hundred_simple_get_requests(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client_sync: AiohttpClient,
    benchmark: BenchmarkFixture,
    conn_type: ConnectionType,
) -> None:
    """Benchmark 100 simple GET requests."""
    message_count = 100

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client_sync(app, server_kwargs=conn_type.s_kwargs)
        for _ in range(message_count):
            await client.get("/", **conn_type.c_kwargs)
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_simple_get_requests_alternating_clients(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client_sync: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 simple GET requests with alternating clients."""
    message_count = 100

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client1 = await aiohttp_client_sync(app)
        client2 = await aiohttp_client_sync(app)
        for i in range(message_count):
            if i % 2 == 0:
                await client1.get("/")
            else:
                await client2.get("/")
        await client1.close()
        await client2.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_simple_get_requests_no_session(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_server_sync: AiohttpServer,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 simple GET requests without a session."""
    message_count = 100

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    app = web.Application()
    app.router.add_route("GET", "/", handler)
    server = event_loop.run_until_complete(aiohttp_server_sync(app))
    url = URL(f"http://{server.host}:{server.port}/")

    async def run_client_benchmark() -> None:
        for _ in range(message_count):
            async with request("GET", url):
                pass

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_simple_get_requests_multiple_methods_route(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 simple GET requests on a route with multiple methods."""
    message_count = 100

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    app = web.Application()
    # GET intentionally registered last to ensure time complexity
    # of the route lookup is benchmarked
    for method in ("DELETE", "HEAD", "OPTIONS", "PATCH", "POST", "PUT", "GET"):
        app.router.add_route(method, "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            await client.get("/")
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_get_requests_with_1024_chunked_payload(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client_sync: AiohttpClient,
    benchmark: BenchmarkFixture,
    conn_type: ConnectionType,
) -> None:
    """Benchmark 100 GET requests with a small payload of 1024 bytes."""
    message_count = 100
    payload = b"a" * 1024

    async def handler(request: web.Request) -> web.Response:
        resp = web.Response(body=payload)
        resp.enable_chunked_encoding()
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client_sync(app, server_kwargs=conn_type.s_kwargs)
        for _ in range(message_count):
            resp = await client.get("/", **conn_type.c_kwargs)
            await resp.read()
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_get_requests_with_30000_chunked_payload(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 GET requests with a payload of 30000 bytes."""
    message_count = 100
    payload = b"a" * 30000

    async def handler(request: web.Request) -> web.Response:
        resp = web.Response(body=payload)
        resp.enable_chunked_encoding()
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            await resp.read()
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_get_requests_with_1mb_chunked_payload(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client_sync: AiohttpClient,
    benchmark: BenchmarkFixture,
    conn_type: ConnectionType,
) -> None:
    """Benchmark 100 GET requests with a 1 MiB chunked payload using read."""
    message_count = 100
    payload = b"a" * 2**20

    async def handler(request: web.Request) -> web.Response:
        resp = web.Response(body=payload)
        resp.enable_chunked_encoding()
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client_sync(app, server_kwargs=conn_type.s_kwargs)
        for _ in range(message_count):
            resp = await client.get("/", **conn_type.c_kwargs)
            await resp.read()
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_get_requests_iter_chunks_on_1mb_chunked_payload(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 GET requests with a 1 MiB chunked payload using iter_chunks."""
    message_count = 100
    payload = b"a" * 2**20

    async def handler(request: web.Request) -> web.Response:
        resp = web.Response(body=payload)
        resp.enable_chunked_encoding()
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            async for _ in resp.content.iter_chunks():
                pass
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


@pytest.mark.usefixtures("parametrize_zlib_backend")
def test_get_request_with_251308_compressed_chunked_payload(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark compressed GET requests with a payload of 251308."""
    # This payload compresses to 251308 bytes
    payload = b"".join(
        [
            bytes((*range(0, i), *range(i, 0, -1)))
            for _ in range(255)
            for i in range(255)
        ]
    )

    async def handler(request: web.Request) -> web.Response:
        resp = web.Response(body=payload, zlib_executor_size=16384)
        resp.enable_compression()
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        resp = await client.get("/")
        await resp.read()
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_get_requests_with_1024_content_length_payload(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 GET requests with a small payload of 1024 bytes."""
    message_count = 100
    payload = b"a" * 1024
    headers = {hdrs.CONTENT_LENGTH: str(len(payload))}

    async def handler(request: web.Request) -> web.Response:
        return web.Response(body=payload, headers=headers)

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            await resp.read()
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_get_requests_with_30000_content_length_payload(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 GET requests with a payload of 30000 bytes."""
    message_count = 100
    payload = b"a" * 30000
    headers = {hdrs.CONTENT_LENGTH: str(len(payload))}

    async def handler(request: web.Request) -> web.Response:
        return web.Response(body=payload, headers=headers)

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            await resp.read()
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_get_requests_with_1mb_content_length_payload(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 GET requests with a content-length payload of 1 MiB."""
    message_count = 100
    payload = b"a" * 2**20
    headers = {hdrs.CONTENT_LENGTH: str(len(payload))}

    async def handler(request: web.Request) -> web.Response:
        return web.Response(body=payload, headers=headers)

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            await resp.read()
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_simple_post_requests(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 simple POST requests."""
    message_count = 100

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    app = web.Application()
    app.router.add_route("POST", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            await client.post("/", data=b"any")
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_one_hundred_json_post_requests(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 100 JSON POST requests that check the content-type."""
    message_count = 100

    async def handler(request: web.Request) -> web.Response:
        _ = request.content_type
        _ = request.charset
        return web.Response()

    app = web.Application()
    app.router.add_route("POST", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            await client.post("/", json={"key": "value"})
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_ten_streamed_responses_iter_any(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 10 streamed responses using iter_any."""
    message_count = 10
    data = b"x" * 65536  # 64 KiB chunk size

    async def handler(request: web.Request) -> web.StreamResponse:
        resp = web.StreamResponse()
        await resp.prepare(request)
        for _ in range(10):
            await resp.write(data)
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            async for _ in resp.content.iter_any():
                pass
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_ten_streamed_responses_iter_chunked_4096(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 10 streamed responses using iter_chunked 4096."""
    message_count = 10
    data = b"x" * 65536  # 64 KiB chunk size, 4096 iter_chunked

    async def handler(request: web.Request) -> web.StreamResponse:
        resp = web.StreamResponse()
        await resp.prepare(request)
        for _ in range(10):
            await resp.write(data)
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            async for _ in resp.content.iter_chunked(4096):
                pass
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_ten_streamed_responses_iter_chunked_1mb(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 10 streamed responses using iter_chunked 1 MiB."""
    message_count = 10
    MB = 2**20
    data = b"x" * 6 * MB

    async def handler(request: web.Request) -> web.StreamResponse:
        resp = web.StreamResponse()
        await resp.prepare(request)
        for _ in range(10):
            await resp.write(data)
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            async for _ in resp.content.iter_chunked(MB):
                pass
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


@pytest.mark.usefixtures("parametrize_zlib_backend")
def test_ten_compressed_responses_iter_chunked_1mb(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark compressed GET request read via large iter_chunked."""
    MB = 2**20
    data = b"x" * 10 * MB

    async def handler(request: web.Request) -> web.Response:
        resp = web.Response(body=data)
        resp.enable_compression()
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        resp = await client.get("/")
        async for _ in resp.content.iter_chunked(MB):
            pass
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())


def test_ten_streamed_responses_iter_chunks(
    event_loop: asyncio.AbstractEventLoop,
    aiohttp_client: AiohttpClient,
    benchmark: BenchmarkFixture,
) -> None:
    """Benchmark 10 streamed responses using iter_chunks."""
    message_count = 10
    data = b"x" * 2**20

    async def handler(request: web.Request) -> web.StreamResponse:
        resp = web.StreamResponse()
        await resp.prepare(request)
        for _ in range(10):
            await resp.write(data)
        return resp

    app = web.Application()
    app.router.add_route("GET", "/", handler)

    async def run_client_benchmark() -> None:
        client = await aiohttp_client(app)
        for _ in range(message_count):
            resp = await client.get("/")
            async for _ in resp.content.iter_chunks():
                pass
        await client.close()

    @benchmark
    def _run() -> None:
        event_loop.run_until_complete(run_client_benchmark())
