source: products/quintagroup.catalogupdater/trunk/quintagroup/catalogupdater/tests.py @ 1765

Last change on this file since 1765 was 1765, checked in by mylan, 14 years ago

Added tests for subtransactions

  • Property svn:eol-style set to native
File size: 4.7 KB
Line 
1import unittest
2import transaction
3
4from zope.interface import Interface
5from zope.component import queryUtility
6from zope.component import provideAdapter
7
8from plone.indexer.decorator import indexer
9from plone.indexer.interfaces import IIndexableObject
10from Testing import ZopeTestCase as ztc
11
12from Products.Five import zcml
13from Products.Five import fiveconfigure
14from Products.CMFCore.utils import getToolByName
15from Products.PloneTestCase import PloneTestCase as ptc
16from Products.PloneTestCase.layer import PloneSite
17from Products.Archetypes.tests.utils import makeContent
18
19from quintagroup.catalogupdater.utility import ICatalogUpdater
20
21class TestCase(ptc.PloneTestCase):
22    class layer(PloneSite):
23        @classmethod
24        def setUp(cls):
25            import quintagroup.catalogupdater
26            fiveconfigure.debug_mode = True
27            zcml.load_config('configure.zcml', quintagroup.catalogupdater)
28            fiveconfigure.debug_mode = False
29
30ptc.setupPloneSite()
31
32
33class TestUtility(TestCase):
34
35    def afterSetUp(self):
36        self.loginAsPortalOwner()
37        self.my_doc = makeContent(self.portal, portal_type='Document', id='my_doc')
38        self.catalog = getToolByName(self.portal, 'portal_catalog')
39        self.logout()
40        self.addIndexer()
41
42    def addIndexer(self):
43        @indexer(Interface)
44        def test_column(obj):
45            return obj.id
46
47        provideAdapter(test_column, name='test_column')
48        self.catalog.addColumn('test_column')
49
50    def testSingleColumnUpdate(self):
51        """ Test is metadata column updated with utility
52        """
53        mydoc = self.catalog.unrestrictedSearchResults(id='my_doc')[0]
54        self.assertFalse(mydoc.test_column == 'my_doc',
55            "'test_column' metadata updated in catalog " \
56            "before utility call: '%s'" % mydoc.test_column)
57
58        cu = queryUtility(ICatalogUpdater, name="catalog_updater")
59        cu.updateMetadata4All(self.catalog, 'test_column')
60
61        mydoc = self.catalog.unrestrictedSearchResults(id='my_doc')[0]
62        self.assertTrue(mydoc.test_column == 'my_doc',
63            "'test_column' metadata has wrong metadata in catalog: " \
64            "'%s'" % mydoc.test_column)
65
66    def testOnlyPointedColumnUpdate(self):
67        """ Update a title property for the my_doc object
68            (without reindexing) then, after utility usage
69            - check is that metadata is leave unchanged.
70        """
71        self.loginAsPortalOwner()
72        self.my_doc.update(title="My document")
73        self.logout()
74
75        mydoc = self.catalog.unrestrictedSearchResults(id='my_doc')[0]
76        self.assertTrue(mydoc.Title == "My document", mydoc.Title)
77
78        self.my_doc.setTitle('New my document') # catalog not updated
79        cu = queryUtility(ICatalogUpdater, name="catalog_updater")
80        cu.updateMetadata4All(self.catalog, 'test_column')
81
82        mydoc = self.catalog.unrestrictedSearchResults(id='my_doc')[0]
83        self.assertTrue(mydoc.Title == 'My document',
84            "Other metadata updated: Title='%s'" % mydoc.Title)
85
86
87    def testAllRecordsUpdate(self):
88        """ Test is all records in catalog updated with utility
89        """
90        cu = queryUtility(ICatalogUpdater, name="catalog_updater")
91        cu.updateMetadata4All(self.catalog, 'test_column')
92
93        num_recs = len(self.catalog._catalog.data)
94        allcat = self.catalog.unrestrictedSearchResults(path='/')
95        num_updated = sum([1 for b in allcat if b.test_column==b.id])
96
97        self.assertTrue(num_updated == num_recs, "Only %d records updated, " \
98            "must be - %d" % (num_updated, num_recs))
99
100
101    def testTransaction(self):
102        """ Test is commited subtransactions
103        """
104        # savepoint patch
105        global sp_commits
106        sp_commits = 1 # Starts from 1 to count last commit
107        orig_trsp = transaction.savepoint
108        def dummy_savepoint(*args, **kwargs):
109            global sp_commits
110            sp_commits += 1
111            orig_trsp(*args, **kwargs)
112        transaction.savepoint = dummy_savepoint
113
114        # set threshold for catalog
115        num_recs = len(self.catalog.unrestrictedSearchResults(path='/'))
116        num_subcommits = 3
117        self.catalog.threshold = num_recs/num_subcommits
118
119        cu = queryUtility(ICatalogUpdater, name="catalog_updater")
120        cu.updateMetadata4All(self.catalog, 'test_column')
121
122        self.assertTrue(sp_commits == num_subcommits,
123            "Wrong number of transaction subcommits: actual:%d, must be: %d" % (
124            sp_commits, num_subcommits))
125
126        transaction.savepoint = orig_trsp
127       
128
129
130def test_suite():
131    return unittest.TestSuite([
132        unittest.makeSuite(TestUtility),
133        ])
134
135if __name__ == '__main__':
136    unittest.main(defaultTest='test_suite')
Note: See TracBrowser for help on using the repository browser.