Now using unittest for the tests infraestructure. Also split the
tests in those who need the network, and that who doesn't.
This commit is contained in:
Facundo Batista
parent
b28ef9b429
commit
c3a35e19cc
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Reference in New Issue