mirror of
https://github.com/python/cpython.git
synced 2025-07-29 14:15:07 +00:00
877fdb01fe
It makes the error message consistent and always sends to stderr. It would be much better for all the networking tests to hit only python.org.
128 lines
3.5 KiB
Python
128 lines
3.5 KiB
Python
# Test just the SSL support in the socket module, in a moderately bogus way.
|
|
|
|
import sys
|
|
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)
|
|
|
|
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()
|
|
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()
|
|
|
|
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()
|
|
|
|
if __name__ == "__main__":
|
|
test_main()
|