-
-
Notifications
You must be signed in to change notification settings - Fork 2.1k
/
Copy pathtest_web_sendfile.py
177 lines (140 loc) · 5.71 KB
/
test_web_sendfile.py
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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
from unittest import mock
from aiohttp import hdrs
from aiohttp.test_utils import make_mocked_coro, make_mocked_request
from aiohttp.web_fileresponse import FileResponse, SendfileStreamWriter
def test_static_handle_eof(loop):
fake_loop = mock.Mock()
with mock.patch('aiohttp.web_fileresponse.os') as m_os:
out_fd = 30
in_fd = 31
fut = loop.create_future()
m_os.sendfile.return_value = 0
writer = SendfileStreamWriter(mock.Mock(), mock.Mock(), fake_loop)
writer._sendfile_cb(fut, out_fd, in_fd, 0, 100, fake_loop, False)
m_os.sendfile.assert_called_with(out_fd, in_fd, 0, 100)
assert fut.done()
assert fut.result() is None
assert not fake_loop.add_writer.called
assert not fake_loop.remove_writer.called
def test_static_handle_again(loop):
fake_loop = mock.Mock()
with mock.patch('aiohttp.web_fileresponse.os') as m_os:
out_fd = 30
in_fd = 31
fut = loop.create_future()
m_os.sendfile.side_effect = BlockingIOError()
writer = SendfileStreamWriter(mock.Mock(), mock.Mock(), fake_loop)
writer._sendfile_cb(fut, out_fd, in_fd, 0, 100, fake_loop, False)
m_os.sendfile.assert_called_with(out_fd, in_fd, 0, 100)
assert not fut.done()
fake_loop.add_writer.assert_called_with(out_fd,
writer._sendfile_cb,
fut, out_fd, in_fd, 0, 100,
fake_loop, True)
assert not fake_loop.remove_writer.called
def test_static_handle_exception(loop):
fake_loop = mock.Mock()
with mock.patch('aiohttp.web_fileresponse.os') as m_os:
out_fd = 30
in_fd = 31
fut = loop.create_future()
exc = OSError()
m_os.sendfile.side_effect = exc
writer = SendfileStreamWriter(mock.Mock(), mock.Mock(), fake_loop)
writer._sendfile_cb(fut, out_fd, in_fd, 0, 100, fake_loop, False)
m_os.sendfile.assert_called_with(out_fd, in_fd, 0, 100)
assert fut.done()
assert exc is fut.exception()
assert not fake_loop.add_writer.called
assert not fake_loop.remove_writer.called
def test__sendfile_cb_return_on_cancelling(loop):
fake_loop = mock.Mock()
with mock.patch('aiohttp.web_fileresponse.os') as m_os:
out_fd = 30
in_fd = 31
fut = loop.create_future()
fut.cancel()
writer = SendfileStreamWriter(mock.Mock(), mock.Mock(), fake_loop)
writer._sendfile_cb(fut, out_fd, in_fd, 0, 100, fake_loop, False)
assert fut.done()
assert not fake_loop.add_writer.called
assert not fake_loop.remove_writer.called
assert not m_os.sendfile.called
def test_using_gzip_if_header_present_and_file_available(loop):
request = make_mocked_request(
'GET', 'http://python.org/logo.png', headers={
hdrs.ACCEPT_ENCODING: 'gzip'
}
)
gz_filepath = mock.Mock()
gz_filepath.open = mock.mock_open()
gz_filepath.is_file.return_value = True
gz_filepath.stat.return_value = mock.MagicMock()
gz_filepath.stat.st_size = 1024
filepath = mock.Mock()
filepath.name = 'logo.png'
filepath.open = mock.mock_open()
filepath.with_name.return_value = gz_filepath
file_sender = FileResponse(filepath)
file_sender._sendfile = make_mocked_coro(None)
loop.run_until_complete(file_sender.prepare(request))
assert not filepath.open.called
assert gz_filepath.open.called
def test_gzip_if_header_not_present_and_file_available(loop):
request = make_mocked_request(
'GET', 'http://python.org/logo.png', headers={
}
)
gz_filepath = mock.Mock()
gz_filepath.open = mock.mock_open()
gz_filepath.is_file.return_value = True
filepath = mock.Mock()
filepath.name = 'logo.png'
filepath.open = mock.mock_open()
filepath.with_name.return_value = gz_filepath
filepath.stat.return_value = mock.MagicMock()
filepath.stat.st_size = 1024
file_sender = FileResponse(filepath)
file_sender._sendfile = make_mocked_coro(None)
loop.run_until_complete(file_sender.prepare(request))
assert filepath.open.called
assert not gz_filepath.open.called
def test_gzip_if_header_not_present_and_file_not_available(loop):
request = make_mocked_request(
'GET', 'http://python.org/logo.png', headers={
}
)
gz_filepath = mock.Mock()
gz_filepath.open = mock.mock_open()
gz_filepath.is_file.return_value = False
filepath = mock.Mock()
filepath.name = 'logo.png'
filepath.open = mock.mock_open()
filepath.with_name.return_value = gz_filepath
filepath.stat.return_value = mock.MagicMock()
filepath.stat.st_size = 1024
file_sender = FileResponse(filepath)
file_sender._sendfile = make_mocked_coro(None)
loop.run_until_complete(file_sender.prepare(request))
assert filepath.open.called
assert not gz_filepath.open.called
def test_gzip_if_header_present_and_file_not_available(loop):
request = make_mocked_request(
'GET', 'http://python.org/logo.png', headers={
hdrs.ACCEPT_ENCODING: 'gzip'
}
)
gz_filepath = mock.Mock()
gz_filepath.open = mock.mock_open()
gz_filepath.is_file.return_value = False
filepath = mock.Mock()
filepath.name = 'logo.png'
filepath.open = mock.mock_open()
filepath.with_name.return_value = gz_filepath
filepath.stat.return_value = mock.MagicMock()
filepath.stat.st_size = 1024
file_sender = FileResponse(filepath)
file_sender._sendfile = make_mocked_coro(None)
loop.run_until_complete(file_sender.prepare(request))
assert filepath.open.called
assert not gz_filepath.open.called