Back to index

plone3  3.1.7
test_WorkflowTool.py
Go to the documentation of this file.
00001 ##############################################################################
00002 #
00003 # Copyright (c) 2002 Zope Corporation and Contributors. All Rights Reserved.
00004 #
00005 # This software is subject to the provisions of the Zope Public License,
00006 # Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
00007 # THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
00008 # WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
00009 # WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
00010 # FOR A PARTICULAR PURPOSE.
00011 #
00012 ##############################################################################
00013 """ Unit tests for WorkflowTool module.
00014 
00015 $Id: test_WorkflowTool.py 77004 2007-06-24 08:57:54Z yuppie $
00016 """
00017 
00018 import unittest
00019 import Testing
00020 
00021 from OFS.SimpleItem import SimpleItem
00022 from zope.component import adapter
00023 from zope.component import provideHandler
00024 from zope.interface import implements
00025 from zope.testing.cleanup import cleanUp
00026 
00027 from Products.CMFCore.interfaces import IActionRaisedExceptionEvent
00028 from Products.CMFCore.interfaces import IActionSucceededEvent
00029 from Products.CMFCore.interfaces import IActionWillBeInvokedEvent
00030 from Products.CMFCore.interfaces import IContentish
00031 from Products.CMFCore.interfaces import IWorkflowDefinition
00032 
00033 
00034 class Dummy( SimpleItem ):
00035 
00036     def __init__( self, id ):
00037         self._id = id
00038 
00039     def getId( self ):
00040         return self._id
00041 
00042 
00043 class DummyWorkflow( Dummy ):
00044 
00045     implements(IWorkflowDefinition)
00046     meta_type = 'DummyWorkflow'
00047     _isAWorkflow = 1
00048     _known_actions=()
00049     _known_info=()
00050 
00051     def __init__( self, id ):
00052         Dummy.__init__( self, id )
00053         self._did_action = {}
00054         self._gave_info = {}
00055         self._notified = {}
00056 
00057     def setKnownActions( self, known_actions ):
00058         self._known_actions = known_actions
00059 
00060     def setKnownInfo( self, known_info ):
00061         self._known_info = known_info
00062 
00063     def didAction( self, action ):
00064         return self._did_action.setdefault( action, [] )
00065 
00066     def gaveInfo( self, name ):
00067         return self._gave_info.setdefault( name, [] )
00068 
00069     def notified( self, name ):
00070         return self._notified.setdefault( name, [] )
00071 
00072     #
00073     #   WorkflowDefinition interface
00074     #
00075     def getCatalogVariablesFor( self, ob ):
00076         return { 'dummy' : '%s: %s' % ( self.getId(), ob.getId() ) }
00077 
00078     def updateRoleMappingsFor( self, ob ):
00079         pass
00080 
00081     def listObjectActions( self, info ):
00082         return () #XXX
00083 
00084     def listGlobalActions( self, info ):
00085         return () #XXX
00086 
00087     def isActionSupported( self, ob, action ):
00088         return action in self._known_actions
00089 
00090     def doActionFor( self, ob, action, *args, **kw ):
00091         self.didAction( action ).append( ob )
00092 
00093     def isInfoSupported( self, ob, name ):
00094         return name in self._known_info
00095 
00096     def getInfoFor( self, ob, name, default, *args, **kw ):
00097         self.gaveInfo( name ).append( ob )
00098         return name in self._known_info and 1 or 0
00099 
00100     def notifyCreated( self, ob ):
00101         self.notified( 'created' ).append( ( ob, ) )
00102 
00103     def notifyBefore( self, ob, action ):
00104         self.notified( 'before' ).append( ( ob, action ) )
00105 
00106     def notifySuccess( self, ob, action, result ):
00107         self.notified( 'success' ).append( ( ob, action, result ) )
00108 
00109     def notifyException( self, ob, action, exc ):
00110         self.notified( 'exception' ).append( ( ob, action, exc ) )
00111 
00112 @adapter(IActionWillBeInvokedEvent)
00113 def notifyBeforeHandler(evt):
00114     evt.workflow.notified( 'before-evt' ).append( (evt.object, evt.action) )
00115 
00116 @adapter(IActionSucceededEvent)
00117 def notifySuccessHandler(evt):
00118     evt.workflow.notified( 'success-evt' ).append( (evt.object, evt.action,
00119                                                     evt.result ) )
00120 
00121 @adapter(IActionRaisedExceptionEvent)
00122 def notifyExceptionHandler(evt):
00123     evt.workflow.notified( 'exception-evt' ).append( (evt.object, evt.action,
00124                                                       evt.exc) )
00125 
00126 class DummyContent( Dummy ):
00127 
00128     implements(IContentish)
00129     meta_type = 'Dummy'
00130 
00131     def getPortalTypeName(self):
00132         return 'Dummy Content'
00133 
00134 
00135 class DummyNotReallyContent( Dummy ):
00136 
00137     meta_type = 'Dummy Content'
00138 
00139 
00140 class DummyTypeInfo( Dummy ):
00141 
00142     pass
00143 
00144 
00145 class DummyTypesTool( SimpleItem ):
00146 
00147     def listTypeInfo( self ):
00148         return [ DummyTypeInfo( 'Dummy Content' ) ]
00149 
00150     def getTypeInfo( self, ob ):
00151         if getattr( ob, 'meta_type', None ) is 'Dummy':
00152             return DummyTypeInfo( 'Dummy Content' )
00153         return None
00154 
00155 
00156 class WorkflowToolTests(unittest.TestCase):
00157 
00158     def _makeOne( self, workflow_ids=() ):
00159         from Products.CMFCore.WorkflowTool import WorkflowTool
00160 
00161         tool = WorkflowTool()
00162 
00163         for workflow_id in workflow_ids:
00164             tool._setObject(workflow_id, DummyWorkflow(workflow_id))
00165 
00166         return tool
00167 
00168     def _makeRoot( self ):
00169 
00170         from OFS.Folder import Folder
00171         root = Folder( 'root' )
00172         tt = DummyTypesTool()
00173         root._setObject( 'portal_types', tt )
00174         return root
00175 
00176     def _makeWithTypes( self ):
00177         root = self._makeRoot()
00178         return self._makeOne( workflow_ids=( 'a', 'b' ) ).__of__( root )
00179 
00180     def _makeWithTypesAndChain( self ):
00181 
00182         tool = self._makeWithTypes()
00183         tool.setChainForPortalTypes( ( 'Dummy Content', ), ( 'a', 'b' ) )
00184         return tool
00185 
00186     def tearDown(self):
00187         cleanUp()
00188 
00189     def test_z2interfaces(self):
00190         from Interface.Verify import verifyClass
00191         from Products.CMFCore.interfaces.portal_actions \
00192                 import ActionProvider as IActionProvider
00193         from Products.CMFCore.interfaces.portal_workflow \
00194                 import portal_workflow as IWorkflowTool
00195         from Products.CMFCore.WorkflowTool import WorkflowTool
00196 
00197         verifyClass(IActionProvider, WorkflowTool)
00198         verifyClass(IWorkflowTool, WorkflowTool)
00199 
00200     def test_z3interfaces(self):
00201         from zope.interface.verify import verifyClass
00202         from Products.CMFCore.interfaces import IActionProvider
00203         from Products.CMFCore.interfaces import IConfigurableWorkflowTool
00204         from Products.CMFCore.interfaces import IWorkflowTool
00205         from Products.CMFCore.WorkflowTool import WorkflowTool
00206 
00207         verifyClass(IActionProvider, WorkflowTool)
00208         verifyClass(IConfigurableWorkflowTool, WorkflowTool)
00209         verifyClass(IWorkflowTool, WorkflowTool)
00210 
00211     def test_empty( self ):
00212 
00213         from Products.CMFCore.WorkflowTool import WorkflowException
00214 
00215         tool = self._makeOne()
00216 
00217         self.failIf( tool.getWorkflowIds() )
00218         self.assertEqual( tool.getWorkflowById( 'default_workflow' ), None )
00219         self.assertEqual( tool.getWorkflowById( 'a' ), None )
00220 
00221         self.assertRaises( WorkflowException, tool.getInfoFor, None, 'hmm' )
00222         self.assertRaises( WorkflowException, tool.doActionFor, None, 'hmm' )
00223 
00224     def test_new_with_wf( self ):
00225 
00226         from Products.CMFCore.WorkflowTool import WorkflowException
00227 
00228         tool = self._makeWithTypes()
00229 
00230         wfids = tool.getWorkflowIds()
00231         self.assertEqual( len( wfids ), 2 )
00232         self.failUnless( 'a' in wfids )
00233         self.failUnless( 'b' in wfids )
00234         self.assertEqual( tool.getWorkflowById( 'default' ), None )
00235         wf = tool.getWorkflowById( 'a' )
00236         self.assertEqual( wf.getId(), 'a' )
00237         wf = tool.getWorkflowById( 'b' )
00238         self.assertEqual( wf.getId(), 'b' )
00239 
00240         self.assertRaises( WorkflowException, tool.getInfoFor, None, 'hmm' )
00241         self.assertRaises( WorkflowException, tool.doActionFor, None, 'hmm' )
00242 
00243     def test_nonContent( self ):
00244 
00245         tool = self._makeWithTypesAndChain()
00246         self.assertEquals( len( tool.getDefaultChainFor( None ) ), 0 )
00247         self.assertEquals( len( tool.getDefaultChain() ), 1 )
00248         self.assertEquals( len( tool.listChainOverrides() ), 1 )
00249         self.assertEquals( len( tool.getChainFor( None ) ), 0 )
00250         self.assertEquals( len( tool.getCatalogVariablesFor( None ) ), 0 )
00251 
00252     def test_notReallyContent( self ):
00253 
00254         tool = self._makeWithTypesAndChain()
00255         dummy = DummyNotReallyContent( 'doh' )
00256         self.assertEquals( len( tool.getDefaultChainFor( dummy ) ), 0 )
00257         self.assertEquals( len( tool.getDefaultChain() ), 1 )
00258         self.assertEquals( len( tool.listChainOverrides() ), 1 )
00259         self.assertEquals( len( tool.getChainFor( dummy ) ), 0 )
00260         self.assertEquals( len( tool.getCatalogVariablesFor( dummy ) ), 0 )
00261 
00262     def test_content_default_chain( self ):
00263 
00264         tool = self._makeWithTypes()
00265         dummy = DummyContent( 'dummy' )
00266         self.assertEquals( len( tool.getDefaultChainFor( dummy ) ), 1 )
00267         self.assertEquals( len( tool.getDefaultChain() ), 1 )
00268         self.assertEquals( len( tool.listChainOverrides() ), 0 )
00269         self.assertEquals( len( tool.getChainFor( dummy ) ), 1 )
00270         self.assertEquals( len( tool.getCatalogVariablesFor( dummy ) ), 0 )
00271         self.assertEquals( tool.getDefaultChain(), tool.getChainFor( dummy ) )
00272 
00273     def test_content_own_chain( self ):
00274 
00275         tool = self._makeWithTypesAndChain()
00276 
00277         dummy = DummyContent( 'dummy' )
00278 
00279         self.assertEquals( len( tool.getDefaultChainFor( dummy ) ), 1 )
00280         self.assertEquals( len( tool.getDefaultChain() ), 1 )
00281         self.assertEquals( len( tool.listChainOverrides() ), 1 )
00282         chain = tool.getChainFor( dummy )
00283         self.assertEquals( len( chain ), 2 )
00284         self.failUnless( 'a' in chain )
00285         self.failUnless( 'b' in chain )
00286 
00287         vars = tool.getCatalogVariablesFor( dummy )
00288         self.assertEquals( len( vars ), 1 )
00289         self.failUnless( 'dummy' in vars.keys() )
00290         self.failUnless( 'a: dummy' in vars.values() )
00291 
00292     def test_setChainForPortalTypes(self):
00293 
00294         tool = self._makeWithTypes()
00295         tool.setDefaultChain('b, a')
00296         dummy = DummyContent('dummy')
00297 
00298         tool.setChainForPortalTypes( ('Dummy Content',), ('a', 'b') )
00299         self.assertEquals( tool.getChainFor(dummy), ('a', 'b') )
00300         tool.setChainForPortalTypes( ('Dummy Content',), 'a, b' )
00301         self.assertEquals( tool.getChainFor(dummy), ('a', 'b') )
00302 
00303         tool.setChainForPortalTypes( ('Dummy Content',), () )
00304         self.assertEquals( tool.getChainFor(dummy), () )
00305         tool.setChainForPortalTypes( ('Dummy Content',), '' )
00306         self.assertEquals( tool.getChainFor(dummy), () )
00307 
00308         tool.setChainForPortalTypes( ('Dummy Content',), None )
00309         self.assertEquals( tool.getChainFor(dummy), ('b', 'a') )
00310 
00311         # Using the '(Default)' keyword
00312         # http://www.zope.org/Collectors/CMF/475
00313         tool.setChainForPortalTypes( ('Dummy Content',), '(Default)' )
00314         self.assertEquals( tool.getChainFor(dummy), () )
00315 
00316     def test_getCatalogVariablesFor( self ):
00317 
00318         tool = self._makeWithTypesAndChain()
00319         dummy = DummyContent( 'dummy' )
00320 
00321         vars = tool.getCatalogVariablesFor( dummy )
00322         self.assertEquals( len( vars ), 1 )
00323         self.failUnless( 'dummy' in vars.keys() )
00324         self.failUnless( 'a: dummy' in vars.values() )
00325 
00326     def test_getInfoFor( self ):
00327 
00328         tool = self._makeWithTypesAndChain()
00329         tool.b.setKnownInfo( ( 'info', ) )
00330         dummy = DummyContent( 'dummy' )
00331 
00332         info = tool.getInfoFor( dummy, 'info' )
00333 
00334         self.assertEqual( info, 1 )
00335         self.failIf( tool.a.gaveInfo( 'info' ) )
00336         self.failUnless( tool.b.gaveInfo( 'info' ) )
00337 
00338     def test_doActionFor( self ):
00339 
00340         tool = self._makeWithTypesAndChain()
00341         tool.a.setKnownActions( ( 'action', ) )
00342         dummy = DummyContent( 'dummy' )
00343 
00344         tool.doActionFor( dummy, 'action' )
00345 
00346         self.failUnless( tool.a.didAction( 'action' ) )
00347         self.failIf( tool.b.didAction( 'action' ) )
00348 
00349     def test_notifyCreated( self ):
00350 
00351         tool = self._makeWithTypesAndChain()
00352 
00353         ob = DummyContent( 'dummy' )
00354         tool.notifyCreated( ob )
00355 
00356         for wf in tool.a, tool.b:
00357             notified = wf.notified( 'created' )
00358             self.assertEqual( len( notified ), 1 )
00359             self.assertEqual( notified[0], ( ob, ) )
00360 
00361     def test_notifyBefore( self ):
00362 
00363         provideHandler(notifyBeforeHandler)
00364 
00365         tool = self._makeWithTypesAndChain()
00366 
00367         ob = DummyContent( 'dummy' )
00368         tool.notifyBefore( ob, 'action' )
00369 
00370         for wf in tool.a, tool.b:
00371             notified = wf.notified( 'before' )
00372             self.assertEqual( len( notified ), 1 )
00373             self.assertEqual( notified[0], ( ob, 'action' ) )
00374 
00375             notified = wf.notified( 'before-evt' )
00376             self.assertEqual( len( notified ), 1 )
00377             self.assertEqual( notified[0], ( ob, 'action' ) )
00378 
00379     def test_notifySuccess( self ):
00380 
00381         provideHandler(notifySuccessHandler)
00382 
00383         tool = self._makeWithTypesAndChain()
00384 
00385         ob = DummyContent( 'dummy' )
00386         tool.notifySuccess( ob, 'action' )
00387 
00388         for wf in tool.a, tool.b:
00389             notified = wf.notified( 'success' )
00390             self.assertEqual( len( notified ), 1 )
00391             self.assertEqual( notified[0], ( ob, 'action', None ) )
00392 
00393             notified = wf.notified( 'success-evt' )
00394             self.assertEqual( len( notified ), 1 )
00395             self.assertEqual( notified[0], ( ob, 'action', None ) )
00396 
00397     def test_notifyException( self ):
00398 
00399         provideHandler(notifyExceptionHandler)
00400 
00401         tool = self._makeWithTypesAndChain()
00402 
00403         ob = DummyContent( 'dummy' )
00404         tool.notifyException( ob, 'action', 'exception' )
00405 
00406         for wf in tool.a, tool.b:
00407             notified = wf.notified( 'exception' )
00408             self.assertEqual( len( notified ), 1 )
00409             self.assertEqual( notified[0], ( ob, 'action', 'exception' ) )
00410 
00411             notified = wf.notified( 'exception-evt' )
00412             self.assertEqual( len( notified ), 1 )
00413             self.assertEqual( notified[0], ( ob, 'action', 'exception' ) )
00414 
00415     def xxx_test_updateRoleMappings( self ):
00416         """
00417             Build a tree of objects, invoke tool.updateRoleMappings,
00418             and then check to see that the workflows each got called;
00419             check the resulting count, as well.
00420         """
00421 
00422 
00423 def test_suite():
00424     return unittest.TestSuite((
00425         unittest.makeSuite(WorkflowToolTests),
00426         ))
00427 
00428 if __name__ == '__main__':
00429     unittest.main(defaultTest='test_suite')