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