1from unittest import mock 2from test import support 3from test.test_httpservers import NoLogRequestHandler 4from unittest import TestCase 5from wsgiref.util import setup_testing_defaults 6from wsgiref.headers import Headers 7from wsgiref.handlers import BaseHandler, BaseCGIHandler, SimpleHandler 8from wsgiref import util 9from wsgiref.validate import validator 10from wsgiref.simple_server import WSGIServer, WSGIRequestHandler 11from wsgiref.simple_server import make_server 12from http.client import HTTPConnection 13from io import StringIO, BytesIO, BufferedReader 14from socketserver import BaseServer 15from platform import python_implementation 16 17import os 18import re 19import signal 20import sys 21import threading 22import unittest 23 24 25class MockServer(WSGIServer): 26 """Non-socket HTTP server""" 27 28 def __init__(self, server_address, RequestHandlerClass): 29 BaseServer.__init__(self, server_address, RequestHandlerClass) 30 self.server_bind() 31 32 def server_bind(self): 33 host, port = self.server_address 34 self.server_name = host 35 self.server_port = port 36 self.setup_environ() 37 38 39class MockHandler(WSGIRequestHandler): 40 """Non-socket HTTP handler""" 41 def setup(self): 42 self.connection = self.request 43 self.rfile, self.wfile = self.connection 44 45 def finish(self): 46 pass 47 48 49def hello_app(environ,start_response): 50 start_response("200 OK", [ 51 ('Content-Type','text/plain'), 52 ('Date','Mon, 05 Jun 2006 18:49:54 GMT') 53 ]) 54 return [b"Hello, world!"] 55 56 57def header_app(environ, start_response): 58 start_response("200 OK", [ 59 ('Content-Type', 'text/plain'), 60 ('Date', 'Mon, 05 Jun 2006 18:49:54 GMT') 61 ]) 62 return [';'.join([ 63 environ['HTTP_X_TEST_HEADER'], environ['QUERY_STRING'], 64 environ['PATH_INFO'] 65 ]).encode('iso-8859-1')] 66 67 68def run_amock(app=hello_app, data=b"GET / HTTP/1.0\n\n"): 69 server = make_server("", 80, app, MockServer, MockHandler) 70 inp = BufferedReader(BytesIO(data)) 71 out = BytesIO() 72 olderr = sys.stderr 73 err = sys.stderr = StringIO() 74 75 try: 76 server.finish_request((inp, out), ("127.0.0.1",8888)) 77 finally: 78 sys.stderr = olderr 79 80 return out.getvalue(), err.getvalue() 81 82def compare_generic_iter(make_it,match): 83 """Utility to compare a generic 2.1/2.2+ iterator with an iterable 84 85 If running under Python 2.2+, this tests the iterator using iter()/next(), 86 as well as __getitem__. 'make_it' must be a function returning a fresh 87 iterator to be tested (since this may test the iterator twice).""" 88 89 it = make_it() 90 n = 0 91 for item in match: 92 if not it[n]==item: raise AssertionError 93 n+=1 94 try: 95 it[n] 96 except IndexError: 97 pass 98 else: 99 raise AssertionError("Too many items from __getitem__",it) 100 101 try: 102 iter, StopIteration 103 except NameError: 104 pass 105 else: 106 # Only test iter mode under 2.2+ 107 it = make_it() 108 if not iter(it) is it: raise AssertionError 109 for item in match: 110 if not next(it) == item: raise AssertionError 111 try: 112 next(it) 113 except StopIteration: 114 pass 115 else: 116 raise AssertionError("Too many items from .__next__()", it) 117 118 119class IntegrationTests(TestCase): 120 121 def check_hello(self, out, has_length=True): 122 pyver = (python_implementation() + "/" + 123 sys.version.split()[0]) 124 self.assertEqual(out, 125 ("HTTP/1.0 200 OK\r\n" 126 "Server: WSGIServer/0.2 " + pyver +"\r\n" 127 "Content-Type: text/plain\r\n" 128 "Date: Mon, 05 Jun 2006 18:49:54 GMT\r\n" + 129 (has_length and "Content-Length: 13\r\n" or "") + 130 "\r\n" 131 "Hello, world!").encode("iso-8859-1") 132 ) 133 134 def test_plain_hello(self): 135 out, err = run_amock() 136 self.check_hello(out) 137 138 def test_environ(self): 139 request = ( 140 b"GET /p%61th/?query=test HTTP/1.0\n" 141 b"X-Test-Header: Python test \n" 142 b"X-Test-Header: Python test 2\n" 143 b"Content-Length: 0\n\n" 144 ) 145 out, err = run_amock(header_app, request) 146 self.assertEqual( 147 out.splitlines()[-1], 148 b"Python test,Python test 2;query=test;/path/" 149 ) 150 151 def test_request_length(self): 152 out, err = run_amock(data=b"GET " + (b"x" * 65537) + b" HTTP/1.0\n\n") 153 self.assertEqual(out.splitlines()[0], 154 b"HTTP/1.0 414 Request-URI Too Long") 155 156 def test_validated_hello(self): 157 out, err = run_amock(validator(hello_app)) 158 # the middleware doesn't support len(), so content-length isn't there 159 self.check_hello(out, has_length=False) 160 161 def test_simple_validation_error(self): 162 def bad_app(environ,start_response): 163 start_response("200 OK", ('Content-Type','text/plain')) 164 return ["Hello, world!"] 165 out, err = run_amock(validator(bad_app)) 166 self.assertTrue(out.endswith( 167 b"A server error occurred. Please contact the administrator." 168 )) 169 self.assertEqual( 170 err.splitlines()[-2], 171 "AssertionError: Headers (('Content-Type', 'text/plain')) must" 172 " be of type list: <class 'tuple'>" 173 ) 174 175 def test_status_validation_errors(self): 176 def create_bad_app(status): 177 def bad_app(environ, start_response): 178 start_response(status, [("Content-Type", "text/plain; charset=utf-8")]) 179 return [b"Hello, world!"] 180 return bad_app 181 182 tests = [ 183 ('200', 'AssertionError: Status must be at least 4 characters'), 184 ('20X OK', 'AssertionError: Status message must begin w/3-digit code'), 185 ('200OK', 'AssertionError: Status message must have a space after code'), 186 ] 187 188 for status, exc_message in tests: 189 with self.subTest(status=status): 190 out, err = run_amock(create_bad_app(status)) 191 self.assertTrue(out.endswith( 192 b"A server error occurred. Please contact the administrator." 193 )) 194 self.assertEqual(err.splitlines()[-2], exc_message) 195 196 def test_wsgi_input(self): 197 def bad_app(e,s): 198 e["wsgi.input"].read() 199 s("200 OK", [("Content-Type", "text/plain; charset=utf-8")]) 200 return [b"data"] 201 out, err = run_amock(validator(bad_app)) 202 self.assertTrue(out.endswith( 203 b"A server error occurred. Please contact the administrator." 204 )) 205 self.assertEqual( 206 err.splitlines()[-2], "AssertionError" 207 ) 208 209 def test_bytes_validation(self): 210 def app(e, s): 211 s("200 OK", [ 212 ("Content-Type", "text/plain; charset=utf-8"), 213 ("Date", "Wed, 24 Dec 2008 13:29:32 GMT"), 214 ]) 215 return [b"data"] 216 out, err = run_amock(validator(app)) 217 self.assertTrue(err.endswith('"GET / HTTP/1.0" 200 4\n')) 218 ver = sys.version.split()[0].encode('ascii') 219 py = python_implementation().encode('ascii') 220 pyver = py + b"/" + ver 221 self.assertEqual( 222 b"HTTP/1.0 200 OK\r\n" 223 b"Server: WSGIServer/0.2 "+ pyver + b"\r\n" 224 b"Content-Type: text/plain; charset=utf-8\r\n" 225 b"Date: Wed, 24 Dec 2008 13:29:32 GMT\r\n" 226 b"\r\n" 227 b"data", 228 out) 229 230 def test_cp1252_url(self): 231 def app(e, s): 232 s("200 OK", [ 233 ("Content-Type", "text/plain"), 234 ("Date", "Wed, 24 Dec 2008 13:29:32 GMT"), 235 ]) 236 # PEP3333 says environ variables are decoded as latin1. 237 # Encode as latin1 to get original bytes 238 return [e["PATH_INFO"].encode("latin1")] 239 240 out, err = run_amock( 241 validator(app), data=b"GET /\x80%80 HTTP/1.0") 242 self.assertEqual( 243 [ 244 b"HTTP/1.0 200 OK", 245 mock.ANY, 246 b"Content-Type: text/plain", 247 b"Date: Wed, 24 Dec 2008 13:29:32 GMT", 248 b"", 249 b"/\x80\x80", 250 ], 251 out.splitlines()) 252 253 def test_interrupted_write(self): 254 # BaseHandler._write() and _flush() have to write all data, even if 255 # it takes multiple send() calls. Test this by interrupting a send() 256 # call with a Unix signal. 257 pthread_kill = support.get_attribute(signal, "pthread_kill") 258 259 def app(environ, start_response): 260 start_response("200 OK", []) 261 return [b'\0' * support.SOCK_MAX_SIZE] 262 263 class WsgiHandler(NoLogRequestHandler, WSGIRequestHandler): 264 pass 265 266 server = make_server(support.HOST, 0, app, handler_class=WsgiHandler) 267 self.addCleanup(server.server_close) 268 interrupted = threading.Event() 269 270 def signal_handler(signum, frame): 271 interrupted.set() 272 273 original = signal.signal(signal.SIGUSR1, signal_handler) 274 self.addCleanup(signal.signal, signal.SIGUSR1, original) 275 received = None 276 main_thread = threading.get_ident() 277 278 def run_client(): 279 http = HTTPConnection(*server.server_address) 280 http.request("GET", "/") 281 with http.getresponse() as response: 282 response.read(100) 283 # The main thread should now be blocking in a send() system 284 # call. But in theory, it could get interrupted by other 285 # signals, and then retried. So keep sending the signal in a 286 # loop, in case an earlier signal happens to be delivered at 287 # an inconvenient moment. 288 while True: 289 pthread_kill(main_thread, signal.SIGUSR1) 290 if interrupted.wait(timeout=float(1)): 291 break 292 nonlocal received 293 received = len(response.read()) 294 http.close() 295 296 background = threading.Thread(target=run_client) 297 background.start() 298 server.handle_request() 299 background.join() 300 self.assertEqual(received, support.SOCK_MAX_SIZE - 100) 301 302 303class UtilityTests(TestCase): 304 305 def checkShift(self,sn_in,pi_in,part,sn_out,pi_out): 306 env = {'SCRIPT_NAME':sn_in,'PATH_INFO':pi_in} 307 util.setup_testing_defaults(env) 308 self.assertEqual(util.shift_path_info(env),part) 309 self.assertEqual(env['PATH_INFO'],pi_out) 310 self.assertEqual(env['SCRIPT_NAME'],sn_out) 311 return env 312 313 def checkDefault(self, key, value, alt=None): 314 # Check defaulting when empty 315 env = {} 316 util.setup_testing_defaults(env) 317 if isinstance(value, StringIO): 318 self.assertIsInstance(env[key], StringIO) 319 elif isinstance(value,BytesIO): 320 self.assertIsInstance(env[key],BytesIO) 321 else: 322 self.assertEqual(env[key], value) 323 324 # Check existing value 325 env = {key:alt} 326 util.setup_testing_defaults(env) 327 self.assertIs(env[key], alt) 328 329 def checkCrossDefault(self,key,value,**kw): 330 util.setup_testing_defaults(kw) 331 self.assertEqual(kw[key],value) 332 333 def checkAppURI(self,uri,**kw): 334 util.setup_testing_defaults(kw) 335 self.assertEqual(util.application_uri(kw),uri) 336 337 def checkReqURI(self,uri,query=1,**kw): 338 util.setup_testing_defaults(kw) 339 self.assertEqual(util.request_uri(kw,query),uri) 340 341 @support.ignore_warnings(category=DeprecationWarning) 342 def checkFW(self,text,size,match): 343 344 def make_it(text=text,size=size): 345 return util.FileWrapper(StringIO(text),size) 346 347 compare_generic_iter(make_it,match) 348 349 it = make_it() 350 self.assertFalse(it.filelike.closed) 351 352 for item in it: 353 pass 354 355 self.assertFalse(it.filelike.closed) 356 357 it.close() 358 self.assertTrue(it.filelike.closed) 359 360 def test_filewrapper_getitem_deprecation(self): 361 wrapper = util.FileWrapper(StringIO('foobar'), 3) 362 with self.assertWarnsRegex(DeprecationWarning, 363 r'Use iterator protocol instead'): 364 # This should have returned 'bar'. 365 self.assertEqual(wrapper[1], 'foo') 366 367 def testSimpleShifts(self): 368 self.checkShift('','/', '', '/', '') 369 self.checkShift('','/x', 'x', '/x', '') 370 self.checkShift('/','', None, '/', '') 371 self.checkShift('/a','/x/y', 'x', '/a/x', '/y') 372 self.checkShift('/a','/x/', 'x', '/a/x', '/') 373 374 def testNormalizedShifts(self): 375 self.checkShift('/a/b', '/../y', '..', '/a', '/y') 376 self.checkShift('', '/../y', '..', '', '/y') 377 self.checkShift('/a/b', '//y', 'y', '/a/b/y', '') 378 self.checkShift('/a/b', '//y/', 'y', '/a/b/y', '/') 379 self.checkShift('/a/b', '/./y', 'y', '/a/b/y', '') 380 self.checkShift('/a/b', '/./y/', 'y', '/a/b/y', '/') 381 self.checkShift('/a/b', '///./..//y/.//', '..', '/a', '/y/') 382 self.checkShift('/a/b', '///', '', '/a/b/', '') 383 self.checkShift('/a/b', '/.//', '', '/a/b/', '') 384 self.checkShift('/a/b', '/x//', 'x', '/a/b/x', '/') 385 self.checkShift('/a/b', '/.', None, '/a/b', '') 386 387 def testDefaults(self): 388 for key, value in [ 389 ('SERVER_NAME','127.0.0.1'), 390 ('SERVER_PORT', '80'), 391 ('SERVER_PROTOCOL','HTTP/1.0'), 392 ('HTTP_HOST','127.0.0.1'), 393 ('REQUEST_METHOD','GET'), 394 ('SCRIPT_NAME',''), 395 ('PATH_INFO','/'), 396 ('wsgi.version', (1,0)), 397 ('wsgi.run_once', 0), 398 ('wsgi.multithread', 0), 399 ('wsgi.multiprocess', 0), 400 ('wsgi.input', BytesIO()), 401 ('wsgi.errors', StringIO()), 402 ('wsgi.url_scheme','http'), 403 ]: 404 self.checkDefault(key,value) 405 406 def testCrossDefaults(self): 407 self.checkCrossDefault('HTTP_HOST',"foo.bar",SERVER_NAME="foo.bar") 408 self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="on") 409 self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="1") 410 self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="yes") 411 self.checkCrossDefault('wsgi.url_scheme',"http",HTTPS="foo") 412 self.checkCrossDefault('SERVER_PORT',"80",HTTPS="foo") 413 self.checkCrossDefault('SERVER_PORT',"443",HTTPS="on") 414 415 def testGuessScheme(self): 416 self.assertEqual(util.guess_scheme({}), "http") 417 self.assertEqual(util.guess_scheme({'HTTPS':"foo"}), "http") 418 self.assertEqual(util.guess_scheme({'HTTPS':"on"}), "https") 419 self.assertEqual(util.guess_scheme({'HTTPS':"yes"}), "https") 420 self.assertEqual(util.guess_scheme({'HTTPS':"1"}), "https") 421 422 def testAppURIs(self): 423 self.checkAppURI("http://127.0.0.1/") 424 self.checkAppURI("http://127.0.0.1/spam", SCRIPT_NAME="/spam") 425 self.checkAppURI("http://127.0.0.1/sp%E4m", SCRIPT_NAME="/sp\xe4m") 426 self.checkAppURI("http://spam.example.com:2071/", 427 HTTP_HOST="spam.example.com:2071", SERVER_PORT="2071") 428 self.checkAppURI("http://spam.example.com/", 429 SERVER_NAME="spam.example.com") 430 self.checkAppURI("http://127.0.0.1/", 431 HTTP_HOST="127.0.0.1", SERVER_NAME="spam.example.com") 432 self.checkAppURI("https://127.0.0.1/", HTTPS="on") 433 self.checkAppURI("http://127.0.0.1:8000/", SERVER_PORT="8000", 434 HTTP_HOST=None) 435 436 def testReqURIs(self): 437 self.checkReqURI("http://127.0.0.1/") 438 self.checkReqURI("http://127.0.0.1/spam", SCRIPT_NAME="/spam") 439 self.checkReqURI("http://127.0.0.1/sp%E4m", SCRIPT_NAME="/sp\xe4m") 440 self.checkReqURI("http://127.0.0.1/spammity/spam", 441 SCRIPT_NAME="/spammity", PATH_INFO="/spam") 442 self.checkReqURI("http://127.0.0.1/spammity/sp%E4m", 443 SCRIPT_NAME="/spammity", PATH_INFO="/sp\xe4m") 444 self.checkReqURI("http://127.0.0.1/spammity/spam;ham", 445 SCRIPT_NAME="/spammity", PATH_INFO="/spam;ham") 446 self.checkReqURI("http://127.0.0.1/spammity/spam;cookie=1234,5678", 447 SCRIPT_NAME="/spammity", PATH_INFO="/spam;cookie=1234,5678") 448 self.checkReqURI("http://127.0.0.1/spammity/spam?say=ni", 449 SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="say=ni") 450 self.checkReqURI("http://127.0.0.1/spammity/spam?s%E4y=ni", 451 SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="s%E4y=ni") 452 self.checkReqURI("http://127.0.0.1/spammity/spam", 0, 453 SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="say=ni") 454 455 def testFileWrapper(self): 456 self.checkFW("xyz"*50, 120, ["xyz"*40,"xyz"*10]) 457 458 def testHopByHop(self): 459 for hop in ( 460 "Connection Keep-Alive Proxy-Authenticate Proxy-Authorization " 461 "TE Trailers Transfer-Encoding Upgrade" 462 ).split(): 463 for alt in hop, hop.title(), hop.upper(), hop.lower(): 464 self.assertTrue(util.is_hop_by_hop(alt)) 465 466 # Not comprehensive, just a few random header names 467 for hop in ( 468 "Accept Cache-Control Date Pragma Trailer Via Warning" 469 ).split(): 470 for alt in hop, hop.title(), hop.upper(), hop.lower(): 471 self.assertFalse(util.is_hop_by_hop(alt)) 472 473class HeaderTests(TestCase): 474 475 def testMappingInterface(self): 476 test = [('x','y')] 477 self.assertEqual(len(Headers()), 0) 478 self.assertEqual(len(Headers([])),0) 479 self.assertEqual(len(Headers(test[:])),1) 480 self.assertEqual(Headers(test[:]).keys(), ['x']) 481 self.assertEqual(Headers(test[:]).values(), ['y']) 482 self.assertEqual(Headers(test[:]).items(), test) 483 self.assertIsNot(Headers(test).items(), test) # must be copy! 484 485 h = Headers() 486 del h['foo'] # should not raise an error 487 488 h['Foo'] = 'bar' 489 for m in h.__contains__, h.get, h.get_all, h.__getitem__: 490 self.assertTrue(m('foo')) 491 self.assertTrue(m('Foo')) 492 self.assertTrue(m('FOO')) 493 self.assertFalse(m('bar')) 494 495 self.assertEqual(h['foo'],'bar') 496 h['foo'] = 'baz' 497 self.assertEqual(h['FOO'],'baz') 498 self.assertEqual(h.get_all('foo'),['baz']) 499 500 self.assertEqual(h.get("foo","whee"), "baz") 501 self.assertEqual(h.get("zoo","whee"), "whee") 502 self.assertEqual(h.setdefault("foo","whee"), "baz") 503 self.assertEqual(h.setdefault("zoo","whee"), "whee") 504 self.assertEqual(h["foo"],"baz") 505 self.assertEqual(h["zoo"],"whee") 506 507 def testRequireList(self): 508 self.assertRaises(TypeError, Headers, "foo") 509 510 def testExtras(self): 511 h = Headers() 512 self.assertEqual(str(h),'\r\n') 513 514 h.add_header('foo','bar',baz="spam") 515 self.assertEqual(h['foo'], 'bar; baz="spam"') 516 self.assertEqual(str(h),'foo: bar; baz="spam"\r\n\r\n') 517 518 h.add_header('Foo','bar',cheese=None) 519 self.assertEqual(h.get_all('foo'), 520 ['bar; baz="spam"', 'bar; cheese']) 521 522 self.assertEqual(str(h), 523 'foo: bar; baz="spam"\r\n' 524 'Foo: bar; cheese\r\n' 525 '\r\n' 526 ) 527 528class ErrorHandler(BaseCGIHandler): 529 """Simple handler subclass for testing BaseHandler""" 530 531 # BaseHandler records the OS environment at import time, but envvars 532 # might have been changed later by other tests, which trips up 533 # HandlerTests.testEnviron(). 534 os_environ = dict(os.environ.items()) 535 536 def __init__(self,**kw): 537 setup_testing_defaults(kw) 538 BaseCGIHandler.__init__( 539 self, BytesIO(), BytesIO(), StringIO(), kw, 540 multithread=True, multiprocess=True 541 ) 542 543class TestHandler(ErrorHandler): 544 """Simple handler subclass for testing BaseHandler, w/error passthru""" 545 546 def handle_error(self): 547 raise # for testing, we want to see what's happening 548 549 550class HandlerTests(TestCase): 551 # testEnviron() can produce long error message 552 maxDiff = 80 * 50 553 554 def testEnviron(self): 555 os_environ = { 556 # very basic environment 557 'HOME': '/my/home', 558 'PATH': '/my/path', 559 'LANG': 'fr_FR.UTF-8', 560 561 # set some WSGI variables 562 'SCRIPT_NAME': 'test_script_name', 563 'SERVER_NAME': 'test_server_name', 564 } 565 566 with support.swap_attr(TestHandler, 'os_environ', os_environ): 567 # override X and HOME variables 568 handler = TestHandler(X="Y", HOME="/override/home") 569 handler.setup_environ() 570 571 # Check that wsgi_xxx attributes are copied to wsgi.xxx variables 572 # of handler.environ 573 for attr in ('version', 'multithread', 'multiprocess', 'run_once', 574 'file_wrapper'): 575 self.assertEqual(getattr(handler, 'wsgi_' + attr), 576 handler.environ['wsgi.' + attr]) 577 578 # Test handler.environ as a dict 579 expected = {} 580 setup_testing_defaults(expected) 581 # Handler inherits os_environ variables which are not overriden 582 # by SimpleHandler.add_cgi_vars() (SimpleHandler.base_env) 583 for key, value in os_environ.items(): 584 if key not in expected: 585 expected[key] = value 586 expected.update({ 587 # X doesn't exist in os_environ 588 "X": "Y", 589 # HOME is overridden by TestHandler 590 'HOME': "/override/home", 591 592 # overridden by setup_testing_defaults() 593 "SCRIPT_NAME": "", 594 "SERVER_NAME": "127.0.0.1", 595 596 # set by BaseHandler.setup_environ() 597 'wsgi.input': handler.get_stdin(), 598 'wsgi.errors': handler.get_stderr(), 599 'wsgi.version': (1, 0), 600 'wsgi.run_once': False, 601 'wsgi.url_scheme': 'http', 602 'wsgi.multithread': True, 603 'wsgi.multiprocess': True, 604 'wsgi.file_wrapper': util.FileWrapper, 605 }) 606 self.assertDictEqual(handler.environ, expected) 607 608 def testCGIEnviron(self): 609 h = BaseCGIHandler(None,None,None,{}) 610 h.setup_environ() 611 for key in 'wsgi.url_scheme', 'wsgi.input', 'wsgi.errors': 612 self.assertIn(key, h.environ) 613 614 def testScheme(self): 615 h=TestHandler(HTTPS="on"); h.setup_environ() 616 self.assertEqual(h.environ['wsgi.url_scheme'],'https') 617 h=TestHandler(); h.setup_environ() 618 self.assertEqual(h.environ['wsgi.url_scheme'],'http') 619 620 def testAbstractMethods(self): 621 h = BaseHandler() 622 for name in [ 623 '_flush','get_stdin','get_stderr','add_cgi_vars' 624 ]: 625 self.assertRaises(NotImplementedError, getattr(h,name)) 626 self.assertRaises(NotImplementedError, h._write, "test") 627 628 def testContentLength(self): 629 # Demo one reason iteration is better than write()... ;) 630 631 def trivial_app1(e,s): 632 s('200 OK',[]) 633 return [e['wsgi.url_scheme'].encode('iso-8859-1')] 634 635 def trivial_app2(e,s): 636 s('200 OK',[])(e['wsgi.url_scheme'].encode('iso-8859-1')) 637 return [] 638 639 def trivial_app3(e,s): 640 s('200 OK',[]) 641 return ['\u0442\u0435\u0441\u0442'.encode("utf-8")] 642 643 def trivial_app4(e,s): 644 # Simulate a response to a HEAD request 645 s('200 OK',[('Content-Length', '12345')]) 646 return [] 647 648 h = TestHandler() 649 h.run(trivial_app1) 650 self.assertEqual(h.stdout.getvalue(), 651 ("Status: 200 OK\r\n" 652 "Content-Length: 4\r\n" 653 "\r\n" 654 "http").encode("iso-8859-1")) 655 656 h = TestHandler() 657 h.run(trivial_app2) 658 self.assertEqual(h.stdout.getvalue(), 659 ("Status: 200 OK\r\n" 660 "\r\n" 661 "http").encode("iso-8859-1")) 662 663 h = TestHandler() 664 h.run(trivial_app3) 665 self.assertEqual(h.stdout.getvalue(), 666 b'Status: 200 OK\r\n' 667 b'Content-Length: 8\r\n' 668 b'\r\n' 669 b'\xd1\x82\xd0\xb5\xd1\x81\xd1\x82') 670 671 h = TestHandler() 672 h.run(trivial_app4) 673 self.assertEqual(h.stdout.getvalue(), 674 b'Status: 200 OK\r\n' 675 b'Content-Length: 12345\r\n' 676 b'\r\n') 677 678 def testBasicErrorOutput(self): 679 680 def non_error_app(e,s): 681 s('200 OK',[]) 682 return [] 683 684 def error_app(e,s): 685 raise AssertionError("This should be caught by handler") 686 687 h = ErrorHandler() 688 h.run(non_error_app) 689 self.assertEqual(h.stdout.getvalue(), 690 ("Status: 200 OK\r\n" 691 "Content-Length: 0\r\n" 692 "\r\n").encode("iso-8859-1")) 693 self.assertEqual(h.stderr.getvalue(),"") 694 695 h = ErrorHandler() 696 h.run(error_app) 697 self.assertEqual(h.stdout.getvalue(), 698 ("Status: %s\r\n" 699 "Content-Type: text/plain\r\n" 700 "Content-Length: %d\r\n" 701 "\r\n" % (h.error_status,len(h.error_body))).encode('iso-8859-1') 702 + h.error_body) 703 704 self.assertIn("AssertionError", h.stderr.getvalue()) 705 706 def testErrorAfterOutput(self): 707 MSG = b"Some output has been sent" 708 def error_app(e,s): 709 s("200 OK",[])(MSG) 710 raise AssertionError("This should be caught by handler") 711 712 h = ErrorHandler() 713 h.run(error_app) 714 self.assertEqual(h.stdout.getvalue(), 715 ("Status: 200 OK\r\n" 716 "\r\n".encode("iso-8859-1")+MSG)) 717 self.assertIn("AssertionError", h.stderr.getvalue()) 718 719 def testHeaderFormats(self): 720 721 def non_error_app(e,s): 722 s('200 OK',[]) 723 return [] 724 725 stdpat = ( 726 r"HTTP/%s 200 OK\r\n" 727 r"Date: \w{3}, [ 0123]\d \w{3} \d{4} \d\d:\d\d:\d\d GMT\r\n" 728 r"%s" r"Content-Length: 0\r\n" r"\r\n" 729 ) 730 shortpat = ( 731 "Status: 200 OK\r\n" "Content-Length: 0\r\n" "\r\n" 732 ).encode("iso-8859-1") 733 734 for ssw in "FooBar/1.0", None: 735 sw = ssw and "Server: %s\r\n" % ssw or "" 736 737 for version in "1.0", "1.1": 738 for proto in "HTTP/0.9", "HTTP/1.0", "HTTP/1.1": 739 740 h = TestHandler(SERVER_PROTOCOL=proto) 741 h.origin_server = False 742 h.http_version = version 743 h.server_software = ssw 744 h.run(non_error_app) 745 self.assertEqual(shortpat,h.stdout.getvalue()) 746 747 h = TestHandler(SERVER_PROTOCOL=proto) 748 h.origin_server = True 749 h.http_version = version 750 h.server_software = ssw 751 h.run(non_error_app) 752 if proto=="HTTP/0.9": 753 self.assertEqual(h.stdout.getvalue(),b"") 754 else: 755 self.assertTrue( 756 re.match((stdpat%(version,sw)).encode("iso-8859-1"), 757 h.stdout.getvalue()), 758 ((stdpat%(version,sw)).encode("iso-8859-1"), 759 h.stdout.getvalue()) 760 ) 761 762 def testBytesData(self): 763 def app(e, s): 764 s("200 OK", [ 765 ("Content-Type", "text/plain; charset=utf-8"), 766 ]) 767 return [b"data"] 768 769 h = TestHandler() 770 h.run(app) 771 self.assertEqual(b"Status: 200 OK\r\n" 772 b"Content-Type: text/plain; charset=utf-8\r\n" 773 b"Content-Length: 4\r\n" 774 b"\r\n" 775 b"data", 776 h.stdout.getvalue()) 777 778 def testCloseOnError(self): 779 side_effects = {'close_called': False} 780 MSG = b"Some output has been sent" 781 def error_app(e,s): 782 s("200 OK",[])(MSG) 783 class CrashyIterable(object): 784 def __iter__(self): 785 while True: 786 yield b'blah' 787 raise AssertionError("This should be caught by handler") 788 def close(self): 789 side_effects['close_called'] = True 790 return CrashyIterable() 791 792 h = ErrorHandler() 793 h.run(error_app) 794 self.assertEqual(side_effects['close_called'], True) 795 796 def testPartialWrite(self): 797 written = bytearray() 798 799 class PartialWriter: 800 def write(self, b): 801 partial = b[:7] 802 written.extend(partial) 803 return len(partial) 804 805 def flush(self): 806 pass 807 808 environ = {"SERVER_PROTOCOL": "HTTP/1.0"} 809 h = SimpleHandler(BytesIO(), PartialWriter(), sys.stderr, environ) 810 msg = "should not do partial writes" 811 with self.assertWarnsRegex(DeprecationWarning, msg): 812 h.run(hello_app) 813 self.assertEqual(b"HTTP/1.0 200 OK\r\n" 814 b"Content-Type: text/plain\r\n" 815 b"Date: Mon, 05 Jun 2006 18:49:54 GMT\r\n" 816 b"Content-Length: 13\r\n" 817 b"\r\n" 818 b"Hello, world!", 819 written) 820 821 def testClientConnectionTerminations(self): 822 environ = {"SERVER_PROTOCOL": "HTTP/1.0"} 823 for exception in ( 824 ConnectionAbortedError, 825 BrokenPipeError, 826 ConnectionResetError, 827 ): 828 with self.subTest(exception=exception): 829 class AbortingWriter: 830 def write(self, b): 831 raise exception 832 833 stderr = StringIO() 834 h = SimpleHandler(BytesIO(), AbortingWriter(), stderr, environ) 835 h.run(hello_app) 836 837 self.assertFalse(stderr.getvalue()) 838 839 def testDontResetInternalStateOnException(self): 840 class CustomException(ValueError): 841 pass 842 843 # We are raising CustomException here to trigger an exception 844 # during the execution of SimpleHandler.finish_response(), so 845 # we can easily test that the internal state of the handler is 846 # preserved in case of an exception. 847 class AbortingWriter: 848 def write(self, b): 849 raise CustomException 850 851 stderr = StringIO() 852 environ = {"SERVER_PROTOCOL": "HTTP/1.0"} 853 h = SimpleHandler(BytesIO(), AbortingWriter(), stderr, environ) 854 h.run(hello_app) 855 856 self.assertIn("CustomException", stderr.getvalue()) 857 858 # Test that the internal state of the handler is preserved. 859 self.assertIsNotNone(h.result) 860 self.assertIsNotNone(h.headers) 861 self.assertIsNotNone(h.status) 862 self.assertIsNotNone(h.environ) 863 864 865if __name__ == "__main__": 866 unittest.main() 867