From c3a35e19cc1e3de5ae851600d3534fb98c999351 Mon Sep 17 00:00:00 2001 From: Facundo Batista Date: Tue, 3 Apr 2007 14:05:08 +0000 Subject: [PATCH] Now using unittest for the tests infraestructure. Also split the tests in those who need the network, and that who doesn't. --- Doc/lib/libpoplib.tex | 5 +- Lib/test/test_socket_ssl.py | 222 ++++++++++++++++++------------------ 2 files changed, 116 insertions(+), 111 deletions(-) diff --git a/Doc/lib/libpoplib.tex b/Doc/lib/libpoplib.tex index 16f3946b2b3..7b2c4a10eb1 100644 --- a/Doc/lib/libpoplib.tex +++ b/Doc/lib/libpoplib.tex @@ -50,8 +50,9 @@ certificate chain file for the SSL connection. One exception is defined as an attribute of the \module{poplib} module: \begin{excdesc}{error_proto} -Exception raised on any errors. The reason for the exception is -passed to the constructor as a string. +Exception raised on any errors from this module (errors from +\module{socket} module are not caught). The reason for the exception +is passed to the constructor as a string. \end{excdesc} \begin{seealso} diff --git a/Lib/test/test_socket_ssl.py b/Lib/test/test_socket_ssl.py index 1f2b9d10ebb..4c502d1d248 100644 --- a/Lib/test/test_socket_ssl.py +++ b/Lib/test/test_socket_ssl.py @@ -1,129 +1,133 @@ # Test just the SSL support in the socket module, in a moderately bogus way. import sys +import unittest from test import test_support import socket import errno +import threading +import subprocess +import time -# Optionally test SSL support. This requires the 'network' resource as given -# on the regrtest command line. -skip_expected = not (test_support.is_resource_enabled('network') and - hasattr(socket, "ssl")) +# Optionally test SSL support, if we have it in the tested platform +skip_expected = not hasattr(socket, "ssl") -def test_basic(): - test_support.requires('network') +class ConnectedTests(unittest.TestCase): - import urllib - - if test_support.verbose: - print "test_basic ..." - - socket.RAND_status() - try: - socket.RAND_egd(1) - except TypeError: - pass - else: - print "didn't raise TypeError" - socket.RAND_add("this is a random string", 75.0) - - with test_support.transient_internet(): - f = urllib.urlopen('https://sf.net') - buf = f.read() - f.close() - -def test_timeout(): - test_support.requires('network') - - def error_msg(extra_msg): - print >> sys.stderr, """\ - WARNING: an attempt to connect to %r %s, in - test_timeout. That may be legitimate, but is not the outcome we hoped - for. If this message is seen often, test_timeout should be changed to - use a more reliable address.""" % (ADDR, extra_msg) - - if test_support.verbose: - print "test_timeout ..." - - # A service which issues a welcome banner (without need to write - # anything). - # XXX ("gmail.org", 995) has been unreliable so far, from time to time - # XXX non-responsive for hours on end (& across all buildbot slaves, - # XXX so that's not just a local thing). - ADDR = "gmail.org", 995 - - s = socket.socket() - s.settimeout(30.0) - try: - s.connect(ADDR) - except socket.timeout: - error_msg('timed out') - return - except socket.error, exc: # In case connection is refused. - if exc.args[0] == errno.ECONNREFUSED: - error_msg('was refused') - return - else: - raise - - ss = socket.ssl(s) - # Read part of return welcome banner twice. - ss.read(1) - ss.read(1) - s.close() - -def test_rude_shutdown(): - if test_support.verbose: - print "test_rude_shutdown ..." - - try: - import threading - except ImportError: - return - - # Some random port to connect to. - PORT = [9934] - - listener_ready = threading.Event() - listener_gone = threading.Event() - - # `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() - PORT[0] = test_support.bind_port(s, '', PORT[0]) - s.listen(5) - listener_ready.set() - s.accept() - s = None # reclaim the socket object, which also closes it - listener_gone.set() - - def connector(): - listener_ready.wait() - s = socket.socket() - s.connect(('localhost', PORT[0])) - listener_gone.wait() + def testBasic(self): + import urllib + + if test_support.verbose: + print "test_basic ..." + + socket.RAND_status() try: - ssl_sock = socket.ssl(s) - except socket.sslerror: + socket.RAND_egd(1) + except TypeError: pass else: - raise test_support.TestFailed( - 'connecting to closed SSL socket should have failed') + print "didn't raise TypeError" + socket.RAND_add("this is a random string", 75.0) + + with test_support.transient_internet(): + f = urllib.urlopen('https://sf.net') + buf = f.read() + f.close() + + def testTimeout(self): + def error_msg(extra_msg): + print >> sys.stderr, """\ + WARNING: an attempt to connect to %r %s, in + test_timeout. That may be legitimate, but is not the outcome we + hoped for. If this message is seen often, test_timeout should be + changed to use a more reliable address.""" % (ADDR, extra_msg) + + if test_support.verbose: + print "test_timeout ..." + + # A service which issues a welcome banner (without need to write + # anything). + # XXX ("gmail.org", 995) has been unreliable so far, from time to + # XXX time non-responsive for hours on end (& across all buildbot + # XXX slaves, so that's not just a local thing). + ADDR = "gmail.org", 995 + + s = socket.socket() + s.settimeout(30.0) + try: + s.connect(ADDR) + except socket.timeout: + error_msg('timed out') + return + except socket.error, exc: # In case connection is refused. + if exc.args[0] == errno.ECONNREFUSED: + error_msg('was refused') + return + else: + raise + + ss = socket.ssl(s) + # Read part of return welcome banner twice. + ss.read(1) + ss.read(1) + s.close() + +class BasicTests(unittest.TestCase): + + def testRudeShutdown(self): + if test_support.verbose: + print "test_rude_shutdown ..." + + # Some random port to connect to. + PORT = [9934] + + listener_ready = threading.Event() + listener_gone = threading.Event() + + # `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() + PORT[0] = test_support.bind_port(s, '', PORT[0]) + s.listen(5) + listener_ready.set() + s.accept() + s = None # reclaim the socket object, which also closes it + listener_gone.set() + + def connector(): + listener_ready.wait() + s = socket.socket() + s.connect(('localhost', PORT[0])) + listener_gone.wait() + try: + ssl_sock = socket.ssl(s) + except socket.sslerror: + pass + else: + raise test_support.TestFailed( + 'connecting to closed SSL socket should have failed') + + t = threading.Thread(target=listener) + t.start() + connector() + t.join() - t = threading.Thread(target=listener) - t.start() - connector() - t.join() def test_main(): if not hasattr(socket, "ssl"): raise test_support.TestSkipped("socket module has no ssl support") - test_rude_shutdown() - test_basic() - test_timeout() + + tests = [BasicTests] + + if test_support.is_resource_enabled('network'): + tests.append(ConnectedTests) + + thread_info = test_support.threading_setup() + test_support.run_unittest(*tests) + test_support.threading_cleanup(*thread_info) if __name__ == "__main__": test_main()