mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-11 14:44:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Remove uses of find_unused_port() in test_ssl, and small cleanups

Browse source
This commit is contained in:
Antoine Pitrou 2010-04-27 08:40:51 +00:00
parent 47b5440f43
commit 150acda57f
1 changed files with 17 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

32
Lib/test/test_ssl.py
View file

@ -696,9 +696,9 @@ else:
self.flag = None self.flag = None
self.active = False self.active = False
self.RootedHTTPRequestHandler.root = os.path.split(CERTFILE)[0] self.RootedHTTPRequestHandler.root = os.path.split(CERTFILE)[0]
self.port = test_support.find_unused_port()
self.server = self.HTTPSServer( self.server = self.HTTPSServer(
(HOST, self.port), self.RootedHTTPRequestHandler, certfile) (HOST, 0), self.RootedHTTPRequestHandler, certfile)
self.port = self.server.server_port
threading.Thread.__init__(self) threading.Thread.__init__(self)
self.daemon = True self.daemon = True
@ -851,28 +851,28 @@ else:
listener_ready = threading.Event() listener_ready = threading.Event()
listener_gone = 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.
# Then it rudely closes the socket, and sets Event `listener_gone`
# to let the main thread know the socket is gone.
def listener():
s = socket.socket() s = socket.socket()
s.bind((HOST, port)) port = test_support.bind_port(s, HOST)
# `listener` runs in a thread. It sits in an accept() until
# the main thread connects. Then it rudely closes the socket,
# and sets Event `listener_gone` to let the main thread know
# the socket is gone.
def listener():
s.listen(5) s.listen(5)
listener_ready.set() listener_ready.set()
s.accept() s.accept()
s = None # reclaim the socket object, which also closes it s.close()
listener_gone.set() listener_gone.set()
def connector(): def connector():
listener_ready.wait() listener_ready.wait()
s = socket.socket() c = socket.socket()
s.connect((HOST, port)) c.connect((HOST, port))
listener_gone.wait() listener_gone.wait()
try: try:
ssl_sock = ssl.wrap_socket(s) ssl_sock = ssl.wrap_socket(c)
except IOError: except IOError:
pass pass
else: else:
@ -881,7 +881,9 @@ else:
t = threading.Thread(target=listener) t = threading.Thread(target=listener)
t.start() t.start()
try:
connector() connector()
finally:
t.join() t.join()
def testEcho (self): def testEcho (self):
@ -1354,10 +1356,6 @@ def test_main(verbose=False):
not os.path.exists(SVN_PYTHON_ORG_ROOT_CERT)): not os.path.exists(SVN_PYTHON_ORG_ROOT_CERT)):
raise test_support.TestFailed("Can't read certificate files!") raise test_support.TestFailed("Can't read certificate files!")
TESTPORT = test_support.find_unused_port()
if not TESTPORT:
raise test_support.TestFailed("Can't find open port to test servers on!")
tests = [BasicTests] tests = [BasicTests]
if test_support.is_resource_enabled('network'): if test_support.is_resource_enabled('network'):