diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/__init__.py | 5 | ||||
| -rw-r--r-- | tests/api_test.py | 36 | ||||
| -rw-r--r-- | tests/socket_test.py | 11 | ||||
| -rw-r--r-- | tests/timeout_test.py | 10 |
4 files changed, 37 insertions, 25 deletions
diff --git a/tests/__init__.py b/tests/__init__.py index 6ff6d2a..15058e6 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -346,6 +346,11 @@ def run_isolated(path, prefix='tests/isolated/', env=None, args=None, timeout=No assert ok, 'Expected single line "pass" in stdout' +def check_is_timeout(obj): + value_text = getattr(obj, 'is_timeout', '(missing)') + assert obj.is_timeout, 'type={0} str={1} .is_timeout={2}'.format(type(obj), str(obj), value_text) + + certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt') private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key') diff --git a/tests/api_test.py b/tests/api_test.py index 1cdab88..18ab9f5 100644 --- a/tests/api_test.py +++ b/tests/api_test.py @@ -1,12 +1,7 @@ -import os -from unittest import TestCase, main - -from nose.tools import eq_ - import eventlet -from eventlet import greenio, hubs, greenthread, spawn +from eventlet import greenio, hubs, greenthread from eventlet.green import ssl -from tests import skip_if_no_ssl +import tests def check_hub(): @@ -21,10 +16,7 @@ def check_hub(): assert not hub.running -class TestApi(TestCase): - - certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt') - private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key') +class TestApi(tests.LimitedTestCase): def test_tcp_listener(self): socket = eventlet.listen(('0.0.0.0', 0)) @@ -50,13 +42,13 @@ class TestApi(TestCase): client = eventlet.connect(('127.0.0.1', server.getsockname()[1])) fd = client.makefile('rb') client.close() - eq_(fd.readline(), b'hello\n') - eq_(fd.read(), b'') + assert fd.readline() == b'hello\n' + assert fd.read() == b'' fd.close() check_hub() - @skip_if_no_ssl + @tests.skip_if_no_ssl def test_connect_ssl(self): def accept_once(listenfd): try: @@ -70,8 +62,8 @@ class TestApi(TestCase): server = eventlet.wrap_ssl( eventlet.listen(('0.0.0.0', 0)), - self.private_key_file, - self.certificate_file, + tests.private_key_file, + tests.certificate_file, server_side=True ) eventlet.spawn_n(accept_once, server) @@ -98,7 +90,7 @@ class TestApi(TestCase): def server(sock): client, addr = sock.accept() eventlet.sleep(0.1) - server_evt = spawn(server, server_sock) + server_evt = eventlet.spawn(server, server_sock) eventlet.sleep(0) try: desc = eventlet.connect(('127.0.0.1', bound_port)) @@ -179,9 +171,9 @@ class TestApi(TestCase): pass -class Foo(object): - pass - +def test_wrap_is_timeout(): + class A(object): + pass -if __name__ == '__main__': - main() + obj = eventlet.wrap_is_timeout(A)() + tests.check_is_timeout(obj) diff --git a/tests/socket_test.py b/tests/socket_test.py index 954eba3..9d725b5 100644 --- a/tests/socket_test.py +++ b/tests/socket_test.py @@ -91,3 +91,14 @@ def test_getaddrinfo_ipv6_scope(): if not socket.has_ipv6: return socket.getaddrinfo('::1%2', 80, socket.AF_INET6) + + +def test_error_is_timeout(): + s1, _ = socket.socketpair() + s1.settimeout(0.01) + try: + s1.recv(1) + except socket.error as e: + tests.check_is_timeout(e) + else: + assert False, 'No timeout, socket.error was not raised' diff --git a/tests/timeout_test.py b/tests/timeout_test.py index 0254b79..e33003f 100644 --- a/tests/timeout_test.py +++ b/tests/timeout_test.py @@ -1,12 +1,12 @@ import eventlet -from tests import LimitedTestCase +import tests DELAY = 0.01 -class TestDirectRaise(LimitedTestCase): +class TestDirectRaise(tests.LimitedTestCase): def test_direct_raise_class(self): try: raise eventlet.Timeout @@ -36,7 +36,7 @@ class TestDirectRaise(LimitedTestCase): str(tm) -class TestWithTimeout(LimitedTestCase): +class TestWithTimeout(tests.LimitedTestCase): def test_with_timeout(self): self.assertRaises(eventlet.Timeout, eventlet.with_timeout, DELAY, eventlet.sleep, DELAY * 10) X = object() @@ -53,3 +53,7 @@ class TestWithTimeout(LimitedTestCase): eventlet.Timeout, eventlet.with_timeout, DELAY, longer_timeout) + + +def test_is_timeout_attribute(): + tests.check_is_timeout(eventlet.Timeout()) |