File: test_web.py

package info (click to toggle)
python-aiohttp 0.17.2-1~bpo8%2B1
  • links: PTS, VCS
  • area: main
  • in suites: jessie-backports
  • size: 2,368 kB
  • sloc: python: 19,899; makefile: 205
file content (162 lines) | stat: -rw-r--r-- 5,150 bytes parent folder | download
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
import asyncio
import unittest
from unittest import mock

from aiohttp import web, log
from aiohttp.multidict import CIMultiDict
from aiohttp.protocol import HttpVersion11, RawRequestMessage


class TestWeb(unittest.TestCase):

    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)

    def tearDown(self):
        self.loop.close()

    def test_handler_returns_not_response(self):
        app = web.Application(loop=self.loop)

        def handler(request):
            return 'abc'

        app.router.add_route('GET', '/', handler)
        h = app.make_handler()()
        message = RawRequestMessage('GET', '/', HttpVersion11,
                                    CIMultiDict(), False, False)
        payload = mock.Mock()

        with self.assertRaises(AssertionError):
            self.loop.run_until_complete(h.handle_request(message, payload))

    def test_app_ctor(self):
        app = web.Application(loop=self.loop)
        self.assertIs(self.loop, app.loop)
        self.assertIs(app.logger, log.web_logger)

    def test_app_call(self):
        app = web.Application(loop=self.loop)
        self.assertIs(app, app())

    def test_app_default_loop(self):
        asyncio.set_event_loop(self.loop)
        app = web.Application()
        self.assertIs(self.loop, app.loop)

    def test_app_register_on_finish(self):
        app = web.Application(loop=self.loop)
        cb1 = mock.Mock()
        cb2 = mock.Mock()
        app.register_on_finish(cb1, 1, b=2)
        app.register_on_finish(cb2, 2, c=3)
        self.loop.run_until_complete(app.finish())
        cb1.assert_called_once_with(app, 1, b=2)
        cb2.assert_called_once_with(app, 2, c=3)

    def test_app_register_coro(self):
        app = web.Application(loop=self.loop)

        fut = asyncio.Future(loop=self.loop)

        @asyncio.coroutine
        def cb(app):
            yield from asyncio.sleep(0.001, loop=self.loop)
            fut.set_result(123)

        app.register_on_finish(cb)
        self.loop.run_until_complete(app.finish())
        self.assertTrue(fut.done())
        self.assertEqual(123, fut.result())

    def test_app_error_in_finish_callbacks(self):
        app = web.Application(loop=self.loop)

        err = RuntimeError('bad call')
        app.register_on_finish(mock.Mock(side_effect=err))
        handler = mock.Mock()
        self.loop.set_exception_handler(handler)
        self.loop.run_until_complete(app.finish())
        exc_info = {'exception': err,
                    'application': app,
                    'message': 'Error in finish callback'}
        handler.assert_called_once_with(self.loop, exc_info)

    def test_non_default_router(self):
        router = web.UrlDispatcher()
        app = web.Application(loop=self.loop, router=router)
        self.assertIs(router, app.router)

    def test_logging(self):
        logger = mock.Mock()
        app = web.Application(loop=self.loop)
        app.logger = logger
        self.assertIs(app.logger, logger)


class TestRequestHandlerFactory(unittest.TestCase):

    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)

    def tearDown(self):
        self.loop.close()

    def test_repr(self):
        app = web.Application(loop=self.loop)
        manager = app.make_handler()
        handler = manager()

        self.assertEqual(
            '<RequestHandler none:none disconnected>', repr(handler))

        handler.transport = object()
        handler._meth = 'GET'
        handler._path = '/index.html'
        self.assertEqual(
            '<RequestHandler GET:/index.html connected>', repr(handler))

    def test_connections(self):
        app = web.Application(loop=self.loop)
        manager = app.make_handler()
        self.assertEqual(manager.connections, [])

        handler = object()
        transport = object()
        manager.connection_made(handler, transport)
        self.assertEqual(manager.connections, [handler])

        manager.connection_lost(handler, None)
        self.assertEqual(manager.connections, [])

    def test_finish_connection_no_timeout(self):
        app = web.Application(loop=self.loop)
        manager = app.make_handler()

        handler = mock.Mock()
        transport = mock.Mock()
        manager.connection_made(handler, transport)

        self.loop.run_until_complete(manager.finish_connections())

        manager.connection_lost(handler, None)
        self.assertEqual(manager.connections, [])
        handler.closing.assert_called_with(timeout=None)
        transport.close.assert_called_with()

    def test_finish_connection_timeout(self):
        app = web.Application(loop=self.loop)
        manager = app.make_handler()

        handler = mock.Mock()
        transport = mock.Mock()
        manager.connection_made(handler, transport)

        self.loop.run_until_complete(manager.finish_connections(timeout=0.1))

        manager.connection_lost(handler, None)
        self.assertEqual(manager.connections, [])
        handler.closing.assert_called_with(timeout=0.09)
        transport.close.assert_called_with()