Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

Merged revisions 80526 via svnmerge from 

svn+ssh://pythondev@svn.python.org/python/trunk

........
  r80526 | antoine.pitrou | 2010-04-27 11:51:18 +0200 (mar., 27 avril 2010) | 5 lines

  replace custom code with standard library functionality (HTTPServer.shutdown())
  + enable test that was never run (!)
  + make tests faster by lowering polling timeout
........
This commit is contained in:
Antoine Pitrou 2010-04-27 09:54:14 +00:00
parent 6535b31ced
commit 77a1c36fb7
1 changed files with 7 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
Lib/test/test_ssl.py
View File

@ -37,7 +37,11 @@ def handle_error(prefix):
if test_support.verbose:
sys.stdout.write(prefix + exc_format)
class BasicTests(unittest.TestCase):
def testSimpleSSLwrap(self):
# A crude test for the legacy API
try:
ssl.sslwrap_simple(socket.socket(socket.AF_INET))
except IOError, e:
@ -53,8 +57,6 @@ def handle_error(prefix):
else:
raise
class BasicTests(unittest.TestCase):
def testSSLconnect(self):
if not test_support.is_resource_enabled('network'):
return
@ -421,7 +423,7 @@ else:
threading.Thread.start(self)
def run (self):
self.sock.settimeout(0.5)
self.sock.settimeout(0.05)
self.sock.listen(5)
self.active = True
if self.flag:
@ -568,49 +570,6 @@ else:
certfile=self.certfile)
return sslconn, addr
# The methods overridden below this are mainly so that we
# can run it in a thread and be able to stop it from another
# You probably wouldn't need them in other uses.
def server_activate(self):
# We want to run this in a thread for testing purposes,
# so we override this to set timeout, so that we get
# a chance to stop the server
self.socket.settimeout(0.5)
HTTPServer.server_activate(self)
def serve_forever(self):
# We want this to run in a thread, so we use a slightly
# modified version of "forever".
self.active = True
while 1:
try:
# We need to lock while handling the request.
# Another thread can close the socket after self.active
# has been checked and before the request is handled.
# This causes an exception when using the closed socket.
with self.active_lock:
if not self.active:
break
self.handle_request()
except socket.timeout:
pass
except KeyboardInterrupt:
self.server_close()
return
except:
sys.stdout.write(''.join(traceback.format_exception(*sys.exc_info())))
break
time.sleep(0.1)
def server_close(self):
# Again, we want this to run in a thread, so we need to override
# close to clear the "active" flag, so that serve_forever() will
# terminate.
with self.active_lock:
HTTPServer.server_close(self)
self.active = False
class RootedHTTPRequestHandler(SimpleHTTPRequestHandler):
# need to override translate_path to get a known root,
@ -676,12 +635,12 @@ else:
self.active = True
if self.flag:
self.flag.set()
self.server.serve_forever()
self.server.serve_forever(0.05)
self.active = False
def stop (self):
self.active = False
self.server.server_close()
self.server.shutdown()
def badCertTest (certfile):