qa: add threaded test helper to filter connection by packet type

neo/tests/threaded/__init__.py

@@ -22,7 +22,7 @@ from collections import deque
 from ConfigParser import SafeConfigParser
 from contextlib import contextmanager
 from itertools import count
-from functools import wraps
+from functools import partial, wraps
 from thread import get_ident
 from zlib import decompress
 from mock import Mock
@@ -36,7 +36,7 @@ from neo.lib.connection import BaseConnection, \
 from neo.lib.connector import SocketConnector, ConnectorException
 from neo.lib.handler import EventHandler
 from neo.lib.locking import SimpleQueue
-from neo.lib.protocol import ClusterStates, NodeStates, NodeTypes
+from neo.lib.protocol import ClusterStates, NodeStates, NodeTypes, Packets
 from neo.lib.util import cached_property, parseMasterList, p64
 from .. import NeoTestBase, Patch, getTempDirectory, setupMySQLdb, \
     ADDRESS_TYPE, IP_VERSION_FORMAT_DICT, DB_PREFIX, DB_SOCKET, DB_USER
@@ -553,6 +553,22 @@ class ConnectionFilter(object):
     def __contains__(self, filter):
         return filter in self.filter_dict
 
+    def byPacket(self, packet_type, *args):
+        patches = []
+        other = []
+        for x in args:
+            (patches if isinstance(x, Patch) else other).append(x)
+        def delay(conn, packet):
+            return isinstance(packet, packet_type) and False not in (
+                callback(conn) for callback in other)
+        self.add(delay, *patches)
+        return delay
+
+    def __getattr__(self, attr):
+        if attr.startswith('delay'):
+            return partial(self.byPacket, getattr(Packets, attr[5:]))
+        return self.__getattribute__(attr)
+
 class NEOCluster(object):
 
     SSL = None

neo/tests/threaded/test.py

@@ -248,8 +248,6 @@ class Test(NEOThreadedTest):
 
     def testDelayedUnlockInformation(self):
         except_list = []
-        def delayUnlockInformation(conn, packet):
-            return isinstance(packet, Packets.NotifyUnlockInformation)
         def onStoreObject(orig, tm, ttid, serial, oid, *args):
             if oid == resume_oid and delayUnlockInformation in m2s:
                 m2s.remove(delayUnlockInformation)
@@ -265,13 +263,13 @@ class Test(NEOThreadedTest):
             c.root()[0] = ob = PCounter()
             with cluster.master.filterConnection(cluster.storage) as m2s:
                 resume_oid = None
-                m2s.add(delayUnlockInformation,
+                delayUnlockInformation = m2s.delayNotifyUnlockInformation(
                     Patch(TransactionManager, storeObject=onStoreObject))
                 t.commit()
                 resume_oid = ob._p_oid
                 ob._p_changed = 1
                 t.commit()
-                self.assertFalse(delayUnlockInformation in m2s)
+                self.assertNotIn(delayUnlockInformation, m2s)
         finally:
             cluster.stop()
         self.assertEqual(except_list, [DelayedError])
@@ -451,8 +449,7 @@ class Test(NEOThreadedTest):
             r[''] = ''
             with Patch(ClientOperationHandler, askObject=askObject):
                 with cluster.master.filterConnection(cluster.storage) as m2s:
-                    m2s.add(lambda conn, packet: # delay unlock
-                        isinstance(packet, Packets.NotifyUnlockInformation))
+                    m2s.delayNotifyUnlockInformation()
                     t.commit()
                     c.cacheMinimize()
                     cluster.client._cache.clear()
@@ -524,8 +521,7 @@ class Test(NEOThreadedTest):
             orig()
         def stop():
             with cluster.master.filterConnection(s0) as m2s0:
-                m2s0.add(lambda conn, packet:
-                    isinstance(packet, Packets.NotifyPartitionChanges))
+                m2s0.delayNotifyPartitionChanges()
                 s1.stop()
                 cluster.join((s1,))
                 self.assertEqual(getClusterState(), ClusterStates.RUNNING)
@@ -566,8 +562,6 @@ class Test(NEOThreadedTest):
 
     def testVerificationCommitUnfinishedTransactions(self):
         """ Verification step should commit locked transactions """
-        def delayUnlockInformation(conn, packet):
-            return isinstance(packet, Packets.NotifyUnlockInformation)
         def onLockTransaction(storage, die=False):
             def lock(orig, *args, **kw):
                 if die:
@@ -608,7 +602,7 @@ class Test(NEOThreadedTest):
                 self.assertEqual([u64(o._p_oid) for o in (r, x, y)], range(3))
                 r[2] = 'ok'
                 with cluster.master.filterConnection(s0) as m2s:
-                    m2s.add(delayUnlockInformation)
+                    m2s.delayNotifyUnlockInformation()
                     t.commit()
                     x.value = 1
                     # s0 will accept to store y (because it's not locked) but will
@@ -916,8 +910,7 @@ class Test(NEOThreadedTest):
             client.store(x1._p_oid, x1._p_serial, y, '', txn)
             # Delay invalidation for x
             with cluster.master.filterConnection(cluster.client) as m2c:
-                m2c.add(lambda conn, packet:
-                    isinstance(packet, Packets.InvalidateObjects))
+                m2c.delayInvalidateObjects()
                 tid = client.tpc_finish(txn, None)
                 # Change to x is committed. Testing connection must ask the
                 # storage node to return original value of x, even if we
@@ -1164,8 +1157,7 @@ class Test(NEOThreadedTest):
                  cluster.master.filterConnection(cluster.storage) as m2s:
                 s2m.add(delayAnswerLockInformation, Patch(cluster.client,
                     _connectToPrimaryNode=_connectToPrimaryNode))
-                m2s.add(lambda conn, packet:
-                    isinstance(packet, Packets.NotifyUnlockInformation))
+                m2s.delayNotifyUnlockInformation()
                 t.commit() # the final TID is returned by the storage (tm)
             t.begin()
             self.assertEqual(r['x'].value, 2)
@@ -1208,8 +1200,6 @@ class Test(NEOThreadedTest):
             cluster.stop()
 
     def testRecycledClientUUID(self):
-        def delayNotifyInformation(conn, packet):
-            return isinstance(packet, Packets.NotifyNodeInformation)
         def notReady(orig, *args):
             m2s.discard(delayNotifyInformation)
             return orig(*args)
@@ -1218,7 +1208,7 @@ class Test(NEOThreadedTest):
             cluster.start()
             cluster.getTransaction()
             with cluster.master.filterConnection(cluster.storage) as m2s:
-                m2s.add(delayNotifyInformation)
+                delayNotifyInformation = m2s.delayNotifyNodeInformation()
                 cluster.client.master_conn.close()
                 with cluster.newClient() as client, Patch(
                         client.storage_bootstrap_handler, notReady=notReady):
@@ -1504,8 +1494,7 @@ class Test(NEOThreadedTest):
             with LockLock() as ll, s1.filterConnection(cluster.client) as f, \
                     Patch(cluster.client.storage_handler,
                           answerStoreObject=answerStoreObject) as p:
-                f.add(lambda conn, packet:
-                    isinstance(packet, Packets.AnswerStoreObject))
+                f.delayAnswerStoreObject()
                 t = self.newThread(t1.commit)
                 ll()
             t.join()

neo/tests/threaded/testReplication.py

@@ -155,8 +155,7 @@ class ReplicationTests(NEOThreadedTest):
                 backup.neoctl.setClusterState(ClusterStates.STARTING_BACKUP)
                 self.tic()
                 with ConnectionFilter() as f:
-                    f.add(lambda conn, packet: conn.getUUID() is None and
-                        isinstance(packet, Packets.AddObject))
+                    f.delayAddObject(lambda conn: conn.getUUID() is None)
                     while not f.filtered_count:
                         importZODB(1)
                     self.tic()
@@ -271,8 +270,7 @@ class ReplicationTests(NEOThreadedTest):
     def testBackupUpstreamStorageDead(self, backup):
         upstream = backup.upstream
         with ConnectionFilter() as f:
-            f.add(lambda conn, packet:
-                isinstance(packet, Packets.InvalidateObjects))
+            f.delayInvalidateObjects()
             upstream.importZODB()(1)
         count = [0]
         def _connect(orig, conn):
@@ -301,8 +299,7 @@ class ReplicationTests(NEOThreadedTest):
         """
         upstream = backup.upstream
         with upstream.master.filterConnection(upstream.storage) as f:
-            f.add(lambda conn, packet:
-                isinstance(packet, Packets.NotifyUnlockInformation))
+            f.delayNotifyUnlockInformation()
             upstream.importZODB()(1)
             self.tic()
         self.tic()
@@ -320,8 +317,7 @@ class ReplicationTests(NEOThreadedTest):
             try:
                 backup.start()
                 with ConnectionFilter() as f:
-                    f.add(lambda conn, packet:
-                        isinstance(packet, Packets.AskPartitionTable) and
+                    f.delayAskPartitionTable(lambda conn:
                         isinstance(conn.getHandler(), BackupHandler))
                     backup.neoctl.setClusterState(ClusterStates.STARTING_BACKUP)
                     upstream.importZODB()(1)
@@ -349,8 +345,7 @@ class ReplicationTests(NEOThreadedTest):
         importZODB(1)
         backup.reset()
         with ConnectionFilter() as f:
-            f.add(lambda conn, packet:
-                isinstance(packet, Packets.AskFetchTransactions))
+            f.delayAskFetchTransactions()
             backup.start()
             self.assertEqual(last_tid, backup.backup_tid)
         self.tic()
@@ -457,8 +452,7 @@ class ReplicationTests(NEOThreadedTest):
             cluster.neoctl.enableStorageList([s1.uuid])
             cluster.neoctl.tweakPartitionTable()
             with cluster.master.filterConnection(cluster.client) as m2c:
-                m2c.add(lambda conn, packet:
-                    isinstance(packet, Packets.NotifyPartitionChanges))
+                m2c.delayNotifyPartitionChanges()
                 self.tic()
                 self.assertEqual('foo', storage.load(oid)[0])
         finally: