add manual tests

2 files changed, 64 insertions, 0 deletions

diff --git a/manual_tests/add_reader.py b/manual_tests/add_reader.py
new file mode 100644
index 0000000..d2ff2b1
--- /dev/null
+++ b/manual_tests/add_reader.py
@@ -0,0 +1,33 @@
+import aiogreen
+from socket import socketpair
+
+# Create a pair of connected file descriptors
+rsock, wsock = socketpair()
+loop = aiogreen.EventLoop()
+
+def reader():
+    data = rsock.recv(100)
+    print("Received: %s" % data.decode())
+    # We are done: unregister the file descriptor
+    loop.remove_reader(rsock)
+    # Stop the event loop
+    loop.stop()
+
+def writer():
+    loop.remove_writer(wsock)
+
+    # Simulate the reception of data from the network
+    print("Send: abc")
+    loop.call_soon(wsock.send, 'abc'.encode())
+
+# Register the file descriptor for read event
+loop.add_reader(rsock, reader)
+loop.add_writer(wsock, writer)
+
+# Run the event loop
+loop.run_forever()
+
+# We are done, close sockets and the event loop
+rsock.close()
+wsock.close()
+loop.close()
diff --git a/manual_tests/tcp_client.py b/manual_tests/tcp_client.py
new file mode 100644
index 0000000..e0c59c2
--- /dev/null
+++ b/manual_tests/tcp_client.py
@@ -0,0 +1,31 @@
+import aiogreen
+import trollius
+import logging; logging.basicConfig(level=logging.DEBUG)
+
+class EchoClientProtocol(trollius.Protocol):
+    def __init__(self, message, loop):
+        self.message = message
+        self.loop = loop
+
+    def connection_made(self, transport):
+        transport.write(self.message.encode())
+        print('Data sent: {!r}'.format(self.message))
+
+    def data_received(self, data):
+        print('Data received: {!r}'.format(data.decode()))
+
+    def connection_lost(self, exc):
+        print('The server closed the connection')
+        print('Stop the event lop')
+        self.loop.stop()
+
+loop = aiogreen.EventLoop()
+message = 'Hello World!'
+coro = loop.create_connection(lambda: EchoClientProtocol(message, loop),
+                              '127.0.0.1', 8888)
+loop.run_until_complete(coro)
+
+loop.run_forever()
+loop.close()
+
+