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