xref: /unit/test/test_static.py (revision 1971:3410f9d2a662)
1import os
2import socket
3
4import pytest
5from unit.applications.proto import TestApplicationProto
6from unit.option import option
7from unit.utils import waitforfiles
8
9
10class TestStatic(TestApplicationProto):
11    prerequisites = {}
12
13    def setup_method(self):
14        os.makedirs(option.temp_dir + '/assets/dir')
15        with open(option.temp_dir + '/assets/index.html', 'w') as index, open(
16            option.temp_dir + '/assets/README', 'w'
17        ) as readme, open(
18            option.temp_dir + '/assets/log.log', 'w'
19        ) as log, open(
20            option.temp_dir + '/assets/dir/file', 'w'
21        ) as file:
22            index.write('0123456789')
23            readme.write('readme')
24            log.write('[debug]')
25            file.write('blah')
26
27        self._load_conf(
28            {
29                "listeners": {"*:7080": {"pass": "routes"}},
30                "routes": [
31                    {"action": {"share": option.temp_dir + "/assets$uri"}}
32                ],
33                "settings": {
34                    "http": {
35                        "static": {
36                            "mime_types": {"text/plain": [".log", "README"]}
37                        }
38                    }
39                },
40            }
41        )
42
43    def test_static_index(self):
44        assert self.get(url='/index.html')['body'] == '0123456789', 'index'
45        assert self.get(url='/')['body'] == '0123456789', 'index 2'
46        assert self.get(url='//')['body'] == '0123456789', 'index 3'
47        assert self.get(url='/.')['body'] == '0123456789', 'index 4'
48        assert self.get(url='/./')['body'] == '0123456789', 'index 5'
49        assert self.get(url='/?blah')['body'] == '0123456789', 'index vars'
50        assert self.get(url='/#blah')['body'] == '0123456789', 'index anchor'
51        assert self.get(url='/dir/')['status'] == 404, 'index not found'
52
53        resp = self.get(url='/index.html/')
54        assert resp['status'] == 404, 'index not found 2 status'
55        assert (
56            resp['headers']['Content-Type'] == 'text/html'
57        ), 'index not found 2 Content-Type'
58
59    def test_static_large_file(self, temp_dir):
60        file_size = 32 * 1024 * 1024
61        with open(temp_dir + '/assets/large', 'wb') as f:
62            f.seek(file_size - 1)
63            f.write(b'\0')
64
65        assert (
66            len(self.get(url='/large', read_buffer_size=1024 * 1024)['body'])
67            == file_size
68        ), 'large file'
69
70    def test_static_etag(self, temp_dir):
71        etag = self.get(url='/')['headers']['ETag']
72        etag_2 = self.get(url='/README')['headers']['ETag']
73
74        assert etag != etag_2, 'different ETag'
75        assert etag == self.get(url='/')['headers']['ETag'], 'same ETag'
76
77        with open(temp_dir + '/assets/index.html', 'w') as f:
78            f.write('blah')
79
80        assert etag != self.get(url='/')['headers']['ETag'], 'new ETag'
81
82    def test_static_redirect(self):
83        resp = self.get(url='/dir')
84        assert resp['status'] == 301, 'redirect status'
85        assert resp['headers']['Location'] == '/dir/', 'redirect Location'
86        assert 'Content-Type' not in resp['headers'], 'redirect Content-Type'
87
88    def test_static_space_in_name(self, temp_dir):
89        os.rename(
90            temp_dir + '/assets/dir/file', temp_dir + '/assets/dir/fi le',
91        )
92        assert waitforfiles(temp_dir + '/assets/dir/fi le')
93        assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name'
94
95        os.rename(temp_dir + '/assets/dir', temp_dir + '/assets/di r')
96        assert waitforfiles(temp_dir + '/assets/di r/fi le')
97        assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name'
98
99        os.rename(temp_dir + '/assets/di r', temp_dir + '/assets/ di r ')
100        assert waitforfiles(temp_dir + '/assets/ di r /fi le')
101        assert (
102            self.get(url='/ di r /fi le')['body'] == 'blah'
103        ), 'dir name enclosing'
104
105        assert (
106            self.get(url='/%20di%20r%20/fi le')['body'] == 'blah'
107        ), 'dir encoded'
108        assert (
109            self.get(url='/ di r %2Ffi le')['body'] == 'blah'
110        ), 'slash encoded'
111        assert (
112            self.get(url='/ di r /fi%20le')['body'] == 'blah'
113        ), 'file encoded'
114        assert (
115            self.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah'
116        ), 'encoded'
117        assert (
118            self.get(url='/%20%64%69%20%72%20%2F%66%69%20%6C%65')['body']
119            == 'blah'
120        ), 'encoded 2'
121
122        os.rename(
123            temp_dir + '/assets/ di r /fi le',
124            temp_dir + '/assets/ di r / fi le ',
125        )
126        assert waitforfiles(temp_dir + '/assets/ di r / fi le ')
127        assert (
128            self.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah'
129        ), 'file name enclosing'
130
131        try:
132            open(temp_dir + '/ф а', 'a').close()
133            utf8 = True
134
135        except KeyboardInterrupt:
136            raise
137
138        except:
139            utf8 = False
140
141        if utf8:
142            os.rename(
143                temp_dir + '/assets/ di r / fi le ',
144                temp_dir + '/assets/ di r /фа йл',
145            )
146            assert waitforfiles(temp_dir + '/assets/ di r /фа йл')
147            assert (
148                self.get(url='/ di r /фа йл')['body'] == 'blah'
149            ), 'file name 2'
150
151            os.rename(
152                temp_dir + '/assets/ di r ', temp_dir + '/assets/ди ректория',
153            )
154            assert waitforfiles(temp_dir + '/assets/ди ректория/фа йл')
155            assert (
156                self.get(url='/ди ректория/фа йл')['body'] == 'blah'
157            ), 'dir name 2'
158
159    def test_static_unix_socket(self, temp_dir):
160        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
161        sock.bind(temp_dir + '/assets/unix_socket')
162
163        assert self.get(url='/unix_socket')['status'] == 404, 'socket'
164
165        sock.close()
166
167    def test_static_unix_fifo(self, temp_dir):
168        os.mkfifo(temp_dir + '/assets/fifo')
169
170        assert self.get(url='/fifo')['status'] == 404, 'fifo'
171
172    def test_static_method(self):
173        resp = self.head()
174        assert resp['status'] == 200, 'HEAD status'
175        assert resp['body'] == '', 'HEAD empty body'
176
177        assert self.delete()['status'] == 405, 'DELETE'
178        assert self.post()['status'] == 405, 'POST'
179        assert self.put()['status'] == 405, 'PUT'
180
181    def test_static_path(self):
182        assert self.get(url='/dir/../dir/file')['status'] == 200, 'relative'
183
184        assert self.get(url='./')['status'] == 400, 'path invalid'
185        assert self.get(url='../')['status'] == 400, 'path invalid 2'
186        assert self.get(url='/..')['status'] == 400, 'path invalid 3'
187        assert self.get(url='../assets/')['status'] == 400, 'path invalid 4'
188        assert self.get(url='/../assets/')['status'] == 400, 'path invalid 5'
189
190    def test_static_two_clients(self):
191        _, sock = self.get(url='/', start=True, no_recv=True)
192        _, sock2 = self.get(url='/', start=True, no_recv=True)
193
194        assert sock.recv(1) == b'H', 'client 1'
195        assert sock2.recv(1) == b'H', 'client 2'
196        assert sock.recv(1) == b'T', 'client 1 again'
197        assert sock2.recv(1) == b'T', 'client 2 again'
198
199        sock.close()
200        sock2.close()
201
202    def test_static_mime_types(self):
203        assert 'success' in self.conf(
204            {
205                "text/x-code/x-blah/x-blah": "readme",
206                "text/plain": [".html", ".log", "file"],
207            },
208            'settings/http/static/mime_types',
209        ), 'configure mime_types'
210
211        assert (
212            self.get(url='/README')['headers']['Content-Type']
213            == 'text/x-code/x-blah/x-blah'
214        ), 'mime_types string case insensitive'
215        assert (
216            self.get(url='/index.html')['headers']['Content-Type']
217            == 'text/plain'
218        ), 'mime_types html'
219        assert (
220            self.get(url='/')['headers']['Content-Type'] == 'text/plain'
221        ), 'mime_types index default'
222        assert (
223            self.get(url='/dir/file')['headers']['Content-Type']
224            == 'text/plain'
225        ), 'mime_types file in dir'
226
227    def test_static_mime_types_partial_match(self):
228        assert 'success' in self.conf(
229            {"text/x-blah": ["ile", "fil", "f", "e", ".file"],},
230            'settings/http/static/mime_types',
231        ), 'configure mime_types'
232        assert 'Content-Type' not in self.get(url='/dir/file'), 'partial match'
233
234    def test_static_mime_types_reconfigure(self):
235        assert 'success' in self.conf(
236            {
237                "text/x-code": "readme",
238                "text/plain": [".html", ".log", "file"],
239            },
240            'settings/http/static/mime_types',
241        ), 'configure mime_types'
242
243        assert self.conf_get('settings/http/static/mime_types') == {
244            'text/x-code': 'readme',
245            'text/plain': ['.html', '.log', 'file'],
246        }, 'mime_types get'
247        assert (
248            self.conf_get('settings/http/static/mime_types/text%2Fx-code')
249            == 'readme'
250        ), 'mime_types get string'
251        assert self.conf_get(
252            'settings/http/static/mime_types/text%2Fplain'
253        ) == ['.html', '.log', 'file'], 'mime_types get array'
254        assert (
255            self.conf_get('settings/http/static/mime_types/text%2Fplain/1')
256            == '.log'
257        ), 'mime_types get array element'
258
259        assert 'success' in self.conf_delete(
260            'settings/http/static/mime_types/text%2Fplain/2'
261        ), 'mime_types remove array element'
262        assert (
263            'Content-Type' not in self.get(url='/dir/file')['headers']
264        ), 'mime_types removed'
265
266        assert 'success' in self.conf_post(
267            '"file"', 'settings/http/static/mime_types/text%2Fplain'
268        ), 'mime_types add array element'
269        assert (
270            self.get(url='/dir/file')['headers']['Content-Type']
271            == 'text/plain'
272        ), 'mime_types reverted'
273
274        assert 'success' in self.conf(
275            '"file"', 'settings/http/static/mime_types/text%2Fplain'
276        ), 'configure mime_types update'
277        assert (
278            self.get(url='/dir/file')['headers']['Content-Type']
279            == 'text/plain'
280        ), 'mime_types updated'
281        assert (
282            'Content-Type' not in self.get(url='/log.log')['headers']
283        ), 'mime_types updated 2'
284
285        assert 'success' in self.conf(
286            '".log"', 'settings/http/static/mime_types/text%2Fblahblahblah'
287        ), 'configure mime_types create'
288        assert (
289            self.get(url='/log.log')['headers']['Content-Type']
290            == 'text/blahblahblah'
291        ), 'mime_types create'
292
293    def test_static_mime_types_correct(self):
294        assert 'error' in self.conf(
295            {"text/x-code": "readme", "text/plain": "readme"},
296            'settings/http/static/mime_types',
297        ), 'mime_types same extensions'
298        assert 'error' in self.conf(
299            {"text/x-code": [".h", ".c"], "text/plain": ".c"},
300            'settings/http/static/mime_types',
301        ), 'mime_types same extensions array'
302        assert 'error' in self.conf(
303            {"text/x-code": [".h", ".c", "readme"], "text/plain": "README",},
304            'settings/http/static/mime_types',
305        ), 'mime_types same extensions case insensitive'
306
307    @pytest.mark.skip('not yet')
308    def test_static_mime_types_invalid(self, temp_dir):
309        assert 'error' in self.http(
310            b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r
311Host: localhost\r
312Connection: close\r
313Content-Length: 6\r
314\r
315\"blah\"""",
316            raw_resp=True,
317            raw=True,
318            sock_type='unix',
319            addr=temp_dir + '/control.unit.sock',
320        ), 'mime_types invalid'
321