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