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