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