Back to index

plone3  3.1.7
test_ArchivistTool.py
Go to the documentation of this file.
00001 #########################################################################
00002 # Copyright (c) 2004, 2005 Alberto Berti, Gregoire Weber.
00003 # Reflab (Vincenzo Di Somma, Francesco Ciriaci, Riccardo Lemmi)
00004 # All Rights Reserved.
00005 #
00006 # This file is part of CMFEditions.
00007 #
00008 # CMFEditions is free software; you can redistribute it and/or modify
00009 # it under the terms of the GNU General Public License as published by
00010 # the Free Software Foundation; either version 2 of the License, or
00011 # (at your option) any later version.
00012 #
00013 # CMFEditions is distributed in the hope that it will be useful,
00014 # but WITHOUT ANY WARRANTY; without even the implied warranty of
00015 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
00016 # GNU General Public License for more details.
00017 #
00018 # You should have received a copy of the GNU General Public License
00019 # along with CMFEditions; if not, write to the Free Software
00020 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
00021 #########################################################################
00022 """Test the standard archivist
00023 
00024 $Id: test_ArchivistTool.py,v 1.10 2005/02/25 22:04:00 tomek1024 Exp $
00025 """
00026 
00027 from Products.PloneTestCase import PloneTestCase
00028 PloneTestCase.setupPloneSite()
00029 
00030 from Interface.Verify import verifyObject
00031 
00032 from Products.CMFEditions.interfaces.IArchivist import IArchivist
00033 from Products.CMFEditions.interfaces.IArchivist import IPurgeSupport
00034 
00035 from DummyTools import DummyModifier
00036 from DummyTools import DummyHistoryIdHandler
00037 from DummyTools import MemoryStorage
00038 from DummyTools import notifyModified
00039 from DummyTools import FolderishContentObjectModifier
00040 
00041 
00042 class TestArchivistToolMemoryStorage(PloneTestCase.PloneTestCase):
00043 
00044     def afterSetUp(self):
00045         self.setRoles(['Manager',])
00046         self.portal.acl_users.userFolderAddUser('reviewer', 'reviewer',
00047                                                 ['Manager'], '')
00048         self.portal.invokeFactory('Document', 'doc')
00049         self.portal.invokeFactory('Folder', 'fol')
00050         self.portal.fol.invokeFactory('Document', 'doc1_inside')
00051         self.portal.fol.invokeFactory('Document', 'doc2_inside')
00052         self.portal.fol.invokeFactory('Document', 'doc3_outside')
00053         tools = (
00054             DummyModifier(),
00055             DummyHistoryIdHandler(),
00056             )
00057         for tool in tools:
00058             self._setDummyTool(tool)
00059 
00060         self.installStorageTool()
00061 
00062         # delete purge policy if there is one installed
00063         try:
00064             del self.portal.portal_purgepolicy
00065         except AttributeError:
00066             pass
00067 
00068     def installStorageTool(self):
00069         self._setDummyTool(MemoryStorage())
00070 
00071     def _setDummyTool(self, tool):
00072         setattr(self.portal, tool.getId(), tool)
00073 
00074     def test00_interface(self):
00075         portal_archivist = self.portal.portal_archivist
00076         verifyObject(IArchivist, portal_archivist)
00077         verifyObject(IPurgeSupport, portal_archivist)
00078 
00079     def test01_registerAttachesAHistoryId(self):
00080         portal_archivist = self.portal.portal_archivist
00081         portal_historyidhandler = self.portal.portal_historyidhandler
00082         portal_historiesstorage = self.portal.portal_historiesstorage
00083         doc = self.portal.doc
00084         prep = portal_archivist.prepare(doc, app_metadata='save number 1')
00085         portal_archivist.register(prep)
00086         history_id = portal_historyidhandler.queryUid(doc)
00087         self.failUnless(history_id)
00088 
00089     def test02_retrieve(self):
00090         portal_archivist = self.portal.portal_archivist
00091         portal_historyidhandler = self.portal.portal_historyidhandler
00092         portal_historiesstorage = self.portal.portal_historiesstorage
00093         doc = self.portal.doc
00094         doc.text = 'text v1'
00095         prep = portal_archivist.prepare(doc, app_metadata='save number 1')
00096         portal_archivist.register(prep)
00097         doc.text = 'text v2'
00098         prep = portal_archivist.prepare(doc, app_metadata='save number 2')
00099         portal_archivist.save(prep)
00100         vdata = portal_archivist.retrieve(obj=doc, selector=0, preserve=('gaga', 'gugus'))
00101         retr_doc = vdata.data.object
00102         retr_meta = vdata.app_metadata
00103         doc_histid = portal_historyidhandler.queryUid(doc)
00104         retr_histid = portal_historyidhandler.queryUid(retr_doc)
00105         self.assertEqual(doc_histid, retr_histid)
00106         # check if correct version retrieved and working object unchanged
00107         self.assertEqual(retr_doc.text , 'text v1')
00108         self.assertEqual(retr_meta , 'save number 1')
00109         self.assertEqual(doc.text , 'text v2')
00110         self.assertEqual(len(vdata.preserved_data), 2)
00111         self.assertEqual(vdata.preserved_data['gaga'], 'gaga')
00112         self.assertEqual(vdata.preserved_data['gugus'], 'gugus')
00113 
00114     def test03_retrieveById(self):
00115         portal_archivist = self.portal.portal_archivist
00116         portal_historyidhandler = self.portal.portal_historyidhandler
00117         portal_historiesstorage = self.portal.portal_historiesstorage
00118         doc = self.portal.doc
00119         doc.text = 'text v1'
00120         prep = portal_archivist.prepare(doc, app_metadata='save number 1')
00121         portal_archivist.register(prep)
00122         doc.text = 'text v2'
00123         prep = portal_archivist.prepare(doc, app_metadata='save number 2')
00124         portal_archivist.save(prep)
00125         doc_histid = portal_historyidhandler.queryUid(doc)
00126         vdata = portal_archivist.retrieve(history_id=doc_histid, selector=0, 
00127                                           preserve=('gaga', 'gugus'))
00128         retr_doc = vdata.data.object
00129         retr_meta = vdata.app_metadata
00130         # check if correct version retrieved and working object unchanged
00131         self.assertEqual(retr_doc.text , 'text v1')
00132         self.assertEqual(retr_meta , 'save number 1')
00133         self.assertEqual(doc.text , 'text v2')
00134         self.assertEqual(len(vdata.preserved_data), 2)
00135         self.assertEqual(vdata.preserved_data['gaga'], 'gaga')
00136         self.assertEqual(vdata.preserved_data['gugus'], 'gugus')
00137 
00138     def test04_getHistory(self):
00139         portal_archivist = self.portal.portal_archivist
00140         portal_historyidhandler = self.portal.portal_historyidhandler
00141         portal_historiesstorage = self.portal.portal_historiesstorage
00142         doc = self.portal.doc
00143 
00144         doc.text = 'text v1'
00145         prep = portal_archivist.prepare(doc, app_metadata='save number 1')
00146         portal_archivist.register(prep)
00147 
00148         doc.text = 'text v2'
00149         prep = portal_archivist.prepare(doc, app_metadata='save number 2')
00150         portal_archivist.save(prep)
00151 
00152         history = portal_archivist.getHistory(doc)
00153 
00154         self.assertEqual(len(history), 2)
00155         # check if timestamp and principal available
00156         self.failUnless(history[0].sys_metadata['timestamp'])
00157         self.failUnless(history[0].sys_metadata['principal'])
00158         # check if correct data and metadata retrieved
00159         self.assertEqual(history[0].data.object.text, 'text v1')
00160         self.assertEqual(history[0].app_metadata, 'save number 1')
00161         self.assertEqual(history[1].data.object.text, 'text v2')
00162         self.assertEqual(history[1].app_metadata, 'save number 2')
00163 
00164     def test05_iterateOverHistory(self):
00165         portal_archivist = self.portal.portal_archivist
00166         doc = self.portal.doc
00167 
00168         doc.text = 'text v1'
00169         prep = portal_archivist.prepare(doc, app_metadata='save number 1')
00170         portal_archivist.register(prep)
00171 
00172         doc.text = 'text v2'
00173         prep = portal_archivist.prepare(doc, app_metadata='save number 2')
00174         portal_archivist.save(prep)
00175 
00176         doc.text = 'text v3'
00177         prep = portal_archivist.prepare(doc, app_metadata='save number 3')
00178         portal_archivist.save(prep)
00179 
00180         counter = 0
00181         
00182         for vdata in portal_archivist.getHistory(doc):
00183             counter += 1
00184             self.assertEqual(vdata.data.object.text, 'text v%s' % counter)
00185             self.assertEqual(vdata.app_metadata, 'save number %s' % counter)
00186 
00187     def test06_getHistoryById(self):
00188         portal_archivist = self.portal.portal_archivist
00189         portal_historyidhandler = self.portal.portal_historyidhandler
00190         portal_historiesstorage = self.portal.portal_historiesstorage
00191         doc = self.portal.doc
00192 
00193         doc.text = 'text v1'
00194         prep = portal_archivist.prepare(doc, app_metadata='save number 1')
00195         portal_archivist.register(prep)
00196 
00197         doc.text = 'text v2'
00198         prep = portal_archivist.prepare(doc, app_metadata='save number 2')
00199         portal_archivist.save(prep)
00200 
00201         doc_histid = portal_historyidhandler.queryUid(doc)
00202         history = portal_archivist.getHistory(history_id=doc_histid)
00203 
00204         self.assertEqual(len(history), 2)
00205         # check if timestamp and principal available
00206         self.failUnless(history[0].sys_metadata['timestamp'])
00207         self.failUnless(history[0].sys_metadata['principal'])
00208         # check if correct data and metadata retrieved
00209         self.assertEqual(history[0].data.object.text, 'text v1')
00210         self.assertEqual(history[0].app_metadata, 'save number 1')
00211         self.assertEqual(history[1].data.object.text, 'text v2')
00212         self.assertEqual(history[1].app_metadata, 'save number 2')
00213 
00214     def test07_prepareObjectWithReferences(self):
00215         # test with a different modifier
00216         self._setDummyTool(FolderishContentObjectModifier())
00217 
00218         portal_archivist = self.portal.portal_archivist
00219         portal_hidhandler = self.portal.portal_historyidhandler
00220         IVersionAwareReference = portal_archivist.interfaces.IVersionAwareReference
00221         fol = self.portal.fol
00222         fol.title = "BLOB title ..."
00223         doc1_inside = fol.doc1_inside
00224         doc2_inside = fol.doc2_inside
00225         doc3_outside = fol.doc3_outside
00226 
00227         doc1_inside.text = 'doc1_inside: inside reference'
00228         doc2_inside.text = 'doc2_inside: inside reference'
00229         doc3_outside.text = 'doc3_outside: outside reference'
00230 
00231         prep = portal_archivist.prepare(fol, app_metadata='save number 1')
00232 
00233         self.assertEqual(fol, prep.original.object)
00234 
00235         # it is important that the clones returned reference info contain
00236         # references to the outgoing references and the python refs are
00237         # replaced by 'IVersionAwareRefrence' objects
00238         inside_refs = prep.clone.inside_refs
00239         outside_refs = prep.clone.outside_refs
00240         self.assertEqual(len(inside_refs), 2)
00241         self.assertEqual(len(outside_refs), 1)
00242         refs = [ref.getAttribute() for ref in inside_refs+outside_refs]
00243         for ref in refs:
00244             self.failUnless(IVersionAwareReference.isImplementedBy(ref))
00245         cloneValues = prep.clone.object.objectValues()
00246         for sub in cloneValues:
00247             self.failUnless(sub in refs)
00248 
00249         # it is important that the originals returned reference info contain
00250         # references to the outgoing references
00251         inside_orefs = prep.original.inside_refs
00252         outside_orefs = prep.original.outside_refs
00253         self.assertEqual(len(inside_orefs), 2)
00254         self.assertEqual(len(outside_orefs), 1)
00255         refs = inside_orefs+outside_orefs
00256         originalValues = prep.original.object.objectValues()
00257 
00258         for sub in originalValues:
00259             self.failUnless(sub in refs)
00260 
00261         # the clones and the originals refs must also reference the
00262         # "same" object
00263         self.assertEqual(prep.clone.object.objectIds(),
00264                          prep.original.object.objectIds())
00265 
00266         self.assertEqual(len(prep.referenced_data), 1)
00267         self.failUnless(prep.referenced_data['title'] is fol.title)
00268         
00269         self.assertEqual(prep.metadata['app_metadata'], 'save number 1')
00270         self.failUnless('timestamp' in prep.metadata['sys_metadata'])
00271         self.failUnless('principal' in prep.metadata['sys_metadata'])
00272         
00273         self._setDummyTool(DummyModifier())
00274 
00275     def test08_retrieveWithReferences(self):
00276         # test with a different modifier
00277         self._setDummyTool(FolderishContentObjectModifier())
00278         
00279         portal_archivist = self.portal.portal_archivist
00280         portal_hidhandler = self.portal.portal_historyidhandler
00281         IVersionAwareReference = portal_archivist.interfaces.IVersionAwareReference
00282         fol = self.portal.fol
00283         fol.title = "BLOB title ..."
00284         doc1_inside = fol.doc1_inside
00285         doc2_inside = fol.doc2_inside
00286         doc3_outside = fol.doc3_outside
00287         
00288         doc1_inside.text = 'doc1_inside: inside reference'
00289         doc2_inside.text = 'doc2_inside: inside reference'
00290         doc3_outside.text = 'doc3_outside: outside reference'
00291         
00292         prep = portal_archivist.prepare(fol, app_metadata='save number 1')
00293         
00294         # just set the info to some value before save to test if the 
00295         # reference stuff is saved and retrieved correctly
00296         inside_refs = prep.clone.inside_refs
00297         outside_refs = prep.clone.outside_refs
00298         refs = [ref.getAttribute() for ref in inside_refs+outside_refs]
00299         for ref in refs:
00300             ref.info = refs.index(ref)
00301         
00302         portal_archivist.save(prep, autoregister=True)
00303         
00304         retr = portal_archivist.retrieve(fol)
00305         
00306         # check metadata
00307         self.assertEqual(retr.app_metadata, 'save number 1')
00308         self.failUnless('timestamp' in retr.sys_metadata)
00309         self.failUnless('principal' in retr.sys_metadata)
00310         
00311         # check the references
00312         inside_refs = retr.data.inside_refs
00313         outside_refs = retr.data.outside_refs
00314         self.assertEqual(len(inside_refs), 2)
00315         self.assertEqual(len(outside_refs), 1)
00316         refs = [ref.getAttribute() for ref in inside_refs+outside_refs]
00317         for ref in refs:
00318             self.failUnless(IVersionAwareReference.isImplementedBy(ref))
00319             # check info value (see note above)
00320             self.assertEquals(ref.info, refs.index(ref))
00321 
00322     def test09_isUpToDate(self):
00323         doc = self.portal.doc
00324         portal_archivist = self.portal.portal_archivist
00325         doc.text = 'text v1'
00326         prep = portal_archivist.prepare(doc, app_metadata='save number 1')
00327         v1 = portal_archivist.register(prep)
00328 
00329         self.failUnless(portal_archivist.isUpToDate(obj=doc))
00330         self.failUnless(portal_archivist.isUpToDate(obj=doc, selector=v1))
00331 
00332         doc.text = 'text v2'
00333         notifyModified(doc)
00334         self.failIf(portal_archivist.isUpToDate(obj=doc))
00335 
00336         prep = portal_archivist.prepare(doc, app_metadata='save number 2')
00337         v2 = portal_archivist.save(prep)
00338 
00339         self.failUnless(portal_archivist.isUpToDate(obj=doc))
00340         self.failUnless(portal_archivist.isUpToDate(obj=doc, selector=v2))
00341         self.failIf(portal_archivist.isUpToDate(obj=doc, selector=v1))
00342 
00343 
00344 class TestArchivistToolZStorage(TestArchivistToolMemoryStorage):
00345 
00346    def installStorageTool(self):
00347        """Test with a real ZODB storage overriding the storage installation
00348           in the super class.
00349        """
00350        pass
00351 
00352 
00353 from unittest import TestSuite, makeSuite
00354 def test_suite():
00355     suite = TestSuite()
00356     suite.addTest(makeSuite(TestArchivistToolMemoryStorage))
00357     suite.addTest(makeSuite(TestArchivistToolZStorage))
00358     return suite