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()
|