1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
|
import eventlet
import tests
socket = eventlet.patcher.original('socket')
threading = eventlet.patcher.original('threading')
class TcpEchoClientProtocol(tests.asyncio.Protocol):
def __init__(self, message, loop):
self.message = message
self.loop = loop
self.state = 'new'
self.received = None
def connection_made(self, transport):
self.state = 'ping'
transport.write(self.message)
def data_received(self, data):
self.state = 'pong'
self.received = data
def connection_lost(self, exc):
self.state = 'closed'
self.loop.stop()
class TcpServer(threading.Thread):
def __init__(self, host, port, event):
super(TcpServer, self).__init__()
self.host = host
self.port = port
self.event = event
self.sock = None
self.client = None
def run(self):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock = sock
try:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((self.host, self.port))
sock.listen(1)
self.event.set()
client, addr = sock.accept()
self.client = client
try:
message = client.recv(100)
client.sendall(message)
finally:
client.close()
self.client = None
finally:
sock.close()
self.sock = None
def stop(self):
self.join()
class NetworkTests(tests.TestCase):
def test_tcp_hello(self):
port = 8888
host = '127.0.0.1'
message = b'Hello World!'
event = threading.Event()
server = TcpServer(host, port, event)
server.start()
self.addCleanup(server.stop)
event.wait()
proto = TcpEchoClientProtocol(message, self.loop)
coro = self.loop.create_connection(lambda: proto, host, port)
self.loop.run_until_complete(coro)
self.assertNotEqual(proto.state, 'new')
self.loop.run_forever()
self.assertEqual(proto.state, 'closed')
self.assertEqual(proto.received, message)
if __name__ == '__main__':
import unittest
unittest.main()
|
