mirror of
https://github.com/python/cpython.git
synced 2025-08-04 08:59:19 +00:00
Added SSL test for HTTPHandler.
This commit is contained in:
parent
91f0e346cc
commit
9ba876104e
1 changed files with 72 additions and 33 deletions
|
@ -1467,34 +1467,44 @@ class SysLogHandlerTest(BaseTest):
|
|||
|
||||
@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 @@ class HTTPHandlerTest(BaseTest):
|
|||
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):
|
||||
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue