1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
"""codspeed benchmarks for client requests."""
import asyncio
from http.cookies import Morsel
from typing import Union
from pytest_codspeed import BenchmarkFixture
from yarl import URL
from aiohttp.client_reqrep import ClientRequest
from aiohttp.http_writer import HttpVersion11
def test_client_request_update_cookies(
loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture
) -> None:
req = ClientRequest("get", URL("http://python.org"), loop=loop)
morsel: "Morsel[str]" = Morsel()
morsel.set(key="string", val="Another string", coded_val="really")
morsel_cookie = {"str": morsel}
@benchmark
def _run() -> None:
req.update_cookies(cookies=morsel_cookie)
def test_create_client_request_with_cookies(
loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture
) -> None:
url = URL("http://python.org")
@benchmark
def _run() -> None:
ClientRequest(
method="get",
url=url,
loop=loop,
headers=None,
data=None,
cookies={"cookie": "value"},
auth=None,
version=HttpVersion11,
compress=False,
chunked=None,
expect100=False,
)
def test_create_client_request_with_headers(
loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture
) -> None:
url = URL("http://python.org")
@benchmark
def _run() -> None:
ClientRequest(
method="get",
url=url,
loop=loop,
headers={"header": "value", "another": "header"},
data=None,
cookies=None,
auth=None,
version=HttpVersion11,
compress=False,
chunked=None,
expect100=False,
)
def test_send_client_request_one_hundred(
loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture
) -> None:
url = URL("http://python.org")
req = ClientRequest("get", url, loop=loop)
class MockTransport(asyncio.Transport):
"""Mock transport for testing that do no real I/O."""
def is_closing(self) -> bool:
"""Swallow is_closing."""
return False
def write(self, data: Union[bytes, bytearray, memoryview]) -> None:
"""Swallow writes."""
class MockProtocol(asyncio.BaseProtocol):
def __init__(self) -> None:
self.transport = MockTransport()
self._paused = False
@property
def writing_paused(self) -> bool:
return False
async def _drain_helper(self) -> None:
"""Swallow drain."""
def start_timeout(self) -> None:
"""Swallow start_timeout."""
class MockConnector:
def __init__(self) -> None:
self.force_close = False
class MockConnection:
def __init__(self) -> None:
self.transport = None
self.protocol = MockProtocol()
self._connector = MockConnector()
conn = MockConnection()
async def send_requests() -> None:
for _ in range(100):
await req.send(conn) # type: ignore[arg-type]
@benchmark
def _run() -> None:
loop.run_until_complete(send_requests())
|
