qa: make closure of NEOCluster more reliable in treaded tests

Instances of NEOCluster were not deleted as soon as the only referrers were
weak proxies (at least that's what a quick check with the 'gc' module showed
at the beginning of tearDown). In some cases, __del__ was called while the next
test was logging a message, which led to deadlocks.

Without those proxies, it may be reliable, but only on CPython. See
  http://doc.pypy.org/en/latest/cpython_differences.html#differences-related-to-garbage-collection-strategies

Relying on __del__ to close a cluster was wrong. NEOCluster is now a context
manager that does it explicitly at exit, in addition to automatically stop it.
The NEOCluster.stop method combines the previous stop/__del__/reset methods.

A new 'with_cluster' decorator is also added to avoid excessive indentation
in tests. Unindentation of existing tests will be done later.

neo/tests/threaded/__init__.py

@@ -624,6 +624,8 @@ class NEOCluster(object):
             patch.revert()
         Serialized.stop()
 
+    started = False
+
     def __init__(self, master_count=1, partitions=1, replicas=0, upstream=None,
                        adapter=os.getenv('NEO_TESTS_ADAPTER', 'SQLite'),
                        storage_count=None, db_list=None, clear_databases=True,
@@ -632,6 +634,7 @@ class NEOCluster(object):
         self.name = 'neo_%s' % self._allocate('name',
             lambda: random.randint(0, 100))
         self.compress = compress
+        self.num_partitions = partitions
         master_list = [MasterApplication.newAddress()
                        for _ in xrange(master_count)]
         self.master_nodes = ' '.join('%s:%s' % x for x in master_list)
@@ -675,7 +678,6 @@ class NEOCluster(object):
         self.storage_list = [StorageApplication(getDatabase=db % x, **kw)
                              for x in db_list]
         self.admin_list = [AdminApplication(**kw)]
-        self.neoctl = NeoCTL(self.admin.getVirtualAddress(), ssl=self.SSL)
 
     def __repr__(self):
         return "<%s(%s) at 0x%x>" % (self.__class__.__name__,
@@ -711,18 +713,16 @@ class NEOCluster(object):
         return master
     ###
 
-    def reset(self, clear_database=False):
-        for node_type in 'master', 'storage', 'admin':
-            kw = {}
-            if node_type == 'storage':
-                kw['clear_database'] = clear_database
-            for node in getattr(self, node_type + '_list'):
-                node.resetNode(**kw)
-        self.neoctl.close()
-        self.neoctl = NeoCTL(self.admin.getVirtualAddress(), ssl=self.SSL)
+    def __enter__(self):
+        return self
+
+    def __exit__(self, t, v, tb):
+        self.stop(None)
 
     def start(self, storage_list=None, fast_startup=False):
+        self.started = True
         self._patch()
+        self.neoctl = NeoCTL(self.admin.getVirtualAddress(), ssl=self.SSL)
         for node_type in 'master', 'admin':
             for node in getattr(self, node_type + '_list'):
                 node.start()
@@ -741,6 +741,40 @@ class NEOCluster(object):
         assert state in (ClusterStates.RUNNING, ClusterStates.BACKINGUP), state
         self.enableStorageList(storage_list)
 
+    def stop(self, clear_database=False, __print_exc=traceback.print_exc):
+        if self.started:
+            del self.started
+            logging.debug("stopping %s", self)
+            client = self.__dict__.get("client")
+            client is None or self.__dict__.pop("db", client).close()
+            node_list = self.admin_list + self.storage_list + self.master_list
+            for node in node_list:
+                node.stop()
+            try:
+                node_list.append(client.poll_thread)
+            except AttributeError: # client is None or thread is already stopped
+                pass
+            self.join(node_list)
+            self.neoctl.close()
+            del self.neoctl
+            logging.debug("stopped %s", self)
+            self._unpatch()
+        if clear_database is None:
+            try:
+                for node_type in 'admin', 'storage', 'master':
+                    for node in getattr(self, node_type + '_list'):
+                        node.close()
+            except:
+                __print_exc()
+                raise
+        else:
+            for node_type in 'master', 'storage', 'admin':
+                kw = {}
+                if node_type == 'storage':
+                    kw['clear_database'] = clear_database
+                for node in getattr(self, node_type + '_list'):
+                    node.resetNode(**kw)
+
     def _newClient(self):
         return ClientApplication(name=self.name, master_nodes=self.master_nodes,
                                  compress=self.compress, ssl=self.SSL)
@@ -801,21 +835,6 @@ class NEOCluster(object):
             Serialized.tic()
             thread_list = [t for t in thread_list if t.is_alive()]
 
-    def stop(self):
-        logging.debug("stopping %s", self)
-        client = self.__dict__.get("client")
-        client is None or self.__dict__.pop("db", client).close()
-        node_list = self.admin_list + self.storage_list + self.master_list
-        for node in node_list:
-            node.stop()
-        try:
-            node_list.append(client.poll_thread)
-        except AttributeError: # client is None or thread is already stopped
-            pass
-        self.join(node_list)
-        logging.debug("stopped %s", self)
-        self._unpatch()
-
     def getNodeState(self, node):
         uuid = node.uuid
         for node in self.neoctl.getNodeList(node.node_type):
@@ -861,16 +880,6 @@ class NEOCluster(object):
         txn = transaction.TransactionManager()
         return txn, (self.db if db is None else db).open(txn)
 
-    def __del__(self, __print_exc=traceback.print_exc):
-        try:
-            self.neoctl.close()
-            for node_type in 'admin', 'storage', 'master':
-                for node in getattr(self, node_type + '_list'):
-                    node.close()
-        except:
-            __print_exc()
-            raise
-
     def extraCellSortKey(self, key):
         return Patch(self.client.cp, getCellSortKey=lambda orig, cell:
             (orig(cell), key(cell)))
@@ -1000,3 +1009,13 @@ def predictable_random(seed=None):
                     = random
         return wraps(wrapped)(wrapper)
     return decorator
+
+def with_cluster(start_cluster=True, **cluster_kw):
+    def decorator(wrapped):
+        def wrapper(self, *args, **kw):
+            with NEOCluster(**cluster_kw) as cluster:
+                if start_cluster:
+                    cluster.start()
+                return wrapped(self, cluster, *args, **kw)
+        return wraps(wrapped)(wrapper)
+    return decorator

neo/tests/threaded/testImporter.py

@@ -172,8 +172,7 @@ class ImporterTests(NEOThreadedTest):
             c.db().close()
         #del importer[0][1][importer.pop()[0]]
         # Start NEO cluster with transparent import of a multi-base ZODB.
-        cluster = NEOCluster(compress=False, importer=importer)
-        try:
+        with NEOCluster(compress=False, importer=importer) as cluster:
             # Suspend import for a while, so that import
             # is finished in the middle of the below 'for' loop.
             # Use a slightly different main loop for storage so that it
@@ -220,8 +219,6 @@ class ImporterTests(NEOThreadedTest):
                 (x[i:] or '.', sorted(y), sorted(z))
                 for x, y, z in os.walk(src_root)))
             t.commit()
-        finally:
-            cluster.stop()
 
 
 if __name__ == "__main__":

neo/tests/zodb/__init__.py

@@ -44,7 +44,10 @@ class ZODBTestCase(TestCase):
     def _tearDown(self, success):
         self._storage.cleanup()
         try:
+            if functional:
                 self.neo.stop()
+            else:
+                self.neo.stop(None)
         except Exception:
             if success:
                 raise