[Python-checkins] cpython: Added SSL test for HTTPHandler.

[vinay.sajip]

http://hg.python.org/cpython/rev/961650dd9c91
changeset:   70244:961650dd9c91
user:        Vinay Sajip <vinay_sajip at yahoo.co.uk>
date:        Sat May 21 11:32:15 2011 +0100
summary:
  Added SSL test for HTTPHandler.

files:
  Lib/test/test_logging.py |  105 ++++++++++++++++++--------
  1 files changed, 72 insertions(+), 33 deletions(-)


diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
--- a/Lib/test/test_logging.py
+++ b/Lib/test/test_logging.py
@@ -1467,34 +1467,44 @@
 
 @unittest.skipUnless(threading, 'Threading required for this test.')
 class HTTPHandlerTest(BaseTest):
-
     """Test for HTTPHandler."""
 
+    PEMFILE = """-----BEGIN RSA PRIVATE KEY-----
+MIICXQIBAAKBgQDGT4xS5r91rbLJQK2nUDenBhBG6qFk+bVOjuAGC/LSHlAoBnvG
+zQG3agOG+e7c5z2XT8m2ktORLqG3E4mYmbxgyhDrzP6ei2Anc+pszmnxPoK3Puh5
+aXV+XKt0bU0C1m2+ACmGGJ0t3P408art82nOxBw8ZHgIg9Dtp6xIUCyOqwIDAQAB
+AoGBAJFTnFboaKh5eUrIzjmNrKsG44jEyy+vWvHN/FgSC4l103HxhmWiuL5Lv3f7
+0tMp1tX7D6xvHwIG9VWvyKb/Cq9rJsDibmDVIOslnOWeQhG+XwJyitR0pq/KlJIB
+5LjORcBw795oKWOAi6RcOb1ON59tysEFYhAGQO9k6VL621gRAkEA/Gb+YXULLpbs
+piXN3q4zcHzeaVANo69tUZ6TjaQqMeTxE4tOYM0G0ZoSeHEdaP59AOZGKXXNGSQy
+2z/MddcYGQJBAMkjLSYIpOLJY11ja8OwwswFG2hEzHe0cS9bzo++R/jc1bHA5R0Y
+i6vA5iPi+wopPFvpytdBol7UuEBe5xZrxWMCQQCWxELRHiP2yWpEeLJ3gGDzoXMN
+PydWjhRju7Bx3AzkTtf+D6lawz1+eGTuEss5i0JKBkMEwvwnN2s1ce+EuF4JAkBb
+E96h1lAzkVW5OAfYOPY8RCPA90ZO/hoyg7PpSxR0ECuDrgERR8gXIeYUYfejBkEa
+rab4CfRoVJKKM28Yq/xZAkBvuq670JRCwOgfUTdww7WpdOQBYPkzQccsKNCslQW8
+/DyW6y06oQusSENUvynT6dr3LJxt/NgZPhZX2+k1eYDV
+-----END RSA PRIVATE KEY-----
+-----BEGIN CERTIFICATE-----
+MIICGzCCAYSgAwIBAgIJAIq84a2Q/OvlMA0GCSqGSIb3DQEBBQUAMBQxEjAQBgNV
+BAMTCWxvY2FsaG9zdDAeFw0xMTA1MjExMDIzMzNaFw03NTAzMjEwMzU1MTdaMBQx
+EjAQBgNVBAMTCWxvY2FsaG9zdDCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEA
+xk+MUua/da2yyUCtp1A3pwYQRuqhZPm1To7gBgvy0h5QKAZ7xs0Bt2oDhvnu3Oc9
+l0/JtpLTkS6htxOJmJm8YMoQ68z+notgJ3PqbM5p8T6Ctz7oeWl1flyrdG1NAtZt
+vgAphhidLdz+NPGq7fNpzsQcPGR4CIPQ7aesSFAsjqsCAwEAAaN1MHMwHQYDVR0O
+BBYEFLWaUPO6N7efGiuoS9i3DVYcUwn0MEQGA1UdIwQ9MDuAFLWaUPO6N7efGiuo
+S9i3DVYcUwn0oRikFjAUMRIwEAYDVQQDEwlsb2NhbGhvc3SCCQCKvOGtkPzr5TAM
+BgNVHRMEBTADAQH/MA0GCSqGSIb3DQEBBQUAA4GBAMK5whPjLNQK1Ivvk88oqJqq
+4f889OwikGP0eUhOBhbFlsZs+jq5YZC2UzHz+evzKBlgAP1u4lP/cB85CnjvWqM+
+1c/lywFHQ6HOdDeQ1L72tSYMrNOG4XNmLn0h7rx6GoTU7dcFRfseahBCq8mv0IDt
+IRbTpvlHWPjsSvHz0ZOH
+-----END CERTIFICATE-----"""
+
     def setUp(self):
         """Set up an HTTP server to receive log messages, and a HTTPHandler
         pointing to that server's address and port."""
         BaseTest.setUp(self)
-        addr = ('localhost', 0)
-        self.server = server = TestHTTPServer(addr, self.handle_request,
-                                                0.01)
-        server.start()
-        server.ready.wait()
-        host = 'localhost:%d' % server.server_port
-        self.h_hdlr = logging.handlers.HTTPHandler(host, '/frob')
-        self.log_data = None
-        self.root_logger.removeHandler(self.root_logger.handlers[0])
-        self.root_logger.addHandler(self.h_hdlr)
         self.handled = threading.Event()
 
-    def tearDown(self):
-        """Shutdown the UDP server."""
-        try:
-            self.server.stop(2.0)
-            self.root_logger.removeHandler(self.h_hdlr)
-            self.h_hdlr.close()
-        finally:
-            BaseTest.tearDown(self)
-
     def handle_request(self, request):
         self.command = request.command
         self.log_data = urlparse(request.path)
@@ -1511,20 +1521,49 @@
     def test_output(self):
         # The log message sent to the HTTPHandler is properly received.
         logger = logging.getLogger("http")
-        for method in ('GET', 'POST'):
-            self.h_hdlr.method = method
-            msg = "sp\xe4m"
-            logger.error(msg)
-            self.handled.wait()
-            self.assertEqual(self.log_data.path, '/frob')
-            self.assertEqual(self.command, method)
-            if method == 'GET':
-                d = parse_qs(self.log_data.query)
+        root_logger = self.root_logger
+        root_logger.removeHandler(self.root_logger.handlers[0])
+        for secure in (False, True):
+            addr = ('localhost', 0)
+            if secure:
+                import ssl
+                fd, fn = tempfile.mkstemp()
+                os.close(fd)
+                with open(fn, 'w') as f:
+                    f.write(self.PEMFILE)
+                sslctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+                sslctx.load_cert_chain(fn)
+                os.unlink(fn)
             else:
-                d = parse_qs(self.post_data.decode('utf-8'))
-            self.assertEqual(d['name'], ['http'])
-            self.assertEqual(d['funcName'], ['test_output'])
-            self.assertEqual(d['msg'], [msg])
+                sslctx = None
+            self.server = server = TestHTTPServer(addr, self.handle_request,
+                                                    0.01, sslctx=sslctx)
+            server.start()
+            server.ready.wait()
+            host = 'localhost:%d' % server.server_port
+            self.h_hdlr = logging.handlers.HTTPHandler(host, '/frob', secure=secure)
+            self.log_data = None
+            root_logger.addHandler(self.h_hdlr)
+
+            for method in ('GET', 'POST'):
+                self.h_hdlr.method = method
+                self.handled.clear()
+                msg = "sp\xe4m"
+                logger.error(msg)
+                self.handled.wait()
+                self.assertEqual(self.log_data.path, '/frob')
+                self.assertEqual(self.command, method)
+                if method == 'GET':
+                    d = parse_qs(self.log_data.query)
+                else:
+                    d = parse_qs(self.post_data.decode('utf-8'))
+                self.assertEqual(d['name'], ['http'])
+                self.assertEqual(d['funcName'], ['test_output'])
+                self.assertEqual(d['msg'], [msg])
+
+            self.server.stop(2.0)
+            self.root_logger.removeHandler(self.h_hdlr)
+            self.h_hdlr.close()
 
 class MemoryTest(BaseTest):
 

-- 
Repository URL: http://hg.python.org/cpython