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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
"""codspeed benchmarks for client requests."""
import asyncio
from http.cookies import BaseCookie
from typing import Union
from multidict import CIMultiDict
from pytest_codspeed import BenchmarkFixture
from yarl import URL
from aiohttp.client_reqrep import ClientRequest, ClientResponse
from aiohttp.cookiejar import CookieJar
from aiohttp.helpers import TimerNoop
from aiohttp.http_writer import HttpVersion11
from aiohttp.tracing import Trace
def test_client_request_update_cookies(
loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture
) -> None:
url = URL("http://python.org")
async def setup():
cookie_jar = CookieJar()
cookie_jar.update_cookies({"string": "Another string"})
cookies = cookie_jar.filter_cookies(url)
assert cookies["string"].value == "Another string"
req = ClientRequest("get", url, loop=loop)
return req, cookies
req, cookies = loop.run_until_complete(setup())
@benchmark
def _run() -> None:
req.update_cookies(cookies=cookies)
def test_create_client_request_with_cookies(
loop: asyncio.AbstractEventLoop, benchmark: BenchmarkFixture
) -> None:
url = URL("http://python.org")
async def setup():
cookie_jar = CookieJar()
cookie_jar.update_cookies({"cookie": "value"})
cookies = cookie_jar.filter_cookies(url)
assert cookies["cookie"].value == "value"
return cookies
cookies = loop.run_until_complete(setup())
timer = TimerNoop()
traces: list[Trace] = []
headers = CIMultiDict[str]()
@benchmark
def _run() -> None:
ClientRequest(
method="get",
url=url,
loop=loop,
params=None,
skip_auto_headers=None,
response_class=ClientResponse,
proxy=None,
proxy_auth=None,
proxy_headers=None,
timer=timer,
session=None,
ssl=True,
traces=traces,
trust_env=False,
server_hostname=None,
headers=headers,
data=None,
cookies=cookies,
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")
timer = TimerNoop()
traces: list[Trace] = []
headers = CIMultiDict({"header": "value", "another": "header"})
cookies = BaseCookie[str]()
@benchmark
def _run() -> None:
ClientRequest(
method="get",
url=url,
loop=loop,
params=None,
skip_auto_headers=None,
response_class=ClientResponse,
proxy=None,
proxy_auth=None,
proxy_headers=None,
timer=timer,
session=None,
ssl=True,
traces=traces,
trust_env=False,
server_hostname=None,
headers=headers,
data=None,
cookies=cookies,
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())
|
