openstack/deb-python-eventlet
Code Issues Proposed changes

Converted test_socket_ssl away from copy-n-paste.

Browse Source
This commit is contained in:
Ryan Williams
2009-08-06 21:48:46 -07:00
parent 884dabd979
commit 0be3ac5ce1
3 changed files with 22 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/stdlib/test_socket.py
View File

@@ -1,7 +1,6 @@
#!/usr/bin/env python #!/usr/bin/env python
from test import test_socket from test import test_socket
from test.test_socket import *
from eventlet.green import socket from eventlet.green import socket
from eventlet.green import select from eventlet.green import select
@@ -15,5 +14,7 @@ test_socket.time = time
test_socket.thread = thread test_socket.thread = thread
test_socket.threading = threading test_socket.threading = threading
from test.test_socket import *
if __name__ == "__main__": if __name__ == "__main__":
test_main() test_main()

20
tests/stdlib/test_socket_ssl.py Normal file
View File

@@ -0,0 +1,20 @@
#!/usr/bin/env python
from test import test_socket_ssl
from eventlet.green import socket
from eventlet.green import urllib
from eventlet.green import threading
test_socket_ssl.socket = socket
# bwahaha
import sys
sys.modules['urllib'] = urllib
sys.modules['threading'] = threading
# to get past the silly 'requires' check
test_socket_ssl.__name__ = '__main__'
from test.test_socket_ssl import *
if __name__ == "__main__":
test_main()

180
tests/test_socket_ssl.py
View File

@@ -1,180 +0,0 @@
# Test just the SSL support in the socket module, in a moderately bogus way.
import sys
from tests import test_support
from eventlet.green import socket
import errno
import unittest
# 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')
from eventlet.green 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)
try:
f = urllib.urlopen('https://sf.net')
except IOError, exc:
if exc.errno == errno.ETIMEDOUT:
raise test_support.ResourceDenied('HTTPS connection is timing out')
else:
raise
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).
ADDR = "pop.gmail.com", 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 ..."
from eventlet.green import threading
# 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_rude_shutdown__write():
if test_support.verbose:
print "test_rude_shutdown__variant ..."
from eventlet.green import threading
# 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)
ssl_sock.write("hello")
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()
class Test(unittest.TestCase):
test_basic = lambda self: test_basic()
test_timeout = lambda self: test_timeout()
test_rude_shutdown = lambda self: test_rude_shutdown()
test_rude_shutdown__write = lambda self: test_rude_shutdown__write()
def test_main():
if not hasattr(socket, "ssl"):
raise test_support.TestSkipped("socket module has no ssl support")
test_support.run_unittest(Test)
if __name__ == "__main__":
test_main()