diff --git a/tests/stdlib/test_socket.py b/tests/stdlib/test_socket.py index d7cfcff..9532222 100644 --- a/tests/stdlib/test_socket.py +++ b/tests/stdlib/test_socket.py @@ -1,7 +1,6 @@ #!/usr/bin/env python from test import test_socket -from test.test_socket import * from eventlet.green import socket from eventlet.green import select @@ -15,5 +14,7 @@ test_socket.time = time test_socket.thread = thread test_socket.threading = threading +from test.test_socket import * + if __name__ == "__main__": test_main() \ No newline at end of file diff --git a/tests/stdlib/test_socket_ssl.py b/tests/stdlib/test_socket_ssl.py new file mode 100644 index 0000000..2a2a579 --- /dev/null +++ b/tests/stdlib/test_socket_ssl.py @@ -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() \ No newline at end of file diff --git a/tests/test_socket_ssl.py b/tests/test_socket_ssl.py deleted file mode 100644 index 86c1c60..0000000 --- a/tests/test_socket_ssl.py +++ /dev/null @@ -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()