Merge pull request #13 from zopefoundation/clearSynchs

added ITransactionManager.registeredSynchs and clearSynchs

transaction/_manager.py

@@ -95,6 +95,16 @@ def unregisterSynch(self, synch):
         """
         self._synchs.remove(synch)
 
+    def clearSynchs(self):
+        """ See ITransactionManager.
+        """
+        self._synchs.clear()
+
+    def registeredSynchs(self):
+        """ See ITransactionManager.
+        """
+        return bool(self._synchs)
+
     def isDoomed(self):
         """ See ITransactionManager.
         """

transaction/interfaces.py

@@ -74,6 +74,20 @@ def unregisterSynch(synch):
         life.  See ISynchronizer for details.
         """
 
+    def clearSynchs():
+        """Unregister all registered ISynchronizers.
+
+        This exists to support test cleanup/initialization
+        """
+
+    def registeredSynchs():
+        """Determine if any ISynchronizers are registered.
+
+        Return true is any are registered, and return False otherwise.
+
+        This exists to support test cleanup/initialization
+        """
+
 class ITransaction(Interface):
     """Object representing a running transaction.
 

transaction/tests/test__manager.py

@@ -105,12 +105,27 @@ def test_unregisterSynch(self):
         tm = self._makeOne()
         synch1 = DummySynch()
         synch2 = DummySynch()
+        self.assertFalse(tm.registeredSynchs())
         tm.registerSynch(synch1)
+        self.assertTrue(tm.registeredSynchs())
         tm.registerSynch(synch2)
+        self.assertTrue(tm.registeredSynchs())
         tm.unregisterSynch(synch1)
+        self.assertTrue(tm.registeredSynchs())
         self.assertEqual(len(tm._synchs), 1)
         self.assertFalse(synch1 in tm._synchs)
         self.assertTrue(synch2 in tm._synchs)
+        tm.unregisterSynch(synch2)
+        self.assertFalse(tm.registeredSynchs())
+
+    def test_clearSynchs(self):
+        tm = self._makeOne()
+        synch1 = DummySynch()
+        synch2 = DummySynch()
+        tm.registerSynch(synch1)
+        tm.registerSynch(synch2)
+        tm.clearSynchs()
+        self.assertEqual(len(tm._synchs), 0)
 
     def test_isDoomed_wo_existing_txn(self):
         tm = self._makeOne()

transaction/tests/test_weakset.py

@@ -48,6 +48,19 @@ def test_remove(self):
         w.remove(dummy)
         self.assertEqual(dummy in w, False)
 
+    def test_clear(self):
+        from transaction.weakset import WeakSet
+        w = WeakSet()
+        dummy = Dummy()
+        w.add(dummy)
+        dummy2 = Dummy()
+        w.add(dummy2)
+        self.assertEqual(dummy in w, True)
+        self.assertEqual(dummy2 in w, True)
+        w.clear()
+        self.assertEqual(dummy in w, False)
+        self.assertEqual(dummy2 in w, False)
+
     def test_as_weakref_list(self):
         import gc
         from transaction.weakset import WeakSet

transaction/weakset.py

@@ -50,6 +50,9 @@ def add(self, obj):
     def remove(self, obj):
         del self.data[id(obj)]
 
+    def clear(self):
+        self.data.clear()
+
     # f is a one-argument function.  Execute f(elt) for each elt in the
     # set.  f's return value is ignored.
     def map(self, f):