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