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