xref: /unit/test/test_static.py (revision 2109:8a068eaba1e7)
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        def set_index(index):
91            assert 'success' in self.conf(
92                {"share": option.temp_dir + "/assets$uri", "index": index},
93                'routes/0/action',
94            ), 'configure index'
95
96        set_index('README')
97        assert self.get()['body'] == 'readme', 'index'
98
99        self.conf_delete('routes/0/action/index')
100        assert self.get()['body'] == '0123456789', 'delete index'
101
102        set_index('')
103        assert self.get()['status'] == 404, 'index empty'
104
105    def test_static_index_default(self):
106        assert self.get(url='/index.html')['body'] == '0123456789', 'index'
107        assert self.get(url='/')['body'] == '0123456789', 'index 2'
108        assert self.get(url='//')['body'] == '0123456789', 'index 3'
109        assert self.get(url='/.')['body'] == '0123456789', 'index 4'
110        assert self.get(url='/./')['body'] == '0123456789', 'index 5'
111        assert self.get(url='/?blah')['body'] == '0123456789', 'index vars'
112        assert self.get(url='/#blah')['body'] == '0123456789', 'index anchor'
113        assert self.get(url='/dir/')['status'] == 404, 'index not found'
114
115        resp = self.get(url='/index.html/')
116        assert resp['status'] == 404, 'index not found 2 status'
117        assert (
118            resp['headers']['Content-Type'] == 'text/html'
119        ), 'index not found 2 Content-Type'
120
121    def test_static_index_invalid(self, skip_alert):
122        skip_alert(r'failed to apply new conf')
123
124        def check_index(index):
125            assert 'error' in self.conf(
126                {"share": option.temp_dir + "/assets$uri", "index": index},
127                'routes/0/action',
128            )
129
130        check_index({})
131        check_index(['index.html', '$blah'])
132
133    def test_static_large_file(self, temp_dir):
134        file_size = 32 * 1024 * 1024
135        with open(temp_dir + '/assets/large', 'wb') as f:
136            f.seek(file_size - 1)
137            f.write(b'\0')
138
139        assert (
140            len(self.get(url='/large', read_buffer_size=1024 * 1024)['body'])
141            == file_size
142        ), 'large file'
143
144    def test_static_etag(self, temp_dir):
145        etag = self.get(url='/')['headers']['ETag']
146        etag_2 = self.get(url='/README')['headers']['ETag']
147
148        assert etag != etag_2, 'different ETag'
149        assert etag == self.get(url='/')['headers']['ETag'], 'same ETag'
150
151        with open(temp_dir + '/assets/index.html', 'w') as f:
152            f.write('blah')
153
154        assert etag != self.get(url='/')['headers']['ETag'], 'new ETag'
155
156    def test_static_redirect(self):
157        resp = self.get(url='/dir')
158        assert resp['status'] == 301, 'redirect status'
159        assert resp['headers']['Location'] == '/dir/', 'redirect Location'
160        assert 'Content-Type' not in resp['headers'], 'redirect Content-Type'
161
162    def test_static_space_in_name(self, temp_dir):
163        os.rename(
164            temp_dir + '/assets/dir/file',
165            temp_dir + '/assets/dir/fi le',
166        )
167        assert waitforfiles(temp_dir + '/assets/dir/fi le')
168        assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name'
169
170        os.rename(temp_dir + '/assets/dir', temp_dir + '/assets/di r')
171        assert waitforfiles(temp_dir + '/assets/di r/fi le')
172        assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name'
173
174        os.rename(temp_dir + '/assets/di r', temp_dir + '/assets/ di r ')
175        assert waitforfiles(temp_dir + '/assets/ di r /fi le')
176        assert (
177            self.get(url='/ di r /fi le')['body'] == 'blah'
178        ), 'dir name enclosing'
179
180        assert (
181            self.get(url='/%20di%20r%20/fi le')['body'] == 'blah'
182        ), 'dir encoded'
183        assert (
184            self.get(url='/ di r %2Ffi le')['body'] == 'blah'
185        ), 'slash encoded'
186        assert self.get(url='/ di r /fi%20le')['body'] == 'blah', 'file encoded'
187        assert (
188            self.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah'
189        ), 'encoded'
190        assert (
191            self.get(url='/%20%64%69%20%72%20%2F%66%69%20%6C%65')['body']
192            == 'blah'
193        ), 'encoded 2'
194
195        os.rename(
196            temp_dir + '/assets/ di r /fi le',
197            temp_dir + '/assets/ di r / fi le ',
198        )
199        assert waitforfiles(temp_dir + '/assets/ di r / fi le ')
200        assert (
201            self.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah'
202        ), 'file name enclosing'
203
204        try:
205            open(temp_dir + '/ф а', 'a').close()
206            utf8 = True
207
208        except KeyboardInterrupt:
209            raise
210
211        except:
212            utf8 = False
213
214        if utf8:
215            os.rename(
216                temp_dir + '/assets/ di r / fi le ',
217                temp_dir + '/assets/ di r /фа йл',
218            )
219            assert waitforfiles(temp_dir + '/assets/ di r /фа йл')
220            assert (
221                self.get(url='/ di r /фа йл')['body'] == 'blah'
222            ), 'file name 2'
223
224            os.rename(
225                temp_dir + '/assets/ di r ',
226                temp_dir + '/assets/ди ректория',
227            )
228            assert waitforfiles(temp_dir + '/assets/ди ректория/фа йл')
229            assert (
230                self.get(url='/ди ректория/фа йл')['body'] == 'blah'
231            ), 'dir name 2'
232
233    def test_static_unix_socket(self, temp_dir):
234        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
235        sock.bind(temp_dir + '/assets/unix_socket')
236
237        assert self.get(url='/unix_socket')['status'] == 404, 'socket'
238
239        sock.close()
240
241    def test_static_unix_fifo(self, temp_dir):
242        os.mkfifo(temp_dir + '/assets/fifo')
243
244        assert self.get(url='/fifo')['status'] == 404, 'fifo'
245
246    def test_static_method(self):
247        resp = self.head()
248        assert resp['status'] == 200, 'HEAD status'
249        assert resp['body'] == '', 'HEAD empty body'
250
251        assert self.delete()['status'] == 405, 'DELETE'
252        assert self.post()['status'] == 405, 'POST'
253        assert self.put()['status'] == 405, 'PUT'
254
255    def test_static_path(self):
256        assert self.get(url='/dir/../dir/file')['status'] == 200, 'relative'
257
258        assert self.get(url='./')['status'] == 400, 'path invalid'
259        assert self.get(url='../')['status'] == 400, 'path invalid 2'
260        assert self.get(url='/..')['status'] == 400, 'path invalid 3'
261        assert self.get(url='../assets/')['status'] == 400, 'path invalid 4'
262        assert self.get(url='/../assets/')['status'] == 400, 'path invalid 5'
263
264    def test_static_two_clients(self):
265        _, sock = self.get(url='/', start=True, no_recv=True)
266        _, sock2 = self.get(url='/', start=True, no_recv=True)
267
268        assert sock.recv(1) == b'H', 'client 1'
269        assert sock2.recv(1) == b'H', 'client 2'
270        assert sock.recv(1) == b'T', 'client 1 again'
271        assert sock2.recv(1) == b'T', 'client 2 again'
272
273        sock.close()
274        sock2.close()
275
276    def test_static_mime_types(self):
277        assert 'success' in self.conf(
278            {
279                "text/x-code/x-blah/x-blah": "readme",
280                "text/plain": [".html", ".log", "file"],
281            },
282            'settings/http/static/mime_types',
283        ), 'configure mime_types'
284
285        assert (
286            self.get(url='/README')['headers']['Content-Type']
287            == 'text/x-code/x-blah/x-blah'
288        ), 'mime_types string case insensitive'
289        assert (
290            self.get(url='/index.html')['headers']['Content-Type']
291            == 'text/plain'
292        ), 'mime_types html'
293        assert (
294            self.get(url='/')['headers']['Content-Type'] == 'text/plain'
295        ), 'mime_types index default'
296        assert (
297            self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
298        ), 'mime_types file in dir'
299
300    def test_static_mime_types_partial_match(self):
301        assert 'success' in self.conf(
302            {
303                "text/x-blah": ["ile", "fil", "f", "e", ".file"],
304            },
305            'settings/http/static/mime_types',
306        ), 'configure mime_types'
307        assert 'Content-Type' not in self.get(url='/dir/file'), 'partial match'
308
309    def test_static_mime_types_reconfigure(self):
310        assert 'success' in self.conf(
311            {
312                "text/x-code": "readme",
313                "text/plain": [".html", ".log", "file"],
314            },
315            'settings/http/static/mime_types',
316        ), 'configure mime_types'
317
318        assert self.conf_get('settings/http/static/mime_types') == {
319            'text/x-code': 'readme',
320            'text/plain': ['.html', '.log', 'file'],
321        }, 'mime_types get'
322        assert (
323            self.conf_get('settings/http/static/mime_types/text%2Fx-code')
324            == 'readme'
325        ), 'mime_types get string'
326        assert self.conf_get(
327            'settings/http/static/mime_types/text%2Fplain'
328        ) == ['.html', '.log', 'file'], 'mime_types get array'
329        assert (
330            self.conf_get('settings/http/static/mime_types/text%2Fplain/1')
331            == '.log'
332        ), 'mime_types get array element'
333
334        assert 'success' in self.conf_delete(
335            'settings/http/static/mime_types/text%2Fplain/2'
336        ), 'mime_types remove array element'
337        assert (
338            'Content-Type' not in self.get(url='/dir/file')['headers']
339        ), 'mime_types removed'
340
341        assert 'success' in self.conf_post(
342            '"file"', 'settings/http/static/mime_types/text%2Fplain'
343        ), 'mime_types add array element'
344        assert (
345            self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
346        ), 'mime_types reverted'
347
348        assert 'success' in self.conf(
349            '"file"', 'settings/http/static/mime_types/text%2Fplain'
350        ), 'configure mime_types update'
351        assert (
352            self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
353        ), 'mime_types updated'
354        assert (
355            'Content-Type' not in self.get(url='/log.log')['headers']
356        ), 'mime_types updated 2'
357
358        assert 'success' in self.conf(
359            '".log"', 'settings/http/static/mime_types/text%2Fblahblahblah'
360        ), 'configure mime_types create'
361        assert (
362            self.get(url='/log.log')['headers']['Content-Type']
363            == 'text/blahblahblah'
364        ), 'mime_types create'
365
366    def test_static_mime_types_correct(self):
367        assert 'error' in self.conf(
368            {"text/x-code": "readme", "text/plain": "readme"},
369            'settings/http/static/mime_types',
370        ), 'mime_types same extensions'
371        assert 'error' in self.conf(
372            {"text/x-code": [".h", ".c"], "text/plain": ".c"},
373            'settings/http/static/mime_types',
374        ), 'mime_types same extensions array'
375        assert 'error' in self.conf(
376            {
377                "text/x-code": [".h", ".c", "readme"],
378                "text/plain": "README",
379            },
380            'settings/http/static/mime_types',
381        ), 'mime_types same extensions case insensitive'
382
383    @pytest.mark.skip('not yet')
384    def test_static_mime_types_invalid(self, temp_dir):
385        assert 'error' in self.http(
386            b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r
387Host: localhost\r
388Connection: close\r
389Content-Length: 6\r
390\r
391\"blah\"""",
392            raw_resp=True,
393            raw=True,
394            sock_type='unix',
395            addr=temp_dir + '/control.unit.sock',
396        ), 'mime_types invalid'
397