summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_ssl.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_ssl.py')
-rw-r--r--Lib/test/test_ssl.py70
1 files changed, 21 insertions, 49 deletions
diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py
index 8f8212a..eb4d00c 100644
--- a/Lib/test/test_ssl.py
+++ b/Lib/test/test_ssl.py
@@ -23,11 +23,10 @@ try:
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:
@@ -269,7 +268,7 @@ else:
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:
@@ -285,12 +284,8 @@ else:
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)
@@ -434,12 +429,13 @@ else:
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)
@@ -465,7 +461,7 @@ else:
def badCertTest (certfile):
- server = ThreadedEchoServer(TESTPORT, CERTFILE,
+ server = ThreadedEchoServer(CERTFILE,
certreqs=ssl.CERT_REQUIRED,
cacerts=CERTFILE, chatty=False)
flag = threading.Event()
@@ -478,7 +474,7 @@ else:
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, x:
if test_support.verbose:
sys.stdout.write("\nSSLError is %s\n" % x[1])
@@ -493,7 +489,7 @@ else:
client_certfile, client_protocol=None, indata="FOO\n",
chatty=True, connectionchatty=False):
- server = ThreadedEchoServer(TESTPORT, certfile,
+ server = ThreadedEchoServer(certfile,
certreqs=certreqs,
ssl_version=protocol,
cacerts=cacertsfile,
@@ -513,7 +509,7 @@ else:
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, x:
raise test_support.TestFailed("Unexpected SSL error: " + str(x))
except Exception, x:
@@ -582,6 +578,7 @@ else:
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.
@@ -589,11 +586,7 @@ else:
# 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()
@@ -603,7 +596,7 @@ else:
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)
@@ -631,7 +624,7 @@ else:
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,
@@ -648,7 +641,7 @@ else:
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, x:
raise test_support.TestFailed(
"Unexpected SSL error: " + str(x))
@@ -748,7 +741,7 @@ else:
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,
@@ -763,7 +756,7 @@ else:
try:
s = socket.socket()
s.setblocking(1)
- s.connect(('127.0.0.1', TESTPORT))
+ s.connect((HOST, server.port))
except Exception, x:
raise test_support.TestFailed("Unexpected exception: " + str(x))
else:
@@ -805,7 +798,7 @@ else:
def testAsyncore(self):
- server = AsyncoreHTTPSServer(TESTPORT, CERTFILE)
+ server = AsyncoreHTTPSServer(CERTFILE)
flag = threading.Event()
server.start(flag)
# wait for it to start
@@ -817,8 +810,8 @@ else:
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):
@@ -842,29 +835,11 @@ else:
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(
@@ -874,9 +849,6 @@ def test_main(verbose=False):
if (not os.path.exists(CERTFILE) or
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]
generated by cgit v0.12 at 2025-04-27 13:12:59 (GMT)