change test_unstable_connection

Author: snower

requirements.dev.txt

@@ -1,3 +1,3 @@
 nose
 coverage
-maproxy
+sevent

run-tests

@@ -3,6 +3,18 @@
 export LOG_LEVEL=debug
 STATUS=0
 
+for file in ./tests/test_*.py
+do
+    echo $file
+    nosetests $file \
+        --logging-format='%(asctime)s [%(name)s] %(levelname)-6s %(message)s' \
+        --with-coverage \
+        --cover-package=tormysql
+    if [ $? != 0 ]; then
+        STATUS=1
+    fi
+done
+
 pip install tornado==5.1.1
 
 for file in ./tests/test_*.py

tests/test_unstable_connection.py

@@ -1,57 +1,105 @@
 # encoding: utf-8
+
+import os
+import time
+import threading
 import socket
 import os
 from tornado import gen
-from tornado.ioloop import IOLoop
 from tormysql.pool import ConnectionNotFoundError
 from pymysql import OperationalError
 from tornado.testing import gen_test
 from tormysql import Connection, ConnectionPool
-from maproxy.proxyserver import ProxyServer
-from maproxy.session import SessionFactory
+import sevent
 from tests import BaseTestCase
 
+class Request(object):
+    def __init__(self, conn, host, port):
+        self.conn = conn
+        self.pconn = sevent.tcp.Socket()
+        self.buffer = None
+        self.connected = False
+
+        self.conn.on("data", self.on_data)
+        self.conn.on("close", self.on_close)
+
+        self.pconn.on("connect", self.on_pconnect)
+        self.pconn.on("data", self.on_pdata)
+        self.pconn.on("close", self.on_pclose)
+
+        self.pconn.connect((host, int(port)))
+
+    def on_data(self, conn, data):
+        if self.connected:
+            self.pconn.write(data)
+        else:
+            self.buffer = data
+
+    def on_pdata(self, conn, data):
+        self.conn.write(data)
+
+    def on_close(self, conn):
+        self.pconn.end()
+        try:
+            TestThroughProxy.proxys.remove(self)
+        except:
+            pass
+
+    def on_pclose(self, conn):
+        self.conn.end()
+        try:
+            TestThroughProxy.proxys.remove(self)
+        except:
+            pass
+
+    def on_pconnect(self, conn):
+        self.connected = True
+        if self.buffer:
+            self.pconn.write(self.buffer)
+            self.buffer = None
 
 class TestThroughProxy(BaseTestCase):
+    proxys = []
+
     def setUp(self):
         super(BaseTestCase, self).setUp()
         self.PARAMS = dict(self.PARAMS)
 
         s = socket.socket()
         s.bind(('127.0.0.1', 0))
-
-        self.host, self.port = self.PARAMS['host'], self.PARAMS['port']
         _, self.pport = s.getsockname()
         s.close()
 
+        self.host, self.port = self.PARAMS['host'], self.PARAMS['port']
+
     def init_proxy(self):
-        self.proxy = ProxyServer(
-            self.host,
-            self.port,
-            session_factory=SessionFactory(),
-        )
+        def on_connect(server, conn):
+            TestThroughProxy.proxys.append(Request(conn, self.host, self.port))
+
+        self.proxy_server = sevent.tcp.Server()
+        self.proxy_server.on("connection", on_connect)
+        self.proxy_server.listen(("0.0.0.0", self.pport))
+
         self.PARAMS['port'] = self.pport
         self.PARAMS['host'] = '127.0.0.1'
 
-        self.proxy.listen(self.pport)
-
     def _close_proxy_sessions(self):
-        for sock in self.proxy.SessionsList:
-            try:
-                sock.c2p_stream.close()
-            except:
-                pass
+        for request in TestThroughProxy.proxys:
+            request.conn.end()
 
     def tearDown(self):
         try:
-            self.proxy.stop()
+            for request in TestThroughProxy.proxys:
+                request.conn.end()
+            self.proxy_server.close()
         except:
             pass
         super(BaseTestCase, self).tearDown()
 
     @gen.coroutine
     def _execute_test_connection_closing(self):
         self.init_proxy()
+
         connection = yield Connection(**self.PARAMS)
         cursor = connection.cursor()
         self._close_proxy_sessions()
@@ -62,14 +110,17 @@ def _execute_test_connection_closing(self):
             pass
         else:
             raise AssertionError("Unexpected normal situation")
-        self.proxy.stop()
+
+        self.proxy_server.close()
 
     @gen.coroutine
     def _execute_test_connection_closed(self):
         self.init_proxy()
+
         conn = yield Connection(**self.PARAMS)
         yield conn.close()
-        self.proxy.stop()
+
+        self.proxy_server.close()
 
         try:
             yield Connection(**self.PARAMS)
@@ -81,6 +132,7 @@ def _execute_test_connection_closed(self):
     @gen.coroutine
     def _execute_test_remote_closing(self):
         self.init_proxy()
+
         pool = ConnectionPool(
             max_connections=int(os.getenv("MYSQL_POOL", "5")),
             idle_seconds=7200,
@@ -90,7 +142,9 @@ def _execute_test_remote_closing(self):
         try:
             conn = yield pool.Connection()
             yield conn.do_close()
-            self.proxy.stop()
+
+            self.proxy_server.close()
+
             yield pool.Connection()
         except OperationalError:
             pass
@@ -102,6 +156,7 @@ def _execute_test_remote_closing(self):
     @gen.coroutine
     def _execute_test_pool_closing(self):
         self.init_proxy()
+
         pool = ConnectionPool(
             max_connections=int(os.getenv("MYSQL_POOL", "5")),
             idle_seconds=7200,
@@ -118,10 +173,19 @@ def _execute_test_pool_closing(self):
             raise AssertionError("Unexpected normal situation")
         finally:
             yield pool.close()
-            self.proxy.stop()
+
+        self.proxy_server.close()
 
     @gen_test
     def test(self):
+        loop = sevent.instance()
+        def run():
+            loop.start()
+
+        self.proxy_thread = threading.Thread(target=run)
+        self.proxy_thread.setDaemon(True)
+        self.proxy_thread.start()
+
         yield self._execute_test_connection_closing()
         yield self._execute_test_connection_closed()
         yield self._execute_test_remote_closing()