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
|
import asyncio
from collections.abc import Coroutine
import pytest
import aiohttp
from aiohttp import web
from aiohttp.client import _RequestContextManager
async def test_await(test_server, loop):
async def handler(request):
return web.HTTPOk()
app = web.Application(loop=loop)
app.router.add_route('GET', '/', handler)
server = await test_server(app)
resp = await aiohttp.get(server.make_url('/'), loop=loop)
assert resp.status == 200
assert resp.connection is not None
await resp.release()
assert resp.connection is None
async def test_response_context_manager(test_server, loop):
async def handler(request):
return web.HTTPOk()
app = web.Application(loop=loop)
app.router.add_route('GET', '/', handler)
server = await test_server(app)
resp = await aiohttp.get(server.make_url('/'), loop=loop)
async with resp:
assert resp.status == 200
assert resp.connection is not None
assert resp.connection is None
async def test_response_context_manager_error(test_server, loop):
async def handler(request):
return web.HTTPOk()
app = web.Application(loop=loop)
app.router.add_route('GET', '/', handler)
server = await test_server(app)
cm = aiohttp.get(server.make_url('/'), loop=loop)
session = cm._session
resp = await cm
with pytest.raises(RuntimeError):
async with resp:
assert resp.status == 200
resp.content.set_exception(RuntimeError())
await resp.read()
assert len(session._connector._conns) == 0
async def test_client_api_context_manager(test_server, loop):
async def handler(request):
return web.HTTPOk()
app = web.Application(loop=loop)
app.router.add_route('GET', '/', handler)
server = await test_server(app)
async with aiohttp.get(server.make_url('/'), loop=loop) as resp:
assert resp.status == 200
assert resp.connection is not None
assert resp.connection is None
def test_ctx_manager_is_coroutine():
assert issubclass(_RequestContextManager, Coroutine)
async def test_context_manager_timeout_on_release(test_server, loop):
async def handler(request):
resp = web.StreamResponse()
await resp.prepare(request)
await asyncio.sleep(10, loop=loop)
return resp
app = web.Application(loop=loop)
app.router.add_route('GET', '/', handler)
server = await test_server(app)
with aiohttp.ClientSession(loop=loop) as session:
resp = await session.get(server.make_url('/'))
with pytest.raises(asyncio.TimeoutError):
with aiohttp.Timeout(0.01, loop=loop):
async with resp:
assert resp.status == 200
assert resp.connection is not None
assert resp.connection is None
async def test_iter_any(test_server, loop):
data = b'0123456789' * 1024
async def handler(request):
buf = []
async for raw in request.content.iter_any():
buf.append(raw)
assert b''.join(buf) == data
return web.Response()
app = web.Application(loop=loop)
app.router.add_route('POST', '/', handler)
server = await test_server(app)
with aiohttp.ClientSession(loop=loop) as session:
async with await session.post(server.make_url('/'), data=data) as resp:
assert resp.status == 200
|
