Now using unittest for the tests infraestructure. Also split the

tests in those who need the network, and that who doesn't.

2 files changed, 118 insertions, 113 deletions

diff --git a/Doc/lib/libpoplib.tex b/Doc/lib/libpoplib.tex
index 16f3946..7b2c4a1 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 1f2b9d1..4c502d1 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
-
-# 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"))
-
-def test_basic():
-    test_support.requires('network')
-
-    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()
+import threading
+import subprocess
+import time
+
+# Optionally test SSL support, if we have it in the tested platform
+skip_expected = not hasattr(socket, "ssl")
+
+class ConnectedTests(unittest.TestCase):
+
+    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(
+            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()