Fix the socket tests so they can be run concurrently. Backport candidate
[python.git] / Lib / test / test_socket_ssl.py
blob074b6279fed6c7a5118ba29bb8871f1a36cb7f24
1 # Test just the SSL support in the socket module, in a moderately bogus way.
3 import sys
4 from test import test_support
5 import socket
7 # Optionally test SSL support. This requires the 'network' resource as given
8 # on the regrtest command line.
9 skip_expected = not (test_support.is_resource_enabled('network') and
10 hasattr(socket, "ssl"))
12 def test_basic():
13 test_support.requires('network')
15 import urllib
17 if test_support.verbose:
18 print "test_basic ..."
20 socket.RAND_status()
21 try:
22 socket.RAND_egd(1)
23 except TypeError:
24 pass
25 else:
26 print "didn't raise TypeError"
27 socket.RAND_add("this is a random string", 75.0)
29 f = urllib.urlopen('https://sf.net')
30 buf = f.read()
31 f.close()
33 def test_timeout():
34 test_support.requires('network')
36 if test_support.verbose:
37 print "test_timeout ..."
39 # A service which issues a welcome banner (without need to write
40 # anything).
41 # XXX ("gmail.org", 995) has been unreliable so far, from time to time
42 # XXX non-responsive for hours on end (& across all buildbot slaves,
43 # XXX so that's not just a local thing).
44 ADDR = "gmail.org", 995
46 s = socket.socket()
47 s.settimeout(30.0)
48 try:
49 s.connect(ADDR)
50 except socket.timeout:
51 print >> sys.stderr, """\
52 WARNING: an attempt to connect to %r timed out, in
53 test_timeout. That may be legitimate, but is not the outcome we hoped
54 for. If this message is seen often, test_timeout should be changed to
55 use a more reliable address.""" % (ADDR,)
56 return
58 ss = socket.ssl(s)
59 # Read part of return welcome banner twice.
60 ss.read(1)
61 ss.read(1)
62 s.close()
64 def test_rude_shutdown():
65 if test_support.verbose:
66 print "test_rude_shutdown ..."
68 try:
69 import threading
70 except ImportError:
71 return
73 # Some random port to connect to.
74 PORT = [9934]
76 listener_ready = threading.Event()
77 listener_gone = threading.Event()
79 # `listener` runs in a thread. It opens a socket listening on PORT, and
80 # sits in an accept() until the main thread connects. Then it rudely
81 # closes the socket, and sets Event `listener_gone` to let the main thread
82 # know the socket is gone.
83 def listener():
84 s = socket.socket()
85 PORT[0] = test_support.bind_port(s, '', PORT[0])
86 s.listen(5)
87 listener_ready.set()
88 s.accept()
89 s = None # reclaim the socket object, which also closes it
90 listener_gone.set()
92 def connector():
93 listener_ready.wait()
94 s = socket.socket()
95 s.connect(('localhost', PORT[0]))
96 listener_gone.wait()
97 try:
98 ssl_sock = socket.ssl(s)
99 except socket.sslerror:
100 pass
101 else:
102 raise test_support.TestFailed(
103 'connecting to closed SSL socket should have failed')
105 t = threading.Thread(target=listener)
106 t.start()
107 connector()
108 t.join()
110 def test_main():
111 if not hasattr(socket, "ssl"):
112 raise test_support.TestSkipped("socket module has no ssl support")
113 test_rude_shutdown()
114 test_basic()
115 test_timeout()
117 if __name__ == "__main__":
118 test_main()