[Python-3000-checkins] r62275 - python/branches/py3k/Lib/test/test_ssl.py
trent.nelson
python-3000-checkins at python.org
Thu Apr 10 22:54:36 CEST 2008
Author: trent.nelson
Date: Thu Apr 10 22:54:35 2008
New Revision: 62275
Modified:
python/branches/py3k/Lib/test/test_ssl.py
Log:
Update test_ssl.py to reflect the new approach for writing network-oriented tests in such a way that contention for ports is eliminated, thus allowing multiple instances of the same test to be run in parallel without issue (import for buildbots to run smoothly). This change is in line with the changes made to network-oriented tests in trunk in r62234. test_ssl.py is an odd case because the py3k version was newer than trunk. When the most recent svnmerge was run from trunk to py3k in r62242, the older trunk version was brought over on top of the newer py3k version. So, the plan was/is: revert the recent svnmerged version (done), manually make the changes to improve running in parallel (done with this commit), copy this version back to trunk and manually 3to2 it (to do next), check that in, then svnmerge block it. Then when future changes need to be made to test_ssl.py, we can make them against trunk, and hopefully the merge to py3k should be a lot less painful.
Modified: python/branches/py3k/Lib/test/test_ssl.py
==============================================================================
--- python/branches/py3k/Lib/test/test_ssl.py (original)
+++ python/branches/py3k/Lib/test/test_ssl.py Thu Apr 10 22:54:35 2008
@@ -25,11 +25,10 @@
except ImportError:
skip_expected = True
+HOST = test_support.HOST
CERTFILE = None
SVN_PYTHON_ORG_ROOT_CERT = None
-TESTPORT = 10025
-
def handle_error(prefix):
exc_format = ' '.join(traceback.format_exception(*sys.exc_info()))
if test_support.verbose:
@@ -299,7 +298,7 @@
except:
handle_error('')
- def __init__(self, port, certificate, ssl_version=None,
+ def __init__(self, certificate, ssl_version=None,
certreqs=None, cacerts=None, expect_bad_connects=False,
chatty=True, connectionchatty=False, starttls_server=False):
if ssl_version is None:
@@ -315,12 +314,8 @@
self.connectionchatty = connectionchatty
self.starttls_server = starttls_server
self.sock = socket.socket()
+ self.port = test_support.bind_port(self.sock)
self.flag = None
- if hasattr(socket, 'SO_REUSEADDR'):
- self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- if hasattr(socket, 'SO_REUSEPORT'):
- self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
- self.sock.bind(('127.0.0.1', port))
self.active = False
threading.Thread.__init__(self)
self.setDaemon(False)
@@ -471,12 +466,13 @@
format%args))
- def __init__(self, port, certfile):
+ def __init__(self, certfile):
self.flag = None
self.active = False
self.RootedHTTPRequestHandler.root = os.path.split(CERTFILE)[0]
+ self.port = test_support.find_unused_port()
self.server = self.HTTPSServer(
- ('', port), self.RootedHTTPRequestHandler, certfile)
+ (HOST, self.port), self.RootedHTTPRequestHandler, certfile)
threading.Thread.__init__(self)
self.setDaemon(True)
@@ -557,10 +553,11 @@
def handle_error(self):
raise
- def __init__(self, port, certfile):
+ def __init__(self, certfile):
self.flag = None
self.active = False
- self.server = self.EchoServer(port, certfile)
+ self.port = test_support.find_unused_port()
+ self.server = self.EchoServer(self.port, certfile)
threading.Thread.__init__(self)
self.setDaemon(True)
@@ -586,7 +583,7 @@
self.server.close()
def badCertTest (certfile):
- server = ThreadedEchoServer(TESTPORT, CERTFILE,
+ server = ThreadedEchoServer(CERTFILE,
certreqs=ssl.CERT_REQUIRED,
cacerts=CERTFILE, chatty=False,
connectionchatty=False)
@@ -600,7 +597,7 @@
s = ssl.wrap_socket(socket.socket(),
certfile=certfile,
ssl_version=ssl.PROTOCOL_TLSv1)
- s.connect(('127.0.0.1', TESTPORT))
+ s.connect((HOST, server.port))
except ssl.SSLError as x:
if test_support.verbose:
sys.stdout.write("\nSSLError is %s\n" % x)
@@ -616,7 +613,7 @@
indata="FOO\n",
chatty=False, connectionchatty=False):
- server = ThreadedEchoServer(TESTPORT, certfile,
+ server = ThreadedEchoServer(certfile,
certreqs=certreqs,
ssl_version=protocol,
cacerts=cacertsfile,
@@ -636,7 +633,7 @@
ca_certs=cacertsfile,
cert_reqs=certreqs,
ssl_version=client_protocol)
- s.connect(('127.0.0.1', TESTPORT))
+ s.connect((HOST, server.port))
except ssl.SSLError as x:
raise test_support.TestFailed("Unexpected SSL error: " + str(x))
except Exception as x:
@@ -716,7 +713,7 @@
if test_support.verbose:
sys.stdout.write("\n")
s2 = socket.socket()
- server = ThreadedEchoServer(TESTPORT, CERTFILE,
+ server = ThreadedEchoServer(CERTFILE,
certreqs=ssl.CERT_NONE,
ssl_version=ssl.PROTOCOL_SSLv23,
cacerts=CERTFILE,
@@ -733,7 +730,7 @@
ca_certs=CERTFILE,
cert_reqs=ssl.CERT_REQUIRED,
ssl_version=ssl.PROTOCOL_SSLv23)
- s.connect(('127.0.0.1', TESTPORT))
+ s.connect((HOST, server.port))
except ssl.SSLError as x:
raise test_support.TestFailed(
"Unexpected SSL error: " + str(x))
@@ -780,6 +777,7 @@
listener_ready = threading.Event()
listener_gone = threading.Event()
+ port = test_support.find_unused_port()
# `listener` runs in a thread. It opens a socket listening on
# PORT, and sits in an accept() until the main thread connects.
@@ -787,11 +785,7 @@
# to let the main thread know the socket is gone.
def listener():
s = socket.socket()
- if hasattr(socket, 'SO_REUSEADDR'):
- s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- if hasattr(socket, 'SO_REUSEPORT'):
- s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
- s.bind(('127.0.0.1', TESTPORT))
+ s.bind((HOST, port))
s.listen(5)
listener_ready.set()
s.accept()
@@ -801,7 +795,7 @@
def connector():
listener_ready.wait()
s = socket.socket()
- s.connect(('127.0.0.1', TESTPORT))
+ s.connect((HOST, port))
listener_gone.wait()
try:
ssl_sock = ssl.wrap_socket(s)
@@ -873,7 +867,7 @@
msgs = ("msg 1", "MSG 2", "STARTTLS", "MSG 3", "msg 4")
- server = ThreadedEchoServer(TESTPORT, CERTFILE,
+ server = ThreadedEchoServer(CERTFILE,
ssl_version=ssl.PROTOCOL_TLSv1,
starttls_server=True,
chatty=True,
@@ -888,7 +882,7 @@
try:
s = socket.socket()
s.setblocking(1)
- s.connect(('127.0.0.1', TESTPORT))
+ s.connect((HOST, server.port))
except Exception as x:
raise test_support.TestFailed("Unexpected exception: " + str(x))
else:
@@ -936,7 +930,7 @@
def testSocketServer(self):
- server = OurHTTPSServer(TESTPORT, CERTFILE)
+ server = OurHTTPSServer(CERTFILE)
flag = threading.Event()
server.start(flag)
# wait for it to start
@@ -948,8 +942,8 @@
d1 = open(CERTFILE, 'rb').read()
d2 = ''
# now fetch the same data from the HTTPS server
- url = 'https://127.0.0.1:%d/%s' % (
- TESTPORT, os.path.split(CERTFILE)[1])
+ url = 'https://%s:%d/%s' % (
+ HOST, server.port, os.path.split(CERTFILE)[1])
f = urllib.urlopen(url)
dlen = f.info().getheader("content-length")
if dlen and (int(dlen) > 0):
@@ -984,7 +978,7 @@
sys.stdout.write("\n")
indata="FOO\n"
- server = AsyncoreEchoServer(TESTPORT, CERTFILE)
+ server = AsyncoreEchoServer(CERTFILE)
flag = threading.Event()
server.start(flag)
# wait for it to start
@@ -992,7 +986,7 @@
# try to connect
try:
s = ssl.wrap_socket(socket.socket())
- s.connect(('127.0.0.1', TESTPORT))
+ s.connect((HOST, server.port))
except ssl.SSLError as x:
raise test_support.TestFailed("Unexpected SSL error: " + str(x))
except Exception as x:
@@ -1019,30 +1013,11 @@
server.stop()
server.join()
-
-def findtestsocket(start, end):
- def testbind(i):
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- try:
- s.bind(("127.0.0.1", i))
- except:
- return 0
- else:
- return 1
- finally:
- s.close()
-
- for i in range(start, end):
- if testbind(i) and testbind(i+1):
- return i
- return 0
-
-
def test_main(verbose=False):
if skip_expected:
raise test_support.TestSkipped("No SSL support")
- global CERTFILE, TESTPORT, SVN_PYTHON_ORG_ROOT_CERT
+ global CERTFILE, SVN_PYTHON_ORG_ROOT_CERT
CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir,
"keycert.pem")
SVN_PYTHON_ORG_ROOT_CERT = os.path.join(
@@ -1053,10 +1028,6 @@
not os.path.exists(SVN_PYTHON_ORG_ROOT_CERT)):
raise test_support.TestFailed("Can't read certificate files!")
- TESTPORT = findtestsocket(10025, 12000)
- if not TESTPORT:
- raise test_support.TestFailed("Can't find open port to test servers on!")
-
tests = [BasicTests]
if test_support.is_resource_enabled('network'):
More information about the Python-3000-checkins
mailing list