[Zope-Checkins] CVS: Zope/lib/python/Products/ZCatalog/tests - testBrains.py:1.1.2.1 testCatalog.py:1.22.12.4.4.1

Casey Duncan casey at zope.com
Tue Mar 23 14:57:24 EST 2004


Update of /cvs-repository/Zope/lib/python/Products/ZCatalog/tests
In directory cvs.zope.org:/tmp/cvs-serv24183/tests

Modified Files:
      Tag: casey-brains-eat-conflicts-branch
	testCatalog.py 
Added Files:
      Tag: casey-brains-eat-conflicts-branch
	testBrains.py 
Log Message:
Reduce catalog brains tendancy to eat conflict errors which could result in random loss of valid results in high-concurrency situations. This problem is especially bad for applications (like CMF) that rely on catalog results as a way to get a list of objects to act on. Specific refactorings:

- CatalogBrains getPath() and getObject() calls will propagate ConflictErrors
  properly.

- getObject() better respects security settings on search results and returns
  None for results that cannot be traversed.

- CatalogBrains now have their own unit tests

- Tone down the Catalog range tests from 10,000 ranges to 1,000 ranges. The
  former took way too long. Note: this test uses random ranges which is
  dubious since it's not really repeatable, I'm punting on fixing that now...


=== Added File Zope/lib/python/Products/ZCatalog/tests/testBrains.py ===
##############################################################################
#
# Copyright (c) 2001 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
#
##############################################################################
"""Unittests for Catalog brains

$Id: testBrains.py,v 1.1.2.1 2004/03/23 19:57:23 caseman Exp $"""

import unittest
import Acquisition
from zExceptions import Unauthorized
from ZODB.POSException import ConflictError

class Happy(Acquisition.Implicit):
    """Happy content"""
    def __init__(self, id):
        self.id = id

class Secret(Happy):
    """Object that raises Unauthorized when accessed"""
    def __of__(self, parent):
        raise Unauthorized

class Conflicter(Happy):
    """Object that raises ConflictError when accessed"""
    def __of__(self, parent):
        raise ConflictError
        
class DummyRequest(Acquisition.Implicit):
    
    def physicalPathToURL(self, path, relative=False):
        if not relative:
            path = 'http://superbad.com' + path
        return path

_marker = object()
        
class DummyCatalog(Acquisition.Implicit):
    
    _objs = {'/happy':Happy('happy'), 
             '/secret':Secret('secret'), 
             '/conflicter':Conflicter('conflicter')}
    _paths = _objs.keys() + ['/zonked']
    _paths.sort()
    
    def restrictedTraverse(self, path, default=_marker):
        try:
            return self._objs[path].__of__(self)
        except (KeyError, Unauthorized):
            if default is not _marker:
                return default
            raise
    
    def getpath(self, rid):
        return self._paths[rid]
    
    def getobject(self, rid):
        return self.restrictedTraverse(self._paths[rid])

    def resolve_url(self, path, REQUEST):
        path =  path[path.find('/', path.find('//')+1):] # strip server part
        return self.restrictedTraverse(path)
        
class ConflictingCatalog(DummyCatalog):
    
    def getpath(self, rid):
        raise ConflictError

class TestBrains(unittest.TestCase):
    
    def setUp(self):
        self.cat = DummyCatalog()
        self.cat.REQUEST = DummyRequest()
    
    def makeBrain(self, rid):
        from Products.ZCatalog.CatalogBrains import AbstractCatalogBrain
        class Brain(AbstractCatalogBrain):
            __record_schema__ = {'test_field': 0, 'data_record_id_':1}
        return Brain(('test', rid)).__of__(self.cat)
    
    def testHasKey(self):
        b = self.makeBrain(1)
        self.failUnless(b.has_key('test_field'))
        self.failUnless(b.has_key('data_record_id_'))
        self.failIf(b.has_key('godel'))
    
    def testGetPath(self):
        b = [self.makeBrain(rid) for rid in range(3)]
        self.assertEqual(b[0].getPath(), '/conflicter')
        self.assertEqual(b[1].getPath(), '/happy')
        self.assertEqual(b[2].getPath(), '/secret')
    
    def testGetPathPropagatesConflictErrors(self):
        self.cat = ConflictingCatalog()
        b = self.makeBrain(0)
        self.assertRaises(ConflictError, b.getPath)
        
    def testGetURL(self):
        b = self.makeBrain(0)
        self.assertEqual(b.getURL(), 'http://superbad.com/conflicter')
    
    def testGetRID(self):
        b = self.makeBrain(42)
        self.assertEqual(b.getRID(), 42)
    
    def testGetObjectHappy(self):
        b = self.makeBrain(1)
        self.assertEqual(b.getPath(), '/happy')
        self.failUnless(b.getObject().aq_base is self.cat.getobject(1).aq_base)
    
    def testGetObjectPropagatesConflictErrors(self):
        b = self.makeBrain(0)
        self.assertEqual(b.getPath(), '/conflicter')
        self.assertRaises(ConflictError, b.getObject)
    
    def testGetObjectReturnsNoneForUnauthorized(self):
        b = self.makeBrain(2)
        self.assertEqual(b.getPath(), '/secret')
        self.assertEqual(b.getObject(), None)
    
    def testGetObjectReturnsNoneForMissing(self):
        b = self.makeBrain(3)
        self.assertEqual(b.getPath(), '/zonked')
        self.assertRaises(KeyError, self.cat.getobject, 3)
        self.assertEqual(b.getObject(), None)        

def test_suite():
    suite = unittest.TestSuite()
    suite.addTest(unittest.makeSuite(TestBrains))
    return suite

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


=== Zope/lib/python/Products/ZCatalog/tests/testCatalog.py 1.22.12.4 => 1.22.12.4.4.1 ===
--- Zope/lib/python/Products/ZCatalog/tests/testCatalog.py:1.22.12.4	Thu Jan  8 02:47:52 2004
+++ Zope/lib/python/Products/ZCatalog/tests/testCatalog.py	Tue Mar 23 14:57:23 2004
@@ -1,6 +1,18 @@
-#!/usr/bin/env python, unittest
+##############################################################################
+#
+# Copyright (c) 2001 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
+#
+##############################################################################
+"""Unittests for Catalog
 
-# Unittests for Catalog
+$Id$"""
 
 import os
 import random
@@ -462,7 +474,7 @@
         self._catalog.aq_parent = objRS(200)
 
     def testRangeSearch(self):
-        for i in range(10000):
+        for i in range(1000):
             m = random.randrange(0,20000)
             n = m + 1000
 




More information about the Zope-Checkins mailing list