convert last manual test to an unit test

1 files changed, 75 insertions, 0 deletions

diff --git a/tests/test_network.py b/tests/test_network.py
new file mode 100644
index 0000000..b1e9731
--- /dev/null
+++ b/tests/test_network.py
@@ -0,0 +1,75 @@
+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):
+        super(TcpServer, self).__init__()
+        self.host = host
+        self.port = port
+        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)
+
+            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!'
+
+        server = TcpServer(host, port)
+        server.start()
+        self.addCleanup(server.stop)
+
+        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)