xref: /unit/test/test_static.py (revision 2066:242192963d93)
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', temp_dir + '/assets/dir/fi le',
137        )
138        assert waitforfiles(temp_dir + '/assets/dir/fi le')
139        assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name'
140
141        os.rename(temp_dir + '/assets/dir', temp_dir + '/assets/di r')
142        assert waitforfiles(temp_dir + '/assets/di r/fi le')
143        assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name'
144
145        os.rename(temp_dir + '/assets/di r', temp_dir + '/assets/ di r ')
146        assert waitforfiles(temp_dir + '/assets/ di r /fi le')
147        assert (
148            self.get(url='/ di r /fi le')['body'] == 'blah'
149        ), 'dir name enclosing'
150
151        assert (
152            self.get(url='/%20di%20r%20/fi le')['body'] == 'blah'
153        ), 'dir encoded'
154        assert (
155            self.get(url='/ di r %2Ffi le')['body'] == 'blah'
156        ), 'slash encoded'
157        assert (
158            self.get(url='/ di r /fi%20le')['body'] == 'blah'
159        ), 'file encoded'
160        assert (
161            self.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah'
162        ), 'encoded'
163        assert (
164            self.get(url='/%20%64%69%20%72%20%2F%66%69%20%6C%65')['body']
165            == 'blah'
166        ), 'encoded 2'
167
168        os.rename(
169            temp_dir + '/assets/ di r /fi le',
170            temp_dir + '/assets/ di r / fi le ',
171        )
172        assert waitforfiles(temp_dir + '/assets/ di r / fi le ')
173        assert (
174            self.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah'
175        ), 'file name enclosing'
176
177        try:
178            open(temp_dir + '/ф а', 'a').close()
179            utf8 = True
180
181        except KeyboardInterrupt:
182            raise
183
184        except:
185            utf8 = False
186
187        if utf8:
188            os.rename(
189                temp_dir + '/assets/ di r / fi le ',
190                temp_dir + '/assets/ di r /фа йл',
191            )
192            assert waitforfiles(temp_dir + '/assets/ di r /фа йл')
193            assert (
194                self.get(url='/ di r /фа йл')['body'] == 'blah'
195            ), 'file name 2'
196
197            os.rename(
198                temp_dir + '/assets/ di r ', 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']
270            == 'text/plain'
271        ), 'mime_types file in dir'
272
273    def test_static_mime_types_partial_match(self):
274        assert 'success' in self.conf(
275            {"text/x-blah": ["ile", "fil", "f", "e", ".file"],},
276            'settings/http/static/mime_types',
277        ), 'configure mime_types'
278        assert 'Content-Type' not in self.get(url='/dir/file'), 'partial match'
279
280    def test_static_mime_types_reconfigure(self):
281        assert 'success' in self.conf(
282            {
283                "text/x-code": "readme",
284                "text/plain": [".html", ".log", "file"],
285            },
286            'settings/http/static/mime_types',
287        ), 'configure mime_types'
288
289        assert self.conf_get('settings/http/static/mime_types') == {
290            'text/x-code': 'readme',
291            'text/plain': ['.html', '.log', 'file'],
292        }, 'mime_types get'
293        assert (
294            self.conf_get('settings/http/static/mime_types/text%2Fx-code')
295            == 'readme'
296        ), 'mime_types get string'
297        assert self.conf_get(
298            'settings/http/static/mime_types/text%2Fplain'
299        ) == ['.html', '.log', 'file'], 'mime_types get array'
300        assert (
301            self.conf_get('settings/http/static/mime_types/text%2Fplain/1')
302            == '.log'
303        ), 'mime_types get array element'
304
305        assert 'success' in self.conf_delete(
306            'settings/http/static/mime_types/text%2Fplain/2'
307        ), 'mime_types remove array element'
308        assert (
309            'Content-Type' not in self.get(url='/dir/file')['headers']
310        ), 'mime_types removed'
311
312        assert 'success' in self.conf_post(
313            '"file"', 'settings/http/static/mime_types/text%2Fplain'
314        ), 'mime_types add array element'
315        assert (
316            self.get(url='/dir/file')['headers']['Content-Type']
317            == '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']
325            == 'text/plain'
326        ), 'mime_types updated'
327        assert (
328            'Content-Type' not in self.get(url='/log.log')['headers']
329        ), 'mime_types updated 2'
330
331        assert 'success' in self.conf(
332            '".log"', 'settings/http/static/mime_types/text%2Fblahblahblah'
333        ), 'configure mime_types create'
334        assert (
335            self.get(url='/log.log')['headers']['Content-Type']
336            == 'text/blahblahblah'
337        ), 'mime_types create'
338
339    def test_static_mime_types_correct(self):
340        assert 'error' in self.conf(
341            {"text/x-code": "readme", "text/plain": "readme"},
342            'settings/http/static/mime_types',
343        ), 'mime_types same extensions'
344        assert 'error' in self.conf(
345            {"text/x-code": [".h", ".c"], "text/plain": ".c"},
346            'settings/http/static/mime_types',
347        ), 'mime_types same extensions array'
348        assert 'error' in self.conf(
349            {"text/x-code": [".h", ".c", "readme"], "text/plain": "README",},
350            'settings/http/static/mime_types',
351        ), 'mime_types same extensions case insensitive'
352
353    @pytest.mark.skip('not yet')
354    def test_static_mime_types_invalid(self, temp_dir):
355        assert 'error' in self.http(
356            b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r
357Host: localhost\r
358Connection: close\r
359Content-Length: 6\r
360\r
361\"blah\"""",
362            raw_resp=True,
363            raw=True,
364            sock_type='unix',
365            addr=temp_dir + '/control.unit.sock',
366        ), 'mime_types invalid'
367