test_static.py (1635:97afbb6c5a15) test_static.py (1654:fc7d0578e124)
1import os
2import socket
3
4import pytest
5
1import os
2import socket
3
4import pytest
5
6from conftest import option
6from conftest import waitforfiles
7from unit.applications.proto import TestApplicationProto
8
9
10class TestStatic(TestApplicationProto):
11 prerequisites = {}
12
13 def setup_method(self):
7from conftest import waitforfiles
8from unit.applications.proto import TestApplicationProto
9
10
11class TestStatic(TestApplicationProto):
12 prerequisites = {}
13
14 def setup_method(self):
14 super().setup_method()
15
16 os.makedirs(self.temp_dir + '/assets/dir')
17 with open(self.temp_dir + '/assets/index.html', 'w') as index, open(
18 self.temp_dir + '/assets/README', 'w'
19 ) as readme, open(self.temp_dir + '/assets/log.log', 'w') as log, open(
20 self.temp_dir + '/assets/dir/file', 'w'
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'
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"}},
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"}},
30 "routes": [{"action": {"share": self.temp_dir + "/assets"}}],
31 "routes": [{"action": {"share": option.temp_dir + "/assets"}}],
31 "settings": {
32 "http": {
33 "static": {
34 "mime_types": {"text/plain": [".log", "README"]}
35 }
36 }
37 },
38 }

--- 10 unchanged lines hidden (view full) ---

49 assert self.get(url='/dir/')['status'] == 404, 'index not found'
50
51 resp = self.get(url='/index.html/')
52 assert resp['status'] == 404, 'index not found 2 status'
53 assert (
54 resp['headers']['Content-Type'] == 'text/html'
55 ), 'index not found 2 Content-Type'
56
32 "settings": {
33 "http": {
34 "static": {
35 "mime_types": {"text/plain": [".log", "README"]}
36 }
37 }
38 },
39 }

--- 10 unchanged lines hidden (view full) ---

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
57 def test_static_large_file(self):
58 def test_static_large_file(self, temp_dir):
58 file_size = 32 * 1024 * 1024
59 file_size = 32 * 1024 * 1024
59 with open(self.temp_dir + '/assets/large', 'wb') as f:
60 with open(temp_dir + '/assets/large', 'wb') as f:
60 f.seek(file_size - 1)
61 f.write(b'\0')
62
63 assert (
64 len(self.get(url='/large', read_buffer_size=1024 * 1024)['body'])
65 == file_size
66 ), 'large file'
67
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
68 def test_static_etag(self):
69 def test_static_etag(self, temp_dir):
69 etag = self.get(url='/')['headers']['ETag']
70 etag_2 = self.get(url='/README')['headers']['ETag']
71
72 assert etag != etag_2, 'different ETag'
73 assert etag == self.get(url='/')['headers']['ETag'], 'same ETag'
74
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
75 with open(self.temp_dir + '/assets/index.html', 'w') as f:
76 with open(temp_dir + '/assets/index.html', 'w') as f:
76 f.write('blah')
77
78 assert etag != self.get(url='/')['headers']['ETag'], 'new ETag'
79
80 def test_static_redirect(self):
81 resp = self.get(url='/dir')
82 assert resp['status'] == 301, 'redirect status'
83 assert resp['headers']['Location'] == '/dir/', 'redirect Location'
84 assert 'Content-Type' not in resp['headers'], 'redirect Content-Type'
85
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
86 def test_static_space_in_name(self):
87 def test_static_space_in_name(self, temp_dir):
87 os.rename(
88 os.rename(
88 self.temp_dir + '/assets/dir/file',
89 self.temp_dir + '/assets/dir/fi le',
89 temp_dir + '/assets/dir/file',
90 temp_dir + '/assets/dir/fi le',
90 )
91 )
91 assert waitforfiles(self.temp_dir + '/assets/dir/fi le')
92 assert waitforfiles(temp_dir + '/assets/dir/fi le')
92 assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name'
93
93 assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name'
94
94 os.rename(self.temp_dir + '/assets/dir', self.temp_dir + '/assets/di r')
95 assert waitforfiles(self.temp_dir + '/assets/di r/fi le')
95 os.rename(temp_dir + '/assets/dir', temp_dir + '/assets/di r')
96 assert waitforfiles(temp_dir + '/assets/di r/fi le')
96 assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name'
97
98 os.rename(
97 assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name'
98
99 os.rename(
99 self.temp_dir + '/assets/di r', self.temp_dir + '/assets/ di r '
100 temp_dir + '/assets/di r', temp_dir + '/assets/ di r '
100 )
101 )
101 assert waitforfiles(self.temp_dir + '/assets/ di r /fi le')
102 assert waitforfiles(temp_dir + '/assets/ di r /fi le')
102 assert (
103 self.get(url='/ di r /fi le')['body'] == 'blah'
104 ), 'dir name enclosing'
105
106 assert (
107 self.get(url='/%20di%20r%20/fi le')['body'] == 'blah'
108 ), 'dir encoded'
109 assert (

--- 6 unchanged lines hidden (view full) ---

116 self.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah'
117 ), 'encoded'
118 assert (
119 self.get(url='/%20%64%69%20%72%20%2F%66%69%20%6C%65')['body']
120 == 'blah'
121 ), 'encoded 2'
122
123 os.rename(
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 (

--- 6 unchanged lines hidden (view full) ---

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(
124 self.temp_dir + '/assets/ di r /fi le',
125 self.temp_dir + '/assets/ di r / fi le ',
125 temp_dir + '/assets/ di r /fi le',
126 temp_dir + '/assets/ di r / fi le ',
126 )
127 )
127 assert waitforfiles(self.temp_dir + '/assets/ di r / fi le ')
128 assert waitforfiles(temp_dir + '/assets/ di r / fi le ')
128 assert (
129 self.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah'
130 ), 'file name enclosing'
131
132 try:
129 assert (
130 self.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah'
131 ), 'file name enclosing'
132
133 try:
133 open(self.temp_dir + '/ф а', 'a').close()
134 open(temp_dir + '/ф а', 'a').close()
134 utf8 = True
135
136 except:
137 utf8 = False
138
139 if utf8:
140 os.rename(
135 utf8 = True
136
137 except:
138 utf8 = False
139
140 if utf8:
141 os.rename(
141 self.temp_dir + '/assets/ di r / fi le ',
142 self.temp_dir + '/assets/ di r /фа йл',
142 temp_dir + '/assets/ di r / fi le ',
143 temp_dir + '/assets/ di r /фа йл',
143 )
144 )
144 assert waitforfiles(self.temp_dir + '/assets/ di r /фа йл')
145 assert waitforfiles(temp_dir + '/assets/ di r /фа йл')
145 assert (
146 self.get(url='/ di r /фа йл')['body'] == 'blah'
147 ), 'file name 2'
148
149 os.rename(
146 assert (
147 self.get(url='/ di r /фа йл')['body'] == 'blah'
148 ), 'file name 2'
149
150 os.rename(
150 self.temp_dir + '/assets/ di r ',
151 self.temp_dir + '/assets/ди ректория',
151 temp_dir + '/assets/ di r ',
152 temp_dir + '/assets/ди ректория',
152 )
153 )
153 assert waitforfiles(self.temp_dir + '/assets/ди ректория/фа йл')
154 assert waitforfiles(temp_dir + '/assets/ди ректория/фа йл')
154 assert (
155 self.get(url='/ди ректория/фа йл')['body'] == 'blah'
156 ), 'dir name 2'
157
155 assert (
156 self.get(url='/ди ректория/фа йл')['body'] == 'blah'
157 ), 'dir name 2'
158
158 def test_static_unix_socket(self):
159 def test_static_unix_socket(self, temp_dir):
159 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
160 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
160 sock.bind(self.temp_dir + '/assets/unix_socket')
161 sock.bind(temp_dir + '/assets/unix_socket')
161
162 assert self.get(url='/unix_socket')['status'] == 404, 'socket'
163
164 sock.close()
165
162
163 assert self.get(url='/unix_socket')['status'] == 404, 'socket'
164
165 sock.close()
166
166 def test_static_unix_fifo(self):
167 os.mkfifo(self.temp_dir + '/assets/fifo')
167 def test_static_unix_fifo(self, temp_dir):
168 os.mkfifo(temp_dir + '/assets/fifo')
168
169 assert self.get(url='/fifo')['status'] == 404, 'fifo'
170
169
170 assert self.get(url='/fifo')['status'] == 404, 'fifo'
171
171 def test_static_symlink(self):
172 os.symlink(self.temp_dir + '/assets/dir', self.temp_dir + '/assets/link')
172 def test_static_symlink(self, temp_dir):
173 os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link')
173
174 assert self.get(url='/dir')['status'] == 301, 'dir'
175 assert self.get(url='/dir/file')['status'] == 200, 'file'
176 assert self.get(url='/link')['status'] == 301, 'symlink dir'
177 assert self.get(url='/link/file')['status'] == 200, 'symlink file'
178
179 def test_static_method(self):
180 resp = self.head()

--- 126 unchanged lines hidden (view full) ---

307 'settings/http/static/mime_types',
308 ), 'mime_types same extensions array'
309 assert 'error' in self.conf(
310 {"text/x-code": [".h", ".c", "readme"], "text/plain": "README",},
311 'settings/http/static/mime_types',
312 ), 'mime_types same extensions case insensitive'
313
314 @pytest.mark.skip('not yet')
174
175 assert self.get(url='/dir')['status'] == 301, 'dir'
176 assert self.get(url='/dir/file')['status'] == 200, 'file'
177 assert self.get(url='/link')['status'] == 301, 'symlink dir'
178 assert self.get(url='/link/file')['status'] == 200, 'symlink file'
179
180 def test_static_method(self):
181 resp = self.head()

--- 126 unchanged lines hidden (view full) ---

308 'settings/http/static/mime_types',
309 ), 'mime_types same extensions array'
310 assert 'error' in self.conf(
311 {"text/x-code": [".h", ".c", "readme"], "text/plain": "README",},
312 'settings/http/static/mime_types',
313 ), 'mime_types same extensions case insensitive'
314
315 @pytest.mark.skip('not yet')
315 def test_static_mime_types_invalid(self):
316 def test_static_mime_types_invalid(self, temp_dir):
316 assert 'error' in self.http(
317 b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r
318Host: localhost\r
319Connection: close\r
320Content-Length: 6\r
321\r
322\"blah\"""",
323 raw_resp=True,
324 raw=True,
325 sock_type='unix',
317 assert 'error' in self.http(
318 b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r
319Host: localhost\r
320Connection: close\r
321Content-Length: 6\r
322\r
323\"blah\"""",
324 raw_resp=True,
325 raw=True,
326 sock_type='unix',
326 addr=self.temp_dir + '/control.unit.sock',
327 addr=temp_dir + '/control.unit.sock',
327 ), 'mime_types invalid'
328 ), 'mime_types invalid'