aiohttp/tests/test_web_middleware.py at master · dimonb/aiohttp · GitHub

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
import asyncio
import pytest
from aiohttp import web


@pytest.mark.run_loop
def test_middleware_modifies_response(create_app_and_client):

    @asyncio.coroutine
    def handler(request):
        return web.Response(body=b'OK')

    @asyncio.coroutine
    def middleware_factory(app, handler):

        @asyncio.coroutine
        def middleware(request):
            resp = yield from handler(request)
            assert 200 == resp.status
            resp.set_status(201)
            resp.text = resp.text + '[MIDDLEWARE]'
            return resp
        return middleware

    app, client = yield from create_app_and_client()
    app.middlewares.append(middleware_factory)
    app.router.add_route('GET', '/', handler)
    resp = yield from client.get('/')
    assert 201 == resp.status
    txt = yield from resp.text()
    assert 'OK[MIDDLEWARE]' == txt


@pytest.mark.run_loop
def test_middleware_handles_exception(create_app_and_client):

    @asyncio.coroutine
    def handler(request):
        raise RuntimeError('Error text')

    @asyncio.coroutine
    def middleware_factory(app, handler):

        @asyncio.coroutine
        def middleware(request):
            with pytest.raises(RuntimeError) as ctx:
                yield from handler(request)
            return web.Response(status=501,
                                text=str(ctx.value) + '[MIDDLEWARE]')

        return middleware

    app, client = yield from create_app_and_client()
    app.middlewares.append(middleware_factory)
    app.router.add_route('GET', '/', handler)
    resp = yield from client.get('/')
    assert 501 == resp.status
    txt = yield from resp.text()
    assert 'Error text[MIDDLEWARE]' == txt


@pytest.mark.run_loop
def test_middleware_chain(create_app_and_client):

    @asyncio.coroutine
    def handler(request):
        return web.Response(text='OK')

    def make_factory(num):

        @asyncio.coroutine
        def factory(app, handler):

            def middleware(request):
                resp = yield from handler(request)
                resp.text = resp.text + '[{}]'.format(num)
                return resp

            return middleware
        return factory

    app, client = yield from create_app_and_client()
    app.middlewares.append(make_factory(1))
    app.middlewares.append(make_factory(2))
    app.router.add_route('GET', '/', handler)
    resp = yield from client.get('/')
    assert 200 == resp.status
    txt = yield from resp.text()
    assert 'OK[2][1]' == txt