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
|
import asyncio
import aiohttp
import pytest
import pytest_asyncio
from aiohttp import web
@pytest_asyncio.fixture()
async def loop():
"""Replace aiohttp loop fixture."""
return asyncio.get_running_loop()
def make_app():
app = web.Application()
async def constant_handler(request):
return web.Response(text="42")
async def ip_handler(request):
protocol = request.query["protocol"]
ip = await get_ip_address(protocol=protocol)
return web.Response(text=f"ip is {ip}")
app.add_routes([web.get("/constant", constant_handler)])
app.add_routes([web.get("/ip", ip_handler)])
return app
async def get_ip_address(protocol):
async with aiohttp.ClientSession() as s:
async with s.get(f"{protocol}://httpbin.org/ip") as resp:
ip = (await resp.json())["origin"]
return ip
@pytest.mark.asyncio
async def test_app_simple_endpoint(aiohttp_client):
client = await aiohttp_client(make_app())
r = await client.get("/constant")
assert (await r.text()) == "42"
@pytest.mark.asyncio
async def test_app_simple_endpoint_with_aresponses(aiohttp_client, aresponses):
"""
when testing your own aiohttp server you must setup passthrough to it
Ideally this wouldn't be necessary but haven't figured that out yet.
Perhaps all local calls should be passthrough.
"""
aresponses.add("127.0.0.1:4241", response=aresponses.passthrough)
client = await aiohttp_client(make_app(), server_kwargs={"port": 4241})
r = await client.get("/constant")
assert (await r.text()) == "42"
@pytest.mark.asyncio
@pytest.mark.parametrize("protocol", ["http", "https"])
async def test_app_with_subrequest_using_aresponses(
aiohttp_client, aresponses, protocol
):
"""
but passthrough doesn't work if the handler itself makes an aiohttp https request
"""
aresponses.add_local_passthrough(repeat=1)
aresponses.add("httpbin.org", response={"origin": "1.2.3.4"})
client = await aiohttp_client(make_app())
r = await client.get(f"/ip?protocol={protocol}")
body = await r.text()
assert r.status == 200, body
assert "ip is" in (await r.text())
aresponses.assert_plan_strictly_followed()
|
