Separate ConnectionRegistry out into its own module so it can be

reused more cleanly.

lib/python/Testing/ZopeTestCase/ZopeTestCase.py

@@ -21,13 +21,14 @@ The fixture consists of:
 The default user is logged in and has the 'Access contents information'
 and 'View' permissions given to his role.
 
-$Id: ZopeTestCase.py,v 1.29 2005/02/09 12:42:40 shh42 Exp $
+$Id$
 """
 
 import base
 import functional
 import interfaces
 import utils
+import connections
 
 from AccessControl import getSecurityManager
 from AccessControl.SecurityManagement import newSecurityManager
@@ -80,7 +81,7 @@ class ZopeTestCase(base.TestCase):
         '''Clears the fixture.'''
         # This code is a wart from the olden days.
         try:
-            if base._connections.contains(self.app._p_jar):
+            if connections.contains(self.app._p_jar):
                 self.app._delObject(folder_name)
         except:
             pass
@@ -121,5 +122,4 @@ class FunctionalTestCase(functional.Functional, ZopeTestCase):
 
 from base import app
 from base import close
-from base import closeConnections
 

lib/python/Testing/ZopeTestCase/base.py

@@ -12,7 +12,7 @@
 ##############################################################################
 """TestCase for Zope testing
 
-$Id: base.py,v 1.1 2004/08/19 13:59:41 shh42 Exp $
+$Id$
 """
 
 import ZopeLite as Zope2
@@ -22,26 +22,20 @@ import transaction
 import profiler
 import utils
 import interfaces
+import connections
 
 from AccessControl.SecurityManagement import noSecurityManager
 
-_connections = utils.ConnectionRegistry()
-
-
 
 def app():
     '''Opens a ZODB connection and returns the app object.'''
     app = Zope2.app()
-    _connections.register(app._p_jar)
+    connections.register(app._p_jar)
     return utils.makerequest(app)
 
 def close(app):
     '''Closes the app's ZODB connection.'''
-    _connections.close(app._p_jar)
-
-def closeConnections():
-    '''Closes all registered ZODB connections.'''
-    _connections.closeAll()
+    connections.close(app._p_jar)
 
 
 
@@ -131,7 +125,7 @@ class TestCase(profiler.Profiled, unittest.TestCase):
     def _close(self):
         '''Closes the ZODB connection.'''
         transaction.abort()
-        closeConnections()
+        connections.closeAll()
 
     def logout(self):
         '''Logs out.'''

lib/python/Testing/ZopeTestCase/connections.py

+##############################################################################
+#
+# Copyright (c) 2005 Zope Corporation and Contributors. All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""ZODB connection registry
+
+$Id$
+"""
+
+class ConnectionRegistry:
+    '''ZODB connection registry'''
+
+    def __init__(self):
+        self._conns = []
+
+    def register(self, conn):
+        self._conns.append(conn)
+
+    def contains(self, conn):
+        return conn in self._conns
+
+    def __len__(self):
+        return len(self._conns)
+
+    def count(self):
+        return len(self)
+
+    def close(self, conn):
+        if self.contains(conn):
+            self._conns.remove(conn)
+        conn.close()
+
+    def closeAll(self):
+        for conn in self._conns:
+            conn.close()
+        self._conns = []
+
+
+registry = ConnectionRegistry()
+register = registry.register
+contains = registry.contains
+count = registry.count
+close = registry.close
+closeAll = registry.closeAll
+

lib/python/Testing/ZopeTestCase/doc/CHANGES.txt

@@ -13,6 +13,8 @@
 - Fixed _refreshSkinData() helper to work with CMF >= 1.5.
 - Fixed a bug where using sessions in sandboxed (functional) tests would cause
   connection pool depletion and subsequent hangs. Thanks to Balazs Ree.
+- Encapsulated the ConnectionRegistry in its own module, connections.py.
+  Reusing the registry from other modules becomes a lot cleaner as a result.
 
 0.9.6
 - Dropped support for Zope 2.5 as it lacks the setSecurityManager() API.

lib/python/Testing/ZopeTestCase/sandbox.py

@@ -12,13 +12,14 @@
 ##############################################################################
 """Support for ZODB sandboxes in ZTC
 
-$Id: sandbox.py,v 1.2 2004/08/19 15:31:26 shh42 Exp $
+$Id$
 """
 
 import ZopeLite as Zope2
 import transaction
 import base
 import utils
+import connections
 
 
 class Sandboxed:
@@ -32,7 +33,7 @@ class Sandboxed:
     def _app(self):
         '''Returns the app object for a test.'''
         app = Zope2.app(Zope2.sandbox().open())
-        base._connections.register(app._p_jar)
+        connections.register(app._p_jar)
         AppZapper().set(app)
         return utils.makerequest(app)
 
@@ -40,7 +41,7 @@ class Sandboxed:
         '''Clears the transaction and the AppZapper.'''
         transaction.abort()
         AppZapper().clear()
-        base.closeConnections()
+        connections.closeAll()
 
 
 class AppZapper:

lib/python/Testing/ZopeTestCase/testBaseTestCase.py

@@ -19,7 +19,7 @@ See testPythonScript.py and testShoppingCart.py for
 example test cases. See testSkeleton.py for a quick
 way of getting started.
 
-$Id: testBaseTestCase.py,v 1.7 2005/02/09 12:42:40 shh42 Exp $
+$Id$
 """
 
 import os, sys
@@ -30,6 +30,7 @@ import transaction
 
 from Testing.ZopeTestCase import base
 from Testing.ZopeTestCase import utils
+from Testing.ZopeTestCase import connections
 
 from AccessControl import getSecurityManager
 from AccessControl.SecurityManagement import newSecurityManager
@@ -80,9 +81,9 @@ class TestTestCase(HookTest):
         self.assertHooks(['beforeTearDown', 'beforeClose', 'afterClear'])
 
     def testAppOpensConnection(self):
-        self.assertEqual(len(base._connections), 1)
+        self.assertEqual(connections.count(), 1)
         self._app()
-        self.assertEqual(len(base._connections), 2)
+        self.assertEqual(connections.count(), 2)
 
     def testClearCallsCloseHook(self):
         self._called = []
@@ -102,15 +103,15 @@ class TestTestCase(HookTest):
         self.assertEqual(len(self.getObjectsInTransaction()), 0)
 
     def testClearClosesConnection(self):
-        self.assertEqual(len(base._connections), 1)
+        self.assertEqual(connections.count(), 1)
         self._clear()
-        self.assertEqual(len(base._connections), 0)
+        self.assertEqual(connections.count(), 0)
 
     def testClearClosesAllConnections(self):
         self._app()
-        self.assertEqual(len(base._connections), 2)
+        self.assertEqual(connections.count(), 2)
         self._clear()
-        self.assertEqual(len(base._connections), 0)
+        self.assertEqual(connections.count(), 0)
 
     def testClearLogsOut(self):
         uf = self.app.acl_users
@@ -128,15 +129,15 @@ class TestTestCase(HookTest):
         self.assertEqual(len(self.getObjectsInTransaction()), 0)
 
     def testCloseClosesConnection(self):
-        self.assertEqual(len(base._connections), 1)
+        self.assertEqual(connections.count(), 1)
         self._close()
-        self.assertEqual(len(base._connections), 0)
+        self.assertEqual(connections.count(), 0)
 
     def testCloseClosesAllConnections(self):
         self._app()
-        self.assertEqual(len(base._connections), 2)
+        self.assertEqual(connections.count(), 2)
         self._close()
-        self.assertEqual(len(base._connections), 0)
+        self.assertEqual(connections.count(), 0)
 
     def testLogoutLogsOut(self):
         uf = self.app.acl_users
@@ -167,7 +168,7 @@ class TestSetUpRaises(HookTest):
         except self.Error:
             self.assertHooks(['beforeSetUp', '_setup', 'afterClear'])
             # Connection has been closed
-            self.assertEqual(len(base._connections), 0)
+            self.assertEqual(connections.count(), 0)
 
     def _setup(self):
         HookTest._setup(self)
@@ -188,7 +189,7 @@ class TestTearDownRaises(HookTest):
         except self.Error:
             self.assertHooks(['beforeTearDown', 'beforeClose', 'afterClear'])
             # Connection has been closed
-            self.assertEqual(len(base._connections), 0)
+            self.assertEqual(connections.count(), 0)
 
     def beforeClose(self):
         HookTest.beforeClose(self)
@@ -206,7 +207,7 @@ class TestConnectionRegistry(base.TestCase):
             self.closed = 1
 
     def afterSetUp(self):
-        self.reg = utils.ConnectionRegistry()
+        self.reg = connections.ConnectionRegistry()
         self.conns = [self.Conn(), self.Conn(), self.Conn()]
 
     def testRegister(self):
@@ -214,6 +215,7 @@ class TestConnectionRegistry(base.TestCase):
         for conn in self.conns:
             self.reg.register(conn)
         assert len(self.reg) == 3
+        assert self.reg.count() == 3
 
     def testCloseConnection(self):
         # Should be able to close a single registered connection

lib/python/Testing/ZopeTestCase/testPortalTestCase.py

@@ -19,7 +19,7 @@ See testPythonScript.py and testShoppingCart.py for
 example test cases. See testSkeleton.py for a quick
 way of getting started.
 
-$Id: testPortalTestCase.py,v 1.30 2005/01/30 14:22:48 shh42 Exp $
+$Id$
 """
 
 import os, sys
@@ -508,8 +508,8 @@ class TestSetUpRaises(HookTest):
         except self.Error:
             self.assertHooks(['beforeSetUp', '_setup', 'afterClear'])
             # Connection has been closed
-            from Testing.ZopeTestCase import base
-            self.assertEqual(len(base._connections), 0)
+            from Testing.ZopeTestCase import connections
+            self.assertEqual(connections.count(), 0)
 
     def _setup(self):
         HookTest._setup(self)

lib/python/Testing/ZopeTestCase/utils.py

@@ -15,7 +15,7 @@
 These functions are designed to be imported and run at
 module level to add functionality to the test environment.
 
-$Id: utils.py,v 1.21 2005/02/11 09:00:21 shh42 Exp $
+$Id$
 """
 
 import os
@@ -163,32 +163,6 @@ def makelist(arg):
     raise ValueError('Argument must be list, tuple, or string')
 
 
-class ConnectionRegistry:
-    '''ZODB connection registry'''
-
-    def __init__(self):
-        self._conns = []
-
-    def register(self, conn):
-        self._conns.append(conn)
-
-    def close(self, conn):
-        if self.contains(conn):
-            self._conns.remove(conn)
-        conn.close()
-
-    def closeAll(self):
-        for conn in self._conns:
-            conn.close()
-        self._conns = []
-
-    def __len__(self):
-        return len(self._conns)
-
-    def contains(self, conn):
-        return conn in self._conns
-
-
 __all__ = [
     'setupCoreSessions',
     'setupSiteErrorLog',