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