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