from eventlet.hubs import get_hub
from eventlet.green import select, socket, time, ssl
from eventlet.support import capture_stderr, get_errno, six
-from tests import (
- LimitedTestCase, main,
- skip_with_pyevent, skipped, skip_if, skip_on_windows,
-)
+import tests
if six.PY3:
return False
-class TestGreenSocket(LimitedTestCase):
+class TestGreenSocket(tests.LimitedTestCase):
def assertWriteToClosedFileRaises(self, fd):
if sys.version_info[0] < 3:
# 2.x socket._fileobjects are odd: writes don't check
server.close()
client.close()
- @skip_with_pyevent
+ @tests.skip_with_pyevent
def test_raised_multiple_readers(self):
debug.hub_prevent_multiple_readers(True)
s.sendall(b'b')
a.wait()
- @skip_with_pyevent
- @skip_if(using_epoll_hub)
- @skip_if(using_kqueue_hub)
+ @tests.skip_with_pyevent
+ @tests.skip_if(using_epoll_hub)
+ @tests.skip_if(using_kqueue_hub)
def test_closure(self):
def spam_to_me(address):
sock = eventlet.connect(address)
assert select.select([], [s1], [], 0) == ([], [s1], [])
assert select.select([], [s1], [], 0) == ([], [s1], [])
+ def test_shutdown_safe(self):
+ sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ sock.close()
+ # should not raise
+ greenio.shutdown_safe(sock)
+
def test_get_fileno_of_a_socket_works():
class DummySocket(object):
assert select.get_fileno(123) == 123
+expected_get_fileno_type_error_message = (
+ 'Expected int or long, got <%s \'str\'>' % ('type' if six.PY2 else 'class'))
+
+
def test_get_fileno_of_wrong_type_fails():
try:
select.get_fileno('foo')
except TypeError as ex:
- assert str(ex) == 'Expected int or long, got <type \'str\'>'
+ assert str(ex) == expected_get_fileno_type_error_message
else:
assert False, 'Expected TypeError not raised'
try:
select.get_fileno(DummySocket())
except TypeError as ex:
- assert str(ex) == 'Expected int or long, got <type \'str\'>'
+ assert str(ex) == expected_get_fileno_type_error_message
else:
assert False, 'Expected TypeError not raised'
-class TestGreenPipe(LimitedTestCase):
- @skip_on_windows
+class TestGreenPipe(tests.LimitedTestCase):
+ @tests.skip_on_windows
def setUp(self):
super(self.__class__, self).setUp()
self.tempdir = tempfile.mkdtemp('_green_pipe_test')
def test_pipe(self):
r, w = os.pipe()
- rf = greenio.GreenPipe(r, 'r')
- wf = greenio.GreenPipe(w, 'w', 0)
+ rf = greenio.GreenPipe(r, 'rb')
+ wf = greenio.GreenPipe(w, 'wb', 0)
def sender(f, content):
for ch in map(six.int2byte, six.iterbytes(content)):
# also ensures that readline() terminates on '\n' and '\r\n'
r, w = os.pipe()
- r = greenio.GreenPipe(r)
- w = greenio.GreenPipe(w, 'w')
+ r = greenio.GreenPipe(r, 'rb')
+ w = greenio.GreenPipe(w, 'wb')
def writer():
eventlet.sleep(.1)
def test_pipe_writes_large_messages(self):
r, w = os.pipe()
- r = greenio.GreenPipe(r)
- w = greenio.GreenPipe(w, 'w')
+ r = greenio.GreenPipe(r, 'rb')
+ w = greenio.GreenPipe(w, 'wb')
large_message = b"".join([1024 * six.int2byte(i) for i in range(65)])
for i in range(65):
buf = r.read(1024)
- expected = 1024 * chr(i)
+ expected = 1024 * six.int2byte(i)
self.assertEqual(
buf, expected,
"expected=%r..%r, found=%r..%r iter=%d"
gt.wait()
def test_seek_on_buffered_pipe(self):
- f = greenio.GreenPipe(self.tempdir + "/TestFile", 'w+', 1024)
+ f = greenio.GreenPipe(self.tempdir + "/TestFile", 'wb+', 1024)
self.assertEqual(f.tell(), 0)
f.seek(0, 2)
self.assertEqual(f.tell(), 0)
self.assertEqual(f.tell(), 10)
f.seek(0)
value = f.read(1)
- self.assertEqual(value, '1')
+ self.assertEqual(value, b'1')
self.assertEqual(f.tell(), 1)
value = f.read(1)
- self.assertEqual(value, '2')
+ self.assertEqual(value, b'2')
self.assertEqual(f.tell(), 2)
f.seek(0, 1)
- self.assertEqual(f.readline(), '34567890')
+ self.assertEqual(f.readline(), b'34567890')
f.seek(-5, 1)
- self.assertEqual(f.readline(), '67890')
+ self.assertEqual(f.readline(), b'67890')
f.seek(0)
- self.assertEqual(f.readline(), '1234567890')
+ self.assertEqual(f.readline(), b'1234567890')
f.seek(0, 2)
- self.assertEqual(f.readline(), '')
+ self.assertEqual(f.readline(), b'')
def test_truncate(self):
- f = greenio.GreenPipe(self.tempdir + "/TestFile", 'w+', 1024)
+ f = greenio.GreenPipe(self.tempdir + "/TestFile", 'wb+', 1024)
f.write(b'1234567890')
f.truncate(9)
self.assertEqual(f.tell(), 9)
-class TestGreenIoLong(LimitedTestCase):
+class TestGreenIoLong(tests.LimitedTestCase):
TEST_TIMEOUT = 10 # the test here might take a while depending on the OS
- @skip_with_pyevent
+ @tests.skip_with_pyevent
def test_multiple_readers(self, clibufsize=False):
debug.hub_prevent_multiple_readers(False)
recvsize = 2 * min_buf_size()
assert len(results2) > 0
debug.hub_prevent_multiple_readers()
- @skipped # by rdw because it fails but it's not clear how to make it pass
- @skip_with_pyevent
+ @tests.skipped # by rdw because it fails but it's not clear how to make it pass
+ @tests.skip_with_pyevent
def test_multiple_readers2(self):
self.test_multiple_readers(clibufsize=True)
-class TestGreenIoStarvation(LimitedTestCase):
+class TestGreenIoStarvation(tests.LimitedTestCase):
# fixme: this doesn't succeed, because of eventlet's predetermined
# ordering. two processes, one with server, one with client eventlets
# might be more reliable?
TEST_TIMEOUT = 300 # the test here might take a while depending on the OS
- @skipped # by rdw, because it fails but it's not clear how to make it pass
- @skip_with_pyevent
+ @tests.skipped # by rdw, because it fails but it's not clear how to make it pass
+ @tests.skip_with_pyevent
def test_server_starvation(self, sendloops=15):
recvsize = 2 * min_buf_size()
sendsize = 10000 * recvsize
assert err.getvalue() == ''
-if __name__ == '__main__':
- main()
+def test_double_close_219():
+ tests.run_isolated('greenio_double_close_219.py')
Code Review / packages / trusty / python-eventlet.git / blobdiff