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 unittest
22
23
24class MockServer(WSGIServer):
25    """Non-socket HTTP server"""
26
27    def __init__(self, server_address, RequestHandlerClass):
28        BaseServer.__init__(self, server_address, RequestHandlerClass)
29        self.server_bind()
30
31    def server_bind(self):
32        host, port = self.server_address
33        self.server_name = host
34        self.server_port = port
35        self.setup_environ()
36
37
38class MockHandler(WSGIRequestHandler):
39    """Non-socket HTTP handler"""
40    def setup(self):
41        self.connection = self.request
42        self.rfile, self.wfile = self.connection
43
44    def finish(self):
45        pass
46
47
48def hello_app(environ,start_response):
49    start_response("200 OK", [
50        ('Content-Type','text/plain'),
51        ('Date','Mon, 05 Jun 2006 18:49:54 GMT')
52    ])
53    return [b"Hello, world!"]
54
55
56def header_app(environ, start_response):
57    start_response("200 OK", [
58        ('Content-Type', 'text/plain'),
59        ('Date', 'Mon, 05 Jun 2006 18:49:54 GMT')
60    ])
61    return [';'.join([
62        environ['HTTP_X_TEST_HEADER'], environ['QUERY_STRING'],
63        environ['PATH_INFO']
64    ]).encode('iso-8859-1')]
65
66
67def run_amock(app=hello_app, data=b"GET / HTTP/1.0\n\n"):
68    server = make_server("", 80, app, MockServer, MockHandler)
69    inp = BufferedReader(BytesIO(data))
70    out = BytesIO()
71    olderr = sys.stderr
72    err = sys.stderr = StringIO()
73
74    try:
75        server.finish_request((inp, out), ("127.0.0.1",8888))
76    finally:
77        sys.stderr = olderr
78
79    return out.getvalue(), err.getvalue()
80
81def compare_generic_iter(make_it,match):
82    """Utility to compare a generic 2.1/2.2+ iterator with an iterable
83
84    If running under Python 2.2+, this tests the iterator using iter()/next(),
85    as well as __getitem__.  'make_it' must be a function returning a fresh
86    iterator to be tested (since this may test the iterator twice)."""
87
88    it = make_it()
89    n = 0
90    for item in match:
91        if not it[n]==item: raise AssertionError
92        n+=1
93    try:
94        it[n]
95    except IndexError:
96        pass
97    else:
98        raise AssertionError("Too many items from __getitem__",it)
99
100    try:
101        iter, StopIteration
102    except NameError:
103        pass
104    else:
105        # Only test iter mode under 2.2+
106        it = make_it()
107        if not iter(it) is it: raise AssertionError
108        for item in match:
109            if not next(it) == item: raise AssertionError
110        try:
111            next(it)
112        except StopIteration:
113            pass
114        else:
115            raise AssertionError("Too many items from .__next__()", it)
116
117
118class IntegrationTests(TestCase):
119
120    def check_hello(self, out, has_length=True):
121        pyver = (python_implementation() + "/" +
122                sys.version.split()[0])
123        self.assertEqual(out,
124            ("HTTP/1.0 200 OK\r\n"
125            "Server: WSGIServer/0.2 " + pyver +"\r\n"
126            "Content-Type: text/plain\r\n"
127            "Date: Mon, 05 Jun 2006 18:49:54 GMT\r\n" +
128            (has_length and  "Content-Length: 13\r\n" or "") +
129            "\r\n"
130            "Hello, world!").encode("iso-8859-1")
131        )
132
133    def test_plain_hello(self):
134        out, err = run_amock()
135        self.check_hello(out)
136
137    def test_environ(self):
138        request = (
139            b"GET /p%61th/?query=test HTTP/1.0\n"
140            b"X-Test-Header: Python test \n"
141            b"X-Test-Header: Python test 2\n"
142            b"Content-Length: 0\n\n"
143        )
144        out, err = run_amock(header_app, request)
145        self.assertEqual(
146            out.splitlines()[-1],
147            b"Python test,Python test 2;query=test;/path/"
148        )
149
150    def test_request_length(self):
151        out, err = run_amock(data=b"GET " + (b"x" * 65537) + b" HTTP/1.0\n\n")
152        self.assertEqual(out.splitlines()[0],
153                         b"HTTP/1.0 414 Request-URI Too Long")
154
155    def test_validated_hello(self):
156        out, err = run_amock(validator(hello_app))
157        # the middleware doesn't support len(), so content-length isn't there
158        self.check_hello(out, has_length=False)
159
160    def test_simple_validation_error(self):
161        def bad_app(environ,start_response):
162            start_response("200 OK", ('Content-Type','text/plain'))
163            return ["Hello, world!"]
164        out, err = run_amock(validator(bad_app))
165        self.assertTrue(out.endswith(
166            b"A server error occurred.  Please contact the administrator."
167        ))
168        self.assertEqual(
169            err.splitlines()[-2],
170            "AssertionError: Headers (('Content-Type', 'text/plain')) must"
171            " be of type list: <class 'tuple'>"
172        )
173
174    def test_status_validation_errors(self):
175        def create_bad_app(status):
176            def bad_app(environ, start_response):
177                start_response(status, [("Content-Type", "text/plain; charset=utf-8")])
178                return [b"Hello, world!"]
179            return bad_app
180
181        tests = [
182            ('200', 'AssertionError: Status must be at least 4 characters'),
183            ('20X OK', 'AssertionError: Status message must begin w/3-digit code'),
184            ('200OK', 'AssertionError: Status message must have a space after code'),
185        ]
186
187        for status, exc_message in tests:
188            with self.subTest(status=status):
189                out, err = run_amock(create_bad_app(status))
190                self.assertTrue(out.endswith(
191                    b"A server error occurred.  Please contact the administrator."
192                ))
193                self.assertEqual(err.splitlines()[-2], exc_message)
194
195    def test_wsgi_input(self):
196        def bad_app(e,s):
197            e["wsgi.input"].read()
198            s("200 OK", [("Content-Type", "text/plain; charset=utf-8")])
199            return [b"data"]
200        out, err = run_amock(validator(bad_app))
201        self.assertTrue(out.endswith(
202            b"A server error occurred.  Please contact the administrator."
203        ))
204        self.assertEqual(
205            err.splitlines()[-2], "AssertionError"
206        )
207
208    def test_bytes_validation(self):
209        def app(e, s):
210            s("200 OK", [
211                ("Content-Type", "text/plain; charset=utf-8"),
212                ("Date", "Wed, 24 Dec 2008 13:29:32 GMT"),
213                ])
214            return [b"data"]
215        out, err = run_amock(validator(app))
216        self.assertTrue(err.endswith('"GET / HTTP/1.0" 200 4\n'))
217        ver = sys.version.split()[0].encode('ascii')
218        py  = python_implementation().encode('ascii')
219        pyver = py + b"/" + ver
220        self.assertEqual(
221                b"HTTP/1.0 200 OK\r\n"
222                b"Server: WSGIServer/0.2 "+ pyver + b"\r\n"
223                b"Content-Type: text/plain; charset=utf-8\r\n"
224                b"Date: Wed, 24 Dec 2008 13:29:32 GMT\r\n"
225                b"\r\n"
226                b"data",
227                out)
228
229    def test_cp1252_url(self):
230        def app(e, s):
231            s("200 OK", [
232                ("Content-Type", "text/plain"),
233                ("Date", "Wed, 24 Dec 2008 13:29:32 GMT"),
234                ])
235            # PEP3333 says environ variables are decoded as latin1.
236            # Encode as latin1 to get original bytes
237            return [e["PATH_INFO"].encode("latin1")]
238
239        out, err = run_amock(
240            validator(app), data=b"GET /\x80%80 HTTP/1.0")
241        self.assertEqual(
242            [
243                b"HTTP/1.0 200 OK",
244                mock.ANY,
245                b"Content-Type: text/plain",
246                b"Date: Wed, 24 Dec 2008 13:29:32 GMT",
247                b"",
248                b"/\x80\x80",
249            ],
250            out.splitlines())
251
252    def test_interrupted_write(self):
253        # BaseHandler._write() and _flush() have to write all data, even if
254        # it takes multiple send() calls.  Test this by interrupting a send()
255        # call with a Unix signal.
256        threading = support.import_module("threading")
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    def checkFW(self,text,size,match):
342
343        def make_it(text=text,size=size):
344            return util.FileWrapper(StringIO(text),size)
345
346        compare_generic_iter(make_it,match)
347
348        it = make_it()
349        self.assertFalse(it.filelike.closed)
350
351        for item in it:
352            pass
353
354        self.assertFalse(it.filelike.closed)
355
356        it.close()
357        self.assertTrue(it.filelike.closed)
358
359    def testSimpleShifts(self):
360        self.checkShift('','/', '', '/', '')
361        self.checkShift('','/x', 'x', '/x', '')
362        self.checkShift('/','', None, '/', '')
363        self.checkShift('/a','/x/y', 'x', '/a/x', '/y')
364        self.checkShift('/a','/x/',  'x', '/a/x', '/')
365
366    def testNormalizedShifts(self):
367        self.checkShift('/a/b', '/../y', '..', '/a', '/y')
368        self.checkShift('', '/../y', '..', '', '/y')
369        self.checkShift('/a/b', '//y', 'y', '/a/b/y', '')
370        self.checkShift('/a/b', '//y/', 'y', '/a/b/y', '/')
371        self.checkShift('/a/b', '/./y', 'y', '/a/b/y', '')
372        self.checkShift('/a/b', '/./y/', 'y', '/a/b/y', '/')
373        self.checkShift('/a/b', '///./..//y/.//', '..', '/a', '/y/')
374        self.checkShift('/a/b', '///', '', '/a/b/', '')
375        self.checkShift('/a/b', '/.//', '', '/a/b/', '')
376        self.checkShift('/a/b', '/x//', 'x', '/a/b/x', '/')
377        self.checkShift('/a/b', '/.', None, '/a/b', '')
378
379    def testDefaults(self):
380        for key, value in [
381            ('SERVER_NAME','127.0.0.1'),
382            ('SERVER_PORT', '80'),
383            ('SERVER_PROTOCOL','HTTP/1.0'),
384            ('HTTP_HOST','127.0.0.1'),
385            ('REQUEST_METHOD','GET'),
386            ('SCRIPT_NAME',''),
387            ('PATH_INFO','/'),
388            ('wsgi.version', (1,0)),
389            ('wsgi.run_once', 0),
390            ('wsgi.multithread', 0),
391            ('wsgi.multiprocess', 0),
392            ('wsgi.input', BytesIO()),
393            ('wsgi.errors', StringIO()),
394            ('wsgi.url_scheme','http'),
395        ]:
396            self.checkDefault(key,value)
397
398    def testCrossDefaults(self):
399        self.checkCrossDefault('HTTP_HOST',"foo.bar",SERVER_NAME="foo.bar")
400        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="on")
401        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="1")
402        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="yes")
403        self.checkCrossDefault('wsgi.url_scheme',"http",HTTPS="foo")
404        self.checkCrossDefault('SERVER_PORT',"80",HTTPS="foo")
405        self.checkCrossDefault('SERVER_PORT',"443",HTTPS="on")
406
407    def testGuessScheme(self):
408        self.assertEqual(util.guess_scheme({}), "http")
409        self.assertEqual(util.guess_scheme({'HTTPS':"foo"}), "http")
410        self.assertEqual(util.guess_scheme({'HTTPS':"on"}), "https")
411        self.assertEqual(util.guess_scheme({'HTTPS':"yes"}), "https")
412        self.assertEqual(util.guess_scheme({'HTTPS':"1"}), "https")
413
414    def testAppURIs(self):
415        self.checkAppURI("http://127.0.0.1/")
416        self.checkAppURI("http://127.0.0.1/spam", SCRIPT_NAME="/spam")
417        self.checkAppURI("http://127.0.0.1/sp%E4m", SCRIPT_NAME="/sp\xe4m")
418        self.checkAppURI("http://spam.example.com:2071/",
419            HTTP_HOST="spam.example.com:2071", SERVER_PORT="2071")
420        self.checkAppURI("http://spam.example.com/",
421            SERVER_NAME="spam.example.com")
422        self.checkAppURI("http://127.0.0.1/",
423            HTTP_HOST="127.0.0.1", SERVER_NAME="spam.example.com")
424        self.checkAppURI("https://127.0.0.1/", HTTPS="on")
425        self.checkAppURI("http://127.0.0.1:8000/", SERVER_PORT="8000",
426            HTTP_HOST=None)
427
428    def testReqURIs(self):
429        self.checkReqURI("http://127.0.0.1/")
430        self.checkReqURI("http://127.0.0.1/spam", SCRIPT_NAME="/spam")
431        self.checkReqURI("http://127.0.0.1/sp%E4m", SCRIPT_NAME="/sp\xe4m")
432        self.checkReqURI("http://127.0.0.1/spammity/spam",
433            SCRIPT_NAME="/spammity", PATH_INFO="/spam")
434        self.checkReqURI("http://127.0.0.1/spammity/sp%E4m",
435            SCRIPT_NAME="/spammity", PATH_INFO="/sp\xe4m")
436        self.checkReqURI("http://127.0.0.1/spammity/spam;ham",
437            SCRIPT_NAME="/spammity", PATH_INFO="/spam;ham")
438        self.checkReqURI("http://127.0.0.1/spammity/spam;cookie=1234,5678",
439            SCRIPT_NAME="/spammity", PATH_INFO="/spam;cookie=1234,5678")
440        self.checkReqURI("http://127.0.0.1/spammity/spam?say=ni",
441            SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="say=ni")
442        self.checkReqURI("http://127.0.0.1/spammity/spam?s%E4y=ni",
443            SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="s%E4y=ni")
444        self.checkReqURI("http://127.0.0.1/spammity/spam", 0,
445            SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="say=ni")
446
447    def testFileWrapper(self):
448        self.checkFW("xyz"*50, 120, ["xyz"*40,"xyz"*10])
449
450    def testHopByHop(self):
451        for hop in (
452            "Connection Keep-Alive Proxy-Authenticate Proxy-Authorization "
453            "TE Trailers Transfer-Encoding Upgrade"
454        ).split():
455            for alt in hop, hop.title(), hop.upper(), hop.lower():
456                self.assertTrue(util.is_hop_by_hop(alt))
457
458        # Not comprehensive, just a few random header names
459        for hop in (
460            "Accept Cache-Control Date Pragma Trailer Via Warning"
461        ).split():
462            for alt in hop, hop.title(), hop.upper(), hop.lower():
463                self.assertFalse(util.is_hop_by_hop(alt))
464
465class HeaderTests(TestCase):
466
467    def testMappingInterface(self):
468        test = [('x','y')]
469        self.assertEqual(len(Headers()), 0)
470        self.assertEqual(len(Headers([])),0)
471        self.assertEqual(len(Headers(test[:])),1)
472        self.assertEqual(Headers(test[:]).keys(), ['x'])
473        self.assertEqual(Headers(test[:]).values(), ['y'])
474        self.assertEqual(Headers(test[:]).items(), test)
475        self.assertIsNot(Headers(test).items(), test)  # must be copy!
476
477        h = Headers()
478        del h['foo']   # should not raise an error
479
480        h['Foo'] = 'bar'
481        for m in h.__contains__, h.get, h.get_all, h.__getitem__:
482            self.assertTrue(m('foo'))
483            self.assertTrue(m('Foo'))
484            self.assertTrue(m('FOO'))
485            self.assertFalse(m('bar'))
486
487        self.assertEqual(h['foo'],'bar')
488        h['foo'] = 'baz'
489        self.assertEqual(h['FOO'],'baz')
490        self.assertEqual(h.get_all('foo'),['baz'])
491
492        self.assertEqual(h.get("foo","whee"), "baz")
493        self.assertEqual(h.get("zoo","whee"), "whee")
494        self.assertEqual(h.setdefault("foo","whee"), "baz")
495        self.assertEqual(h.setdefault("zoo","whee"), "whee")
496        self.assertEqual(h["foo"],"baz")
497        self.assertEqual(h["zoo"],"whee")
498
499    def testRequireList(self):
500        self.assertRaises(TypeError, Headers, "foo")
501
502    def testExtras(self):
503        h = Headers()
504        self.assertEqual(str(h),'\r\n')
505
506        h.add_header('foo','bar',baz="spam")
507        self.assertEqual(h['foo'], 'bar; baz="spam"')
508        self.assertEqual(str(h),'foo: bar; baz="spam"\r\n\r\n')
509
510        h.add_header('Foo','bar',cheese=None)
511        self.assertEqual(h.get_all('foo'),
512            ['bar; baz="spam"', 'bar; cheese'])
513
514        self.assertEqual(str(h),
515            'foo: bar; baz="spam"\r\n'
516            'Foo: bar; cheese\r\n'
517            '\r\n'
518        )
519
520class ErrorHandler(BaseCGIHandler):
521    """Simple handler subclass for testing BaseHandler"""
522
523    # BaseHandler records the OS environment at import time, but envvars
524    # might have been changed later by other tests, which trips up
525    # HandlerTests.testEnviron().
526    os_environ = dict(os.environ.items())
527
528    def __init__(self,**kw):
529        setup_testing_defaults(kw)
530        BaseCGIHandler.__init__(
531            self, BytesIO(), BytesIO(), StringIO(), kw,
532            multithread=True, multiprocess=True
533        )
534
535class TestHandler(ErrorHandler):
536    """Simple handler subclass for testing BaseHandler, w/error passthru"""
537
538    def handle_error(self):
539        raise   # for testing, we want to see what's happening
540
541
542class HandlerTests(TestCase):
543
544    def checkEnvironAttrs(self, handler):
545        env = handler.environ
546        for attr in [
547            'version','multithread','multiprocess','run_once','file_wrapper'
548        ]:
549            if attr=='file_wrapper' and handler.wsgi_file_wrapper is None:
550                continue
551            self.assertEqual(getattr(handler,'wsgi_'+attr),env['wsgi.'+attr])
552
553    def checkOSEnviron(self,handler):
554        empty = {}; setup_testing_defaults(empty)
555        env = handler.environ
556        from os import environ
557        for k,v in environ.items():
558            if k not in empty:
559                self.assertEqual(env[k],v)
560        for k,v in empty.items():
561            self.assertIn(k, env)
562
563    def testEnviron(self):
564        h = TestHandler(X="Y")
565        h.setup_environ()
566        self.checkEnvironAttrs(h)
567        self.checkOSEnviron(h)
568        self.assertEqual(h.environ["X"],"Y")
569
570    def testCGIEnviron(self):
571        h = BaseCGIHandler(None,None,None,{})
572        h.setup_environ()
573        for key in 'wsgi.url_scheme', 'wsgi.input', 'wsgi.errors':
574            self.assertIn(key, h.environ)
575
576    def testScheme(self):
577        h=TestHandler(HTTPS="on"); h.setup_environ()
578        self.assertEqual(h.environ['wsgi.url_scheme'],'https')
579        h=TestHandler(); h.setup_environ()
580        self.assertEqual(h.environ['wsgi.url_scheme'],'http')
581
582    def testAbstractMethods(self):
583        h = BaseHandler()
584        for name in [
585            '_flush','get_stdin','get_stderr','add_cgi_vars'
586        ]:
587            self.assertRaises(NotImplementedError, getattr(h,name))
588        self.assertRaises(NotImplementedError, h._write, "test")
589
590    def testContentLength(self):
591        # Demo one reason iteration is better than write()...  ;)
592
593        def trivial_app1(e,s):
594            s('200 OK',[])
595            return [e['wsgi.url_scheme'].encode('iso-8859-1')]
596
597        def trivial_app2(e,s):
598            s('200 OK',[])(e['wsgi.url_scheme'].encode('iso-8859-1'))
599            return []
600
601        def trivial_app3(e,s):
602            s('200 OK',[])
603            return ['\u0442\u0435\u0441\u0442'.encode("utf-8")]
604
605        def trivial_app4(e,s):
606            # Simulate a response to a HEAD request
607            s('200 OK',[('Content-Length', '12345')])
608            return []
609
610        h = TestHandler()
611        h.run(trivial_app1)
612        self.assertEqual(h.stdout.getvalue(),
613            ("Status: 200 OK\r\n"
614            "Content-Length: 4\r\n"
615            "\r\n"
616            "http").encode("iso-8859-1"))
617
618        h = TestHandler()
619        h.run(trivial_app2)
620        self.assertEqual(h.stdout.getvalue(),
621            ("Status: 200 OK\r\n"
622            "\r\n"
623            "http").encode("iso-8859-1"))
624
625        h = TestHandler()
626        h.run(trivial_app3)
627        self.assertEqual(h.stdout.getvalue(),
628            b'Status: 200 OK\r\n'
629            b'Content-Length: 8\r\n'
630            b'\r\n'
631            b'\xd1\x82\xd0\xb5\xd1\x81\xd1\x82')
632
633        h = TestHandler()
634        h.run(trivial_app4)
635        self.assertEqual(h.stdout.getvalue(),
636            b'Status: 200 OK\r\n'
637            b'Content-Length: 12345\r\n'
638            b'\r\n')
639
640    def testBasicErrorOutput(self):
641
642        def non_error_app(e,s):
643            s('200 OK',[])
644            return []
645
646        def error_app(e,s):
647            raise AssertionError("This should be caught by handler")
648
649        h = ErrorHandler()
650        h.run(non_error_app)
651        self.assertEqual(h.stdout.getvalue(),
652            ("Status: 200 OK\r\n"
653            "Content-Length: 0\r\n"
654            "\r\n").encode("iso-8859-1"))
655        self.assertEqual(h.stderr.getvalue(),"")
656
657        h = ErrorHandler()
658        h.run(error_app)
659        self.assertEqual(h.stdout.getvalue(),
660            ("Status: %s\r\n"
661            "Content-Type: text/plain\r\n"
662            "Content-Length: %d\r\n"
663            "\r\n" % (h.error_status,len(h.error_body))).encode('iso-8859-1')
664            + h.error_body)
665
666        self.assertIn("AssertionError", h.stderr.getvalue())
667
668    def testErrorAfterOutput(self):
669        MSG = b"Some output has been sent"
670        def error_app(e,s):
671            s("200 OK",[])(MSG)
672            raise AssertionError("This should be caught by handler")
673
674        h = ErrorHandler()
675        h.run(error_app)
676        self.assertEqual(h.stdout.getvalue(),
677            ("Status: 200 OK\r\n"
678            "\r\n".encode("iso-8859-1")+MSG))
679        self.assertIn("AssertionError", h.stderr.getvalue())
680
681    def testHeaderFormats(self):
682
683        def non_error_app(e,s):
684            s('200 OK',[])
685            return []
686
687        stdpat = (
688            r"HTTP/%s 200 OK\r\n"
689            r"Date: \w{3}, [ 0123]\d \w{3} \d{4} \d\d:\d\d:\d\d GMT\r\n"
690            r"%s" r"Content-Length: 0\r\n" r"\r\n"
691        )
692        shortpat = (
693            "Status: 200 OK\r\n" "Content-Length: 0\r\n" "\r\n"
694        ).encode("iso-8859-1")
695
696        for ssw in "FooBar/1.0", None:
697            sw = ssw and "Server: %s\r\n" % ssw or ""
698
699            for version in "1.0", "1.1":
700                for proto in "HTTP/0.9", "HTTP/1.0", "HTTP/1.1":
701
702                    h = TestHandler(SERVER_PROTOCOL=proto)
703                    h.origin_server = False
704                    h.http_version = version
705                    h.server_software = ssw
706                    h.run(non_error_app)
707                    self.assertEqual(shortpat,h.stdout.getvalue())
708
709                    h = TestHandler(SERVER_PROTOCOL=proto)
710                    h.origin_server = True
711                    h.http_version = version
712                    h.server_software = ssw
713                    h.run(non_error_app)
714                    if proto=="HTTP/0.9":
715                        self.assertEqual(h.stdout.getvalue(),b"")
716                    else:
717                        self.assertTrue(
718                            re.match((stdpat%(version,sw)).encode("iso-8859-1"),
719                                h.stdout.getvalue()),
720                            ((stdpat%(version,sw)).encode("iso-8859-1"),
721                                h.stdout.getvalue())
722                        )
723
724    def testBytesData(self):
725        def app(e, s):
726            s("200 OK", [
727                ("Content-Type", "text/plain; charset=utf-8"),
728                ])
729            return [b"data"]
730
731        h = TestHandler()
732        h.run(app)
733        self.assertEqual(b"Status: 200 OK\r\n"
734            b"Content-Type: text/plain; charset=utf-8\r\n"
735            b"Content-Length: 4\r\n"
736            b"\r\n"
737            b"data",
738            h.stdout.getvalue())
739
740    def testCloseOnError(self):
741        side_effects = {'close_called': False}
742        MSG = b"Some output has been sent"
743        def error_app(e,s):
744            s("200 OK",[])(MSG)
745            class CrashyIterable(object):
746                def __iter__(self):
747                    while True:
748                        yield b'blah'
749                        raise AssertionError("This should be caught by handler")
750                def close(self):
751                    side_effects['close_called'] = True
752            return CrashyIterable()
753
754        h = ErrorHandler()
755        h.run(error_app)
756        self.assertEqual(side_effects['close_called'], True)
757
758    def testPartialWrite(self):
759        written = bytearray()
760
761        class PartialWriter:
762            def write(self, b):
763                partial = b[:7]
764                written.extend(partial)
765                return len(partial)
766
767            def flush(self):
768                pass
769
770        environ = {"SERVER_PROTOCOL": "HTTP/1.0"}
771        h = SimpleHandler(BytesIO(), PartialWriter(), sys.stderr, environ)
772        msg = "should not do partial writes"
773        with self.assertWarnsRegex(DeprecationWarning, msg):
774            h.run(hello_app)
775        self.assertEqual(b"HTTP/1.0 200 OK\r\n"
776            b"Content-Type: text/plain\r\n"
777            b"Date: Mon, 05 Jun 2006 18:49:54 GMT\r\n"
778            b"Content-Length: 13\r\n"
779            b"\r\n"
780            b"Hello, world!",
781            written)
782
783
784if __name__ == "__main__":
785    unittest.main()
786