[Zope-CVS] CVS: Products/DBTab/tests - __init__.py:1.1 test.conf:1.1 testDBTab.py:1.1

[Shane Hathaway]

Update of /cvs-repository/Products/DBTab/tests
In directory cvs.zope.org:/tmp/cvs-serv28232/tests

Added Files:
	__init__.py test.conf testDBTab.py 
Log Message:
Fixed a race condition on connection close.

The root connection was returned to the pool before the mounted
connections were closed.  If another thread pulled the root connection
out of the pool before the original thread finished closing mounted
connections, when the original thread got control back it closed the
mounted connections even though the new thread was using them.  The
symptom was spurious "Should not load state when connection closed"
errors under high load.

Also began unit tests.



=== Added File Products/DBTab/tests/__init__.py ===
"""DBTab tests package"""


=== Added File Products/DBTab/tests/test.conf ===

# DBTab configuration for testing purposes

[Storage: Main]
type=DemoStorage

[Database: Main]
mount_paths=/


[Storage: Mount1]
type=DemoStorage

[Database: Mount1]
mount_paths=/mount1
container_class=OFS.Folder.Folder


[Storage: Mount2]
type=DemoStorage

[Database: Mount2]
mount_paths=/mount2
container_class=OFS.Folder.Folder




=== Added File Products/DBTab/tests/testDBTab.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (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
#
##############################################################################
"""Tests of DBTab
"""

import os
import sys
import unittest
import Testing
import ZODB
from OFS.Application import Application
from Products.DBTab.DBTab import DBTab
from Products.DBTab.MountedObject import setConfiguration, manage_addMounts

try:
    __file__
except NameError:
    __file__ = os.path.abspath(sys.argv[0])


class DBTabTests (unittest.TestCase):

    def setUp(self):
        conf = DBTab([os.path.join(os.path.dirname(__file__), 'test.conf')])
        setConfiguration(conf)
        self.conf = conf
        db = conf.getDatabase('/')
        self.db = db
        conn = db.open()
        root = conn.root()
        root['Application'] = app = Application()
        self.app = app
        get_transaction().commit()  # Get app._p_jar set
        manage_addMounts(app, ('/mount1', '/mount2'))
        get_transaction().commit()  # Get the mount points ready


    def tearDown(self):
        setConfiguration(None)
        get_transaction().abort()
        self.app._p_jar.close()
        del self.app
        del self.db
        for db in self.conf.opened.values():
            db.close()
        del self.conf


    def testRead(self):
        self.assertEqual(self.app.mount1.id, 'mount1')
        self.assertEqual(self.app.mount2.id, 'mount2')


    def testWrite(self):
        app = self.app
        app.mount1.a1 = '1'
        app.mount2.a2 = '2'
        app.a3 = '3'
        self.assertEqual(app.mount1._p_changed, 1)
        self.assertEqual(app.mount2._p_changed, 1)
        self.assertEqual(app._p_changed, 1)
        get_transaction().commit()
        self.assertEqual(app.mount1._p_changed, 0)
        self.assertEqual(app.mount2._p_changed, 0)
        self.assertEqual(app._p_changed, 0)


    def testRaceOnClose(self):
        # There used to be a race condition in
        # ConnectionPatches.close().  The root connection was returned
        # to the pool before the mounted connections were closed.  If
        # another thread pulled the root connection out of the pool
        # before the original thread finished closing mounted
        # connections, when the original thread got control back it
        # closed the mounted connections even though the new thread
        # was using them.

        # Test by patching to watch for a vulnerable moment.

        from ZODB.DB import DB

        def _closeConnection(self, connection):
            self._real_closeConnection(connection)
            mc = connection._mounted_connections
            if mc is not None:
                for c in mc.values():
                    if c._storage is not None:
                        raise AssertionError, "Connection remained partly open"

        DB._real_closeConnection = DB._closeConnection
        DB._closeConnection = _closeConnection
        try:
            conn = self.db.open()
            conn.root()['Application']['mount1']
            conn.root()['Application']['mount2']
            conn.close()
        finally:
            DB._closeConnection = DB._real_closeConnection
            del DB._real_closeConnection


if __name__ == '__main__':
    unittest.main()