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
|
import sys
import unittest
import six
if six.PY3:
from unittest import mock
else:
import mock
if sys.version_info >= (3, 5):
from engineio import async_aiohttp
@unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+')
class AiohttpTests(unittest.TestCase):
def test_create_route(self):
app = mock.MagicMock()
mock_server = mock.MagicMock()
async_aiohttp.create_route(app, mock_server, '/foo')
app.router.add_get.assert_any_call('/foo', mock_server.handle_request)
app.router.add_post.assert_any_call('/foo', mock_server.handle_request)
def test_translate_request(self):
request = mock.MagicMock()
request._message.method = 'PUT'
request._message.path = '/foo/bar?baz=1'
request._message.version = (1, 1)
request._message.headers = {'a': 'b', 'c-c': 'd', 'c_c': 'e',
'content-type': 'application/json',
'content-length': 123}
request._payload = b'hello world'
environ = async_aiohttp.translate_request(request)
expected_environ = {
'REQUEST_METHOD': 'PUT',
'PATH_INFO': '/foo/bar',
'QUERY_STRING': 'baz=1',
'CONTENT_TYPE': 'application/json',
'CONTENT_LENGTH': 123,
'HTTP_A': 'b',
# 'HTTP_C_C': 'd,e',
'RAW_URI': '/foo/bar?baz=1',
'SERVER_PROTOCOL': 'HTTP/1.1',
'wsgi.input': b'hello world',
'aiohttp.request': request,
}
for k, v in expected_environ.items():
self.assertEqual(v, environ[k])
self.assertTrue(
environ['HTTP_C_C'] == 'd,e' or environ['HTTP_C_C'] == 'e,d')
# @mock.patch('async_aiohttp.aiohttp.web.Response')
def test_make_response(self):
rv = async_aiohttp.make_response('202 ACCEPTED', {'foo': 'bar'},
b'payload', {})
self.assertEqual(rv.status, 202)
self.assertEqual(rv.headers['foo'], 'bar')
self.assertEqual(rv.body, b'payload')
|
