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
|
from pathlib import Path
from stat import S_IFREG, S_IRUSR, S_IWUSR
from unittest import mock
from aiohttp import hdrs
from aiohttp.http_writer import StreamWriter
from aiohttp.test_utils import make_mocked_request
from aiohttp.web_fileresponse import FileResponse
MOCK_MODE = S_IFREG | S_IRUSR | S_IWUSR
def test_using_gzip_if_header_present_and_file_available(loop) -> None:
request = make_mocked_request(
"GET",
"http://python.org/logo.png",
# Header uses some uppercase to ensure case-insensitive treatment
headers={hdrs.ACCEPT_ENCODING: "GZip"},
)
gz_filepath = mock.create_autospec(Path, spec_set=True)
gz_filepath.lstat.return_value.st_size = 1024
gz_filepath.lstat.return_value.st_mtime_ns = 1603733507222449291
gz_filepath.lstat.return_value.st_mode = MOCK_MODE
filepath = mock.create_autospec(Path, spec_set=True)
filepath.name = "logo.png"
filepath.with_suffix.return_value = gz_filepath
file_sender = FileResponse(filepath)
file_sender._path = filepath
file_sender._sendfile = mock.AsyncMock(return_value=None) # type: ignore[method-assign]
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) -> None:
request = make_mocked_request("GET", "http://python.org/logo.png", headers={})
gz_filepath = mock.create_autospec(Path, spec_set=True)
gz_filepath.lstat.return_value.st_size = 1024
gz_filepath.lstat.return_value.st_mtime_ns = 1603733507222449291
gz_filepath.lstat.return_value.st_mode = MOCK_MODE
filepath = mock.create_autospec(Path, spec_set=True)
filepath.name = "logo.png"
filepath.with_suffix.return_value = gz_filepath
filepath.stat.return_value.st_size = 1024
filepath.stat.return_value.st_mtime_ns = 1603733507222449291
filepath.stat.return_value.st_mode = MOCK_MODE
file_sender = FileResponse(filepath)
file_sender._path = filepath
file_sender._sendfile = mock.AsyncMock(return_value=None) # type: ignore[method-assign]
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) -> None:
request = make_mocked_request("GET", "http://python.org/logo.png", headers={})
gz_filepath = mock.create_autospec(Path, spec_set=True)
gz_filepath.stat.side_effect = OSError(2, "No such file or directory")
filepath = mock.create_autospec(Path, spec_set=True)
filepath.name = "logo.png"
filepath.with_suffix.return_value = gz_filepath
filepath.stat.return_value.st_size = 1024
filepath.stat.return_value.st_mtime_ns = 1603733507222449291
filepath.stat.return_value.st_mode = MOCK_MODE
file_sender = FileResponse(filepath)
file_sender._path = filepath
file_sender._sendfile = mock.AsyncMock(return_value=None) # type: ignore[method-assign]
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) -> None:
request = make_mocked_request(
"GET", "http://python.org/logo.png", headers={hdrs.ACCEPT_ENCODING: "gzip"}
)
gz_filepath = mock.create_autospec(Path, spec_set=True)
gz_filepath.lstat.side_effect = OSError(2, "No such file or directory")
filepath = mock.create_autospec(Path, spec_set=True)
filepath.name = "logo.png"
filepath.with_suffix.return_value = gz_filepath
filepath.stat.return_value.st_size = 1024
filepath.stat.return_value.st_mtime_ns = 1603733507222449291
filepath.stat.return_value.st_mode = MOCK_MODE
file_sender = FileResponse(filepath)
file_sender._path = filepath
file_sender._sendfile = mock.AsyncMock(return_value=None) # type: ignore[method-assign]
loop.run_until_complete(file_sender.prepare(request))
assert filepath.open.called
assert not gz_filepath.open.called
def test_status_controlled_by_user(loop) -> None:
request = make_mocked_request("GET", "http://python.org/logo.png", headers={})
filepath = mock.create_autospec(Path, spec_set=True)
filepath.name = "logo.png"
filepath.stat.return_value.st_size = 1024
filepath.stat.return_value.st_mtime_ns = 1603733507222449291
filepath.stat.return_value.st_mode = MOCK_MODE
file_sender = FileResponse(filepath, status=203)
file_sender._path = filepath
file_sender._sendfile = mock.AsyncMock(return_value=None) # type: ignore[method-assign]
loop.run_until_complete(file_sender.prepare(request))
assert file_sender._status == 203
async def test_file_response_sends_headers_immediately() -> None:
"""Test that FileResponse sends headers immediately (inherits from StreamResponse)."""
writer = mock.create_autospec(StreamWriter, spec_set=True)
writer.write_headers = mock.AsyncMock()
writer.send_headers = mock.Mock()
writer.write_eof = mock.AsyncMock()
request = make_mocked_request("GET", "http://python.org/logo.png", writer=writer)
filepath = mock.create_autospec(Path, spec_set=True)
filepath.name = "logo.png"
filepath.stat.return_value.st_size = 1024
filepath.stat.return_value.st_mtime_ns = 1603733507222449291
filepath.stat.return_value.st_mode = MOCK_MODE
file_sender = FileResponse(filepath)
file_sender._path = filepath
file_sender._sendfile = mock.AsyncMock(return_value=None) # type: ignore[method-assign]
# FileResponse inherits from StreamResponse, so should send immediately
assert file_sender._send_headers_immediately is True
# Prepare the response
await file_sender.prepare(request)
# Headers should be sent immediately
writer.send_headers.assert_called_once()
|
