Commit 27f67868 authored by Denis Bilenko's avatar Denis Bilenko

add test for issue #26. Thanks to Cyril Bay for the bug report.

parent ce2380d0
import gevent
from gevent import socket
import greentest
# XXX also test: send, sendall, recvfrom, recvfrom_into, sendto
class Test(greentest.TestCase):
def setUp(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect(('www.google.com', 80))
def tearDown(self):
self.sock.close()
def test_recv_closed(self):
receiver = gevent.spawn(self.sock.recv, 25)
gevent.sleep(0.001)
self.sock.close()
receiver.join(timeout=0.001)
assert receiver.ready(), receiver
self.assertEqual(receiver.value, '')
def test_recv_twice(self):
receiver = gevent.spawn(self.sock.recv, 25)
gevent.sleep(0.001)
self.assertRaises(AssertionError, self.sock.recv, 25)
self.assertRaises(AssertionError, self.sock.recv, 25)
receiver.kill()
if __name__ == '__main__':
greentest.main()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment