# HTTP client functional tests against aiohttp.web server

import asyncio

import datetime

import http.cookies

import io

import json

import logging

import pathlib

import socket

import ssl

import sys

import tarfile

import time

import zipfile

import zlib

from collections.abc import AsyncIterator, Awaitable, Callable

from contextlib import suppress

from typing import Any, NoReturn

from unittest import mock

try:

try:

import brotlicffi as brotli

except ImportError:

import brotli

except ImportError: # pragma: no cover

brotli = None

try:

from backports.zstd import ZstdCompressor

except ImportError:

ZstdCompressor = None # type: ignore[assignment,misc] # pragma: no cover

import pytest

import trustme

from multidict import MultiDict

from pytest_mock import MockerFixture

from yarl import URL, Query

import aiohttp

from aiohttp import Fingerprint, ServerFingerprintMismatch, hdrs, payload, web

from aiohttp.abc import AbstractResolver, ResolveResult

from aiohttp.client_exceptions import (

ClientResponseError,

InvalidURL,

InvalidUrlClientError,

InvalidUrlRedirectClientError,

NonHttpUrlClientError,

NonHttpUrlRedirectClientError,

SocketTimeoutError,

TooManyRedirects,

)

from aiohttp.client_reqrep import ClientRequest

from aiohttp.helpers import DEFAULT_CHUNK_SIZE

from aiohttp.payload import (

AsyncIterablePayload,

BufferedReaderPayload,

BytesIOPayload,

BytesPayload,

StringIOPayload,

StringPayload,

)

from aiohttp.pytest_plugin import AiohttpClient, AiohttpServer

from aiohttp.test_utils import TestClient, TestServer, unused_port

from aiohttp.typedefs import Handler

@pytest.fixture(autouse=True)

def cleanup(

cleanup_payload_pending_file_closes: None,

) -> None:

"""Ensure all pending file close operations complete during test teardown."""

@pytest.fixture

def here() -> pathlib.Path:

return pathlib.Path(__file__).parent

@pytest.fixture

def fname(here: pathlib.Path) -> pathlib.Path:

return here / "conftest.py"

@pytest.fixture

def headers_echo_client(

aiohttp_client: AiohttpClient,

) -> Callable[..., Awaitable[TestClient[web.Request, web.Application]]]:

"""Create a client with an app that echoes request headers as JSON."""

async def factory(**kwargs: Any) -> TestClient[web.Request, web.Application]:

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

return web.json_response({"headers": dict(request.headers)})

app = web.Application()

app.router.add_get("/", handler)

return await aiohttp_client(app, **kwargs)

return factory

async def test_keepalive_two_requests_success(aiohttp_client: AiohttpClient) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(body=b"OK")

app = web.Application()

app.router.add_route("GET", "/", handler)

connector = aiohttp.TCPConnector(limit=1)

client = await aiohttp_client(app, connector=connector)

async with client.get("/") as resp1:

await resp1.read()

async with client.get("/") as resp2:

await resp2.read()

assert client._session.connector is not None

assert 1 == len(client._session.connector._conns)

async def test_keepalive_after_head_requests_success(

aiohttp_client: AiohttpClient,

) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(body=b"OK")

cnt_conn_reuse = 0

async def on_reuseconn(session: object, ctx: object, params: object) -> None:

nonlocal cnt_conn_reuse

cnt_conn_reuse += 1

trace_config = aiohttp.TraceConfig()

trace_config._on_connection_reuseconn.append(on_reuseconn)

app = web.Application()

app.router.add_route("GET", "/", handler)

app.router.add_route("HEAD", "/", handler)

connector = aiohttp.TCPConnector(limit=1)

client = await aiohttp_client(

app, connector=connector, trace_configs=[trace_config]

)

async with client.head("/") as resp1:

await resp1.read()

async with client.get("/") as resp2:

await resp2.read()

assert 1 == cnt_conn_reuse

@pytest.mark.parametrize("status", (101, 204, 304))

async def test_keepalive_after_empty_body_status(

aiohttp_client: AiohttpClient, status: int

) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(status=status)

cnt_conn_reuse = 0

async def on_reuseconn(session: object, ctx: object, params: object) -> None:

nonlocal cnt_conn_reuse

cnt_conn_reuse += 1

trace_config = aiohttp.TraceConfig()

trace_config._on_connection_reuseconn.append(on_reuseconn)

app = web.Application()

app.router.add_route("GET", "/", handler)

connector = aiohttp.TCPConnector(limit=1)

client = await aiohttp_client(

app, connector=connector, trace_configs=[trace_config]

)

async with client.get("/") as resp1:

await resp1.read()

async with client.get("/") as resp2:

await resp2.read()

assert cnt_conn_reuse == 1

@pytest.mark.parametrize("status", (101, 204, 304))

async def test_keepalive_after_empty_body_status_stream_response(

aiohttp_client: AiohttpClient, status: int

) -> None:

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

stream_response = web.StreamResponse(status=status)

await stream_response.prepare(request)

return stream_response

cnt_conn_reuse = 0

async def on_reuseconn(session: object, ctx: object, params: object) -> None:

nonlocal cnt_conn_reuse

cnt_conn_reuse += 1

trace_config = aiohttp.TraceConfig()

trace_config._on_connection_reuseconn.append(on_reuseconn)

app = web.Application()

app.router.add_route("GET", "/", handler)

connector = aiohttp.TCPConnector(limit=1)

client = await aiohttp_client(

app, connector=connector, trace_configs=[trace_config]

)

async with client.get("/") as resp1:

await resp1.read()

async with client.get("/") as resp2:

await resp2.read()

assert cnt_conn_reuse == 1

async def test_keepalive_response_released(aiohttp_client: AiohttpClient) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(body=b"OK")

app = web.Application()

app.router.add_route("GET", "/", handler)

connector = aiohttp.TCPConnector(limit=1)

client = await aiohttp_client(app, connector=connector)

resp1 = await client.get("/")

resp1.release()

resp2 = await client.get("/")

resp2.release()

assert client._session.connector is not None

assert 1 == len(client._session.connector._conns)

async def test_upgrade_connection_not_released_after_read(

aiohttp_client: AiohttpClient,

) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(

status=101, headers={"Connection": "Upgrade", "Upgrade": "tcp"}

)

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/") as resp:

await resp.read()

assert resp.connection is not None

assert not resp.closed

async def test_keepalive_server_force_close_connection(

aiohttp_client: AiohttpClient,

) -> None:

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

body = await request.read()

assert b"" == body

response = web.Response(body=b"OK")

response.force_close()

return response

app = web.Application()

app.router.add_route("GET", "/", handler)

connector = aiohttp.TCPConnector(limit=1)

client = await aiohttp_client(app, connector=connector)

resp1 = await client.get("/")

resp1.close()

resp2 = await client.get("/")

resp2.close()

assert client._session.connector is not None

assert 0 == len(client._session.connector._conns)

async def test_keepalive_timeout_async_sleep(unused_port_socket: socket.socket) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(body=b"OK")

app = web.Application()

app.router.add_route("GET", "/", handler)

runner = web.AppRunner(app, tcp_keepalive=True, keepalive_timeout=0.001)

await runner.setup()

site = web.SockSite(runner, unused_port_socket)

await site.start()

host, port = unused_port_socket.getsockname()[:2]

try:

async with aiohttp.ClientSession() as sess:

resp1 = await sess.get(f"http://{host}:{port}/")

await resp1.read()

# wait for server keepalive_timeout

await asyncio.sleep(0.01)

resp2 = await sess.get(f"http://{host}:{port}/")

await resp2.read()

finally:

await asyncio.gather(runner.shutdown(), site.stop())

@pytest.mark.skipif(

sys.version_info[:2] == (3, 11),

reason="https://github.com/pytest-dev/pytest/issues/10763",

)

async def test_keepalive_timeout_sync_sleep(unused_port_socket: socket.socket) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(body=b"OK")

app = web.Application()

app.router.add_route("GET", "/", handler)

runner = web.AppRunner(app, tcp_keepalive=True, keepalive_timeout=0.001)

await runner.setup()

site = web.SockSite(runner, unused_port_socket)

await site.start()

host, port = unused_port_socket.getsockname()[:2]

try:

async with aiohttp.ClientSession() as sess:

resp1 = await sess.get(f"http://{host}:{port}/")

await resp1.read()

# wait for server keepalive_timeout

# time.sleep is a more challenging scenario than asyncio.sleep

time.sleep(0.01)

resp2 = await sess.get(f"http://{host}:{port}/")

await resp2.read()

finally:

await asyncio.gather(runner.shutdown(), site.stop())

async def test_release_early(aiohttp_client: AiohttpClient) -> None:

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

await request.read()

return web.Response(body=b"OK")

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

resp = await client.get("/")

assert resp.closed

await resp.wait_for_close()

assert client._session.connector is not None

assert 1 == len(client._session.connector._conns)

async def test_HTTP_304(aiohttp_client: AiohttpClient) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(status=304)

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/") as resp:

assert resp.status == 304

content = await resp.read()

assert content == b""

async def test_stream_request_on_server_eof(aiohttp_client: AiohttpClient) -> None:

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

return web.Response(text="OK", status=200)

app = web.Application()

app.add_routes([web.get("/", handler)])

app.add_routes([web.put("/", handler)])

client = await aiohttp_client(app)

async def data_gen() -> AsyncIterator[bytes]:

for _ in range(2): # pragma: no branch

yield b"just data"

await asyncio.sleep(0.1)

assert client.session.connector is not None

async with client.put("/", data=data_gen()) as resp:

assert 200 == resp.status

assert len(client.session.connector._acquired) == 1

conn = next(iter(client.session.connector._acquired))

async with client.get("/") as resp:

assert 200 == resp.status

# First connection should have been closed, otherwise server won't know if it

# received the full message.

conns = next(iter(client.session.connector._conns.values()))

assert len(conns) == 1

assert conns[0][0] is not conn

async def test_stream_request_on_server_eof_nested(

aiohttp_client: AiohttpClient,

) -> None:

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

return web.Response(text="OK", status=200)

app = web.Application()

app.add_routes([web.get("/", handler)])

app.add_routes([web.put("/", handler)])

client = await aiohttp_client(app)

async def data_gen() -> AsyncIterator[bytes]:

for _ in range(2): # pragma: no branch

yield b"just data"

await asyncio.sleep(0.1)

assert client.session.connector is not None

async with client.put("/", data=data_gen()) as resp:

first_conn = next(iter(client.session.connector._acquired))

assert 200 == resp.status

async with client.get("/") as resp2:

assert 200 == resp2.status

# Should be 2 separate connections

conns = next(iter(client.session.connector._conns.values()))

assert len(conns) == 1

assert first_conn is not None

assert not first_conn.is_connected()

assert first_conn is not conns[0][0]

async def test_HTTP_304_WITH_BODY(aiohttp_client: AiohttpClient) -> None:

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

body = await request.read()

assert b"" == body

return web.Response(body=b"test", status=304)

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/") as resp:

assert resp.status == 304

content = await resp.read()

assert content == b""

async def test_auto_header_user_agent(aiohttp_client: AiohttpClient) -> None:

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

assert "aiohttp" in request.headers["user-agent"]

return web.Response()

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/") as resp:

assert 200 == resp.status

async def test_skip_auto_headers_user_agent(aiohttp_client: AiohttpClient) -> None:

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

assert hdrs.USER_AGENT not in request.headers

return web.Response()

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/", skip_auto_headers=["user-agent"]) as resp:

assert 200 == resp.status

async def test_skip_default_auto_headers_user_agent(

aiohttp_client: AiohttpClient,

) -> None:

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

assert hdrs.USER_AGENT not in request.headers

return web.Response()

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app, skip_auto_headers=["user-agent"])

async with client.get("/") as resp:

assert 200 == resp.status

async def test_skip_auto_headers_content_type(aiohttp_client: AiohttpClient) -> None:

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

assert hdrs.CONTENT_TYPE not in request.headers

return web.Response()

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/", skip_auto_headers=["content-type"]) as resp:

assert 200 == resp.status

async def test_post_data_bytesio(aiohttp_client: AiohttpClient) -> None:

data = b"some buffer"

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

assert len(data) == request.content_length

val = await request.read()

assert data == val

return web.Response()

app = web.Application()

app.router.add_route("POST", "/", handler)

client = await aiohttp_client(app)

with io.BytesIO(data) as file_handle:

async with client.post("/", data=file_handle) as resp:

assert 200 == resp.status

async def test_post_data_with_bytesio_file(aiohttp_client: AiohttpClient) -> None:

data = b"some buffer"

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

post_data = await request.post()

assert ["file"] == list(post_data.keys())

file_field = post_data["file"]

assert isinstance(file_field, web.FileField)

assert data == await asyncio.to_thread(file_field.file.read)

return web.Response()

app = web.Application()

app.router.add_route("POST", "/", handler)

client = await aiohttp_client(app)

with io.BytesIO(data) as file_handle:

async with client.post("/", data={"file": file_handle}) as resp:

assert 200 == resp.status

async def test_post_data_stringio(aiohttp_client: AiohttpClient) -> None:

data = "some buffer"

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

assert len(data) == request.content_length

assert request.headers["CONTENT-TYPE"] == "text/plain; charset=utf-8"

val = await request.text()

assert data == val

return web.Response()

app = web.Application()

app.router.add_route("POST", "/", handler)

client = await aiohttp_client(app)

async with client.post("/", data=io.StringIO(data)) as resp:

assert 200 == resp.status

async def test_post_data_textio_encoding(aiohttp_client: AiohttpClient) -> None:

data = "текст"

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

assert request.headers["CONTENT-TYPE"] == "text/plain; charset=koi8-r"

val = await request.text()

assert data == val

return web.Response()

app = web.Application()

app.router.add_route("POST", "/", handler)

client = await aiohttp_client(app)

pl = aiohttp.TextIOPayload(io.StringIO(data), encoding="koi8-r")

async with client.post("/", data=pl) as resp:

assert 200 == resp.status

async def test_post_data_zipfile_filelike(aiohttp_client: AiohttpClient) -> None:

data = b"This is a zip file payload text file."

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

val = await request.read()

assert data == val, "Transmitted zipfile member failed to match original data."

return web.Response()

app = web.Application()

app.router.add_route("POST", "/", handler)

client = await aiohttp_client(app)

buf = io.BytesIO()

with zipfile.ZipFile(file=buf, mode="w") as zf:

with zf.open("payload1.txt", mode="w") as zip_filelike_writing:

zip_filelike_writing.write(data)

buf.seek(0)

zf = zipfile.ZipFile(file=buf, mode="r")

async with client.post("/", data=zf.open("payload1.txt")) as resp:

assert resp.status == 200

async def test_post_data_tarfile_filelike(aiohttp_client: AiohttpClient) -> None:

data = b"This is a tar file payload text file."

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

val = await request.read()

assert data == val, "Transmitted tarfile member failed to match original data."

return web.Response()

app = web.Application()

app.router.add_route("POST", "/", handler)

client = await aiohttp_client(app)

buf = io.BytesIO()

with tarfile.open(fileobj=buf, mode="w") as tf:

ti = tarfile.TarInfo(name="payload1.txt")

ti.size = len(data)

tf.addfile(tarinfo=ti, fileobj=io.BytesIO(data))

# Random-access tarfile.

buf.seek(0)

tf = tarfile.open(fileobj=buf, mode="r:")

async with client.post("/", data=tf.extractfile("payload1.txt")) as resp:

assert resp.status == 200

# Streaming tarfile.

buf.seek(0)

tf = tarfile.open(fileobj=buf, mode="r|")

for entry in tf:

async with client.post("/", data=tf.extractfile(entry)) as resp:

assert resp.status == 200

async def test_post_bytes_data_content_length_from_body(

aiohttp_client: AiohttpClient,

) -> None:

"""Test that Content-Length is set from body payload size when sending bytes."""

data = b"test payload data"

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

# Verify Content-Length header was set correctly

assert request.content_length == len(data)

assert request.headers.get("Content-Length") == str(len(data))

# Verify we can read the data

val = await request.read()

assert data == val

return web.Response()

app = web.Application()

app.router.add_route("POST", "/", handler)

client = await aiohttp_client(app)

# Send bytes data - this should trigger the code path where

# Content-Length is set from body.size in update_transfer_encoding

async with client.post("/", data=data) as resp:

assert resp.status == 200

async def test_post_custom_payload_without_content_length(

aiohttp_client: AiohttpClient,

) -> None:

"""Test that Content-Length is set from payload.size when not explicitly provided."""

data = b"custom payload data"

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

# Verify Content-Length header was set from payload size

assert request.content_length == len(data)

assert request.headers.get("Content-Length") == str(len(data))

# Verify we can read the data

val = await request.read()

assert data == val

return web.Response()

app = web.Application()

app.router.add_route("POST", "/", handler)

client = await aiohttp_client(app)

# Create a BytesPayload directly - this ensures we test the path

# where update_transfer_encoding sets Content-Length from body.size

bytes_payload = payload.BytesPayload(data)

# Don't set Content-Length header explicitly

async with client.post("/", data=bytes_payload) as resp:

assert resp.status == 200

async def test_ssl_client(

aiohttp_server: AiohttpServer,

ssl_ctx: ssl.SSLContext,

aiohttp_client: AiohttpClient,

client_ssl_ctx: ssl.SSLContext,

) -> None:

connector = aiohttp.TCPConnector(ssl=client_ssl_ctx)

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

return web.Response(text="Test message")

app = web.Application()

app.router.add_route("GET", "/", handler)

server = await aiohttp_server(app, ssl=ssl_ctx)

client = await aiohttp_client(server, connector=connector)

async with client.get("/") as resp:

assert resp.status == 200

txt = await resp.text()

assert txt == "Test message"

@pytest.mark.skipif(

sys.version_info < (3, 11), reason="ssl_shutdown_timeout requires Python 3.11+"

)

async def test_ssl_client_shutdown_timeout(

aiohttp_server: AiohttpServer,

ssl_ctx: ssl.SSLContext,

aiohttp_client: AiohttpClient,

client_ssl_ctx: ssl.SSLContext,

) -> None:

# Test that ssl_shutdown_timeout is properly used during connection closure

with pytest.warns(

DeprecationWarning, match="ssl_shutdown_timeout parameter is deprecated"

):

connector = aiohttp.TCPConnector(ssl=client_ssl_ctx, ssl_shutdown_timeout=0.1)

async def streaming_handler(request: web.Request) -> NoReturn:

# Create a streaming response that continuously sends data

response = web.StreamResponse()

await response.prepare(request)

# Keep sending data until connection is closed

while True:

await response.write(b"data chunk\n")

await asyncio.sleep(0.01) # Small delay between chunks

assert False, "not reached"

app = web.Application()

app.router.add_route("GET", "/stream", streaming_handler)

server = await aiohttp_server(app, ssl=ssl_ctx)

client = await aiohttp_client(server, connector=connector)

# Verify the connector has the correct timeout

assert connector._ssl_shutdown_timeout == 0.1

# Start a streaming request to establish SSL connection with active data transfer

resp = await client.get("/stream")

assert resp.status == 200

# Create a background task that continuously reads data

async def read_loop() -> None:

while True:

# Read "data chunk\n"

await resp.content.read(11)

read_task = asyncio.create_task(read_loop())

await asyncio.sleep(0) # Yield control to ensure read_task starts

# Record the time before closing

start_time = time.monotonic()

# Now close the connector while the stream is still active

# This will test the ssl_shutdown_timeout during an active connection

await connector.close()

# Verify the connection was closed within a reasonable time

# Should be close to ssl_shutdown_timeout (0.1s) but allow some margin

elapsed = time.monotonic() - start_time

assert elapsed < 0.3, f"Connection closure took too long: {elapsed}s"

read_task.cancel()

with suppress(asyncio.CancelledError):

await read_task

assert read_task.done(), "Read task should be cancelled after connection closure"

async def test_ssl_client_alpn(

aiohttp_server: AiohttpServer,

aiohttp_client: AiohttpClient,

ssl_ctx: ssl.SSLContext,

) -> None:

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

assert request.transport is not None

sslobj = request.transport.get_extra_info("ssl_object")

return web.Response(text=sslobj.selected_alpn_protocol())

app = web.Application()

app.router.add_route("GET", "/", handler)

ssl_ctx.set_alpn_protocols(("http/1.1",))

server = await aiohttp_server(app, ssl=ssl_ctx)

connector = aiohttp.TCPConnector(ssl=False)

client = await aiohttp_client(server, connector=connector)

async with client.get("/") as resp:

assert resp.status == 200

txt = await resp.text()

assert txt == "http/1.1"

async def test_tcp_connector_fingerprint_ok(

aiohttp_server: AiohttpServer,

aiohttp_client: AiohttpClient,

ssl_ctx: ssl.SSLContext,

tls_certificate_fingerprint_sha256: bytes,

) -> None:

tls_fingerprint = Fingerprint(tls_certificate_fingerprint_sha256)

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

return web.Response(text="Test message")

connector = aiohttp.TCPConnector(ssl=tls_fingerprint)

app = web.Application()

app.router.add_route("GET", "/", handler)

server = await aiohttp_server(app, ssl=ssl_ctx)

client = await aiohttp_client(server, connector=connector)

async with client.get("/") as resp:

assert resp.status == 200

async def test_tcp_connector_fingerprint_fail(

aiohttp_server: AiohttpServer,

aiohttp_client: AiohttpClient,

ssl_ctx: ssl.SSLContext,

tls_certificate_fingerprint_sha256: bytes,

) -> None:

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

assert False

bad_fingerprint = b"\x00" * len(tls_certificate_fingerprint_sha256)

connector = aiohttp.TCPConnector(ssl=Fingerprint(bad_fingerprint))

app = web.Application()

app.router.add_route("GET", "/", handler)

server = await aiohttp_server(app, ssl=ssl_ctx)

client = await aiohttp_client(server, connector=connector)

with pytest.raises(ServerFingerprintMismatch) as cm:

await client.get("/")

exc = cm.value

assert exc.expected == bad_fingerprint

assert exc.got == tls_certificate_fingerprint_sha256

async def test_format_task_get(aiohttp_server: AiohttpServer) -> None:

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

return web.Response(body=b"OK")

app = web.Application()

app.router.add_route("GET", "/", handler)

server = await aiohttp_server(app)

client = aiohttp.ClientSession()

task = asyncio.create_task(client.get(server.make_url("/")))

assert f"{task}".startswith("<Task pending")

resp = await task

resp.close()

await client.close()

async def test_str_params(aiohttp_client: AiohttpClient) -> None:

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

assert "q=t est" in request.rel_url.query_string

return web.Response()

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/", params="q=t+est") as resp:

assert 200 == resp.status

async def test_params_and_query_string(aiohttp_client: AiohttpClient) -> None:

"""Test combining params with an existing query_string."""

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

assert request.rel_url.query_string == "q=abc&q=test&d=dog"

return web.Response()

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/?q=abc", params="q=test&d=dog") as resp:

assert resp.status == 200

@pytest.mark.parametrize("params", [None, "", {}, MultiDict()])

async def test_empty_params_and_query_string(

aiohttp_client: AiohttpClient, params: Query

) -> None:

"""Test combining empty params with an existing query_string."""

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

assert request.rel_url.query_string == "q=abc"

return web.Response()

app = web.Application()

app.router.add_route("GET", "/", handler)

client = await aiohttp_client(app)

async with client.get("/?q=abc", params=params) as resp:

assert resp.status == 200

async def test_drop_params_on_redirect(aiohttp_client: AiohttpClient) -> None:

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

return web.Response(status=301, headers={"Location": "/ok?a=redirect"})

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

assert request.rel_url.query_string == "a=redirect"

return web.Response(status=200)

app = web.Application()

app.router.add_route("GET", "/ok", handler_ok)

app.router.add_route("GET", "/redirect", handler_redirect)

client = await aiohttp_client(app)

async with client.get("/redirect", params={"a": "initial"}) as resp:

assert resp.status == 200

async def test_drop_fragment_on_redirect(aiohttp_client: AiohttpClient) -> None:

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

return web.Response(status=301, headers={"Location": "/ok#fragment"})

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

return web.Response(status=200)

app = web.Application()

app.router.add_route("GET", "/ok", handler_ok)

app.router.add_route("GET", "/redirect", handler_redirect)

client = await aiohttp_client(app)

async with client.get("/redirect") as resp:

assert resp.status == 200

assert resp.url.path == "/ok"

async def test_drop_fragment(aiohttp_client: AiohttpClient) -> None:

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

return web.Response(status=200)

app = web.Application()

app.router.add_route("GET", "/ok", handler_ok)

client = await aiohttp_client(app)

async with client.get("/ok#fragment") as resp:

assert resp.status == 200

assert resp.url.path == "/ok"

async def test_history(aiohttp_client: AiohttpClient) -> None:

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

return web.Response(status=301, headers={"Location": "/ok"})

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

return web.Response(status=200)

app = web.Application()

app.router.add_route("GET", "/ok", handler_ok)

app.router.add_route("GET", "/redirect", handler_redirect)

client = await aiohttp_client(app)

async with client.get("/ok") as resp:

assert len(resp.history) == 0

assert resp.status == 200

async with client.get("/redirect") as resp_redirect:

assert len(resp_redirect.history) == 1

assert resp_redirect.history[0].status == 301

assert resp_redirect.status == 200

async def test_keepalive_closed_by_server(aiohttp_client: AiohttpClient) -> None:

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

body = await request.read()

assert b"" == body

resp = web.Response(body=b"OK")