qa: fix resource leak with loopback connections in threaded tests

neo/tests/threaded/__init__.py

@@ -321,6 +321,7 @@ class ServerNode(Node):
             master_nodes = kw.get('master_nodes', cluster.master_nodes)
             name = kw.get('name', cluster.name)
         port = address[1]
+        if address is not BIND:
             self._node_list[port] = weakref.proxy(self)
         self._init_args = init_args = kw.copy()
         init_args['cluster'] = cluster
@@ -933,20 +934,17 @@ class NEOThreadedTest(NeoTestBase):
 
     tic = Serialized.tic
 
+    @contextmanager
     def getLoopbackConnection(self):
-        app = MasterApplication(getSSL=NEOCluster.SSL,
-            getReplicas=0, getPartitions=1)
+        app = MasterApplication(address=BIND,
+            getSSL=NEOCluster.SSL, getReplicas=0, getPartitions=1)
+        try:
             handler = EventHandler(app)
             app.listening_conn = ListeningConnection(app, handler, app.server)
-        node = app.nm.createMaster(address=app.listening_conn.getAddress(),
-                                   uuid=app.uuid)
-        conn = ClientConnection.__new__(ClientConnection)
-        def reset():
-            conn.__dict__.clear()
-            conn.__init__(app, handler, node)
-            conn.reset = reset
-        reset()
-        return conn
+            yield ClientConnection(app, handler, app.nm.createMaster(
+                address=app.listening_conn.getAddress(), uuid=app.uuid))
+        finally:
+            app.close()
 
     def getUnpickler(self, conn):
         reader = conn._reader

neo/tests/threaded/test.py

@@ -1209,14 +1209,14 @@ class Test(NEOThreadedTest):
                 self.assertEqual(s.dm.getLastIDs()[0], truncate_tid)
 
     def testConnectionTimeout(self):
-        conn = self.getLoopbackConnection()
+        with self.getLoopbackConnection() as conn:
             conn.KEEP_ALIVE
-        with Patch(conn, KEEP_ALIVE=0):
-            while conn.connecting:
-                conn.em.poll(1)
             def onTimeout(orig):
                 conn.idle()
                 orig()
+            with Patch(conn, KEEP_ALIVE=0):
+                while conn.connecting:
+                    conn.em.poll(1)
                 with Patch(conn, onTimeout=onTimeout):
                     conn.em.poll(1)
             self.assertFalse(conn.isClosed())

neo/tests/threaded/testSSL.py

@@ -36,12 +36,8 @@ class SSLTests(SSLMixin, test.Test):
     testDeadlockAvoidance = None
     testUndoConflict = testUndoConflictDuringStore = None
 
-    def testAbortConnection(self):
-        for after_handshake in 1, 0:
-            try:
-                conn.reset()
-            except UnboundLocalError:
-                conn = self.getLoopbackConnection()
+    def testAbortConnection(self, after_handshake=1):
+        with self.getLoopbackConnection() as conn:
             conn.ask(Packets.Ping())
             connector = conn.getConnector()
             del connector.connect_limit[connector.addr]
@@ -58,6 +54,9 @@ class SSLTests(SSLMixin, test.Test):
                 conn.em.poll(1)
             self.assertIs(conn.getConnector(), None)
 
+    def testAbortConnectionBeforeHandshake(self):
+        self.testAbortConnection(0)
+
 class SSLReplicationTests(SSLMixin, testReplication.ReplicationTests):
     # do not repeat slowest tests with SSL
     testBackupNodeLost = testBackupNormalCase = None