2
from gevent import socket
4
from gevent.server import StreamServer
7
from test__server import SimpleStreamServer
10
class Test(unittest.TestCase):
12
ServerSubClass = SimpleStreamServer
14
def makefile(self, timeout=0.1, bufsize=1):
15
sock = socket.create_connection((self.server.server_host, self.server.server_port))
16
sock.settimeout(timeout)
17
return sock.makefile(bufsize=bufsize)
19
def assertConnectionRefused(self):
21
conn = self.makefile()
22
raise AssertionError('Connection was not refused: %r' % (conn._sock, ))
23
except socket.error, ex:
24
if ex[0] != errno.ECONNREFUSED:
27
def assertRequestSucceeded(self):
28
conn = self.makefile()
29
conn.write('GET /ping HTTP/1.0\r\n\r\n')
31
assert result.endswith('\r\n\r\nPONG'), repr(result)
33
def init_server(self):
34
self.server = self.ServerSubClass(('127.0.0.1', 0))
38
def test_socket_shutdown(self):
40
self.server.socket.shutdown(socket.SHUT_RDWR)
41
self.assertConnectionRefused()
42
assert not self.server.started, self.server
44
def test_socket_close(self):
45
self.server = self.ServerSubClass(('127.0.0.1', 0))
47
self.server.socket.close()
48
self.assertConnectionRefused()
49
#assert not self.server.started
51
def test_socket_close_fileno(self):
52
self.server = self.ServerSubClass(('127.0.0.1', 0))
54
os.close(self.server.socket.fileno())
55
self.assertConnectionRefused()
56
#assert not self.server.started
58
def test_socket_file(self):
59
self.server = self.ServerSubClass(('127.0.0.1', 0))
61
os.close(self.server.socket.fileno())
62
f = open("/dev/zero", "r")
63
self.assertConnectionRefused()
67
if __name__ == '__main__':
added
removed
greentest/test__server_close.py
