Back to index

plone3  3.1.7
test_exportimport.py
Go to the documentation of this file.
00001 ##############################################################################
00002 #
00003 # Copyright (c) 2004 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 """DCWorkflow export / import unit tests.
00014 
00015 $Id: test_exportimport.py 80705 2007-10-08 07:41:41Z rossp $
00016 """
00017 
00018 import unittest
00019 import Testing
00020 
00021 from Products.PythonScripts.PythonScript import PythonScript
00022 from Products.ExternalMethod.ExternalMethod import ExternalMethod
00023 
00024 from Products.CMFCore.exportimport.tests.test_workflow \
00025         import _BINDINGS_TOOL_EXPORT
00026 from Products.CMFCore.exportimport.tests.test_workflow \
00027         import _EMPTY_TOOL_EXPORT
00028 from Products.CMFCore.exportimport.tests.test_workflow \
00029         import _WorkflowSetup as WorkflowSetupBase
00030 from Products.CMFCore.testing import DummyWorkflow
00031 from Products.DCWorkflow.DCWorkflow import DCWorkflowDefinition
00032 from Products.DCWorkflow.testing import ExportImportZCMLLayer
00033 from Products.DCWorkflow.Transitions import TRIGGER_USER_ACTION
00034 from Products.DCWorkflow.Transitions import TRIGGER_AUTOMATIC
00035 from Products.GenericSetup.tests.common import DummyExportContext
00036 from Products.GenericSetup.tests.common import DummyImportContext
00037 
00038 
00039 class _GuardChecker:
00040 
00041     def _genGuardProps( self, permissions, roles, groups, expr ):
00042 
00043         return { 'guard_permissions'   : '; '.join( permissions )
00044                , 'guard_roles'         : '; '.join( roles )
00045                , 'guard_groups'        : '; '.join( groups )
00046                , 'guard_expr'          : expr
00047                }
00048 
00049     def _assertGuard( self, info, permissions, roles, groups, expr ):
00050 
00051         self.assertEqual( len( info[ 'guard_permissions' ] )
00052                         , len( permissions ) )
00053 
00054         for expected in permissions:
00055             self.failUnless( expected in info[ 'guard_permissions' ] )
00056 
00057         self.assertEqual( len( info[ 'guard_roles' ] )
00058                         , len( roles ) )
00059 
00060         for expected in roles:
00061             self.failUnless( expected in info[ 'guard_roles' ] )
00062 
00063         self.assertEqual( len( info[ 'guard_groups' ] )
00064                         , len( groups ) )
00065 
00066         for expected in groups:
00067             self.failUnless( expected in info[ 'guard_groups' ] )
00068 
00069         self.assertEqual( info[ 'guard_expr' ], expr )
00070 
00071 
00072 class _WorkflowSetup(WorkflowSetupBase):
00073 
00074     def _initDCWorkflow( self, workflow_id ):
00075 
00076         wf_tool = self.root.site.portal_workflow
00077         wf_tool._setObject( workflow_id, DCWorkflowDefinition( workflow_id ) )
00078 
00079         return wf_tool._getOb( workflow_id )
00080 
00081     def _initVariables( self, dcworkflow ):
00082 
00083         for id, args in _WF_VARIABLES.items():
00084 
00085             dcworkflow.variables.addVariable( id )
00086             variable = dcworkflow.variables._getOb( id )
00087 
00088             ( descr, def_val, def_exp, for_cat, for_stat, upd_alw
00089             ) = args[ :-4 ]
00090 
00091             variable.setProperties( description=args[0]
00092                                   , default_value=args[1]
00093                                   , default_expr=args[2]
00094                                   , for_catalog=args[3]
00095                                   , for_status=args[4]
00096                                   , update_always=args[5]
00097                                   , props=self._genGuardProps( *args[ -4: ] )
00098                                   )
00099 
00100     def _initStates( self, dcworkflow ):
00101 
00102         dcworkflow.groups = _WF_GROUPS
00103 
00104         for k, v in _WF_STATES.items():
00105 
00106             dcworkflow.states.addState( k )
00107             state = dcworkflow.states._getOb( k )
00108 
00109             state.setProperties( title=v[ 0 ]
00110                                , description=v[ 1 ]
00111                                , transitions=v[ 2 ]
00112                                )
00113             if not v[ 3 ]:
00114                 state.permission_roles = None
00115 
00116             for permission, roles in v[ 3 ].items():
00117                 state.setPermission( permission
00118                                    , not isinstance( roles, tuple )
00119                                    , roles
00120                                    )
00121             faux_request = {}
00122 
00123             for group_id, roles in v[ 4 ]:
00124                 for role in roles:
00125                     faux_request[ '%s|%s' % ( group_id, role ) ] = True
00126 
00127             state.setGroups( REQUEST=faux_request )
00128 
00129             for k, v in v[ 5 ].items():
00130                 state.addVariable( k, v )
00131 
00132     def _initTransitions( self, dcworkflow ):
00133 
00134         for k, v in _WF_TRANSITIONS.items():
00135 
00136             dcworkflow.transitions.addTransition( k )
00137             transition = dcworkflow.transitions._getOb( k )
00138 
00139             transition.setProperties( title=v[ 0 ]
00140                                     , description=v[ 1 ]
00141                                     , new_state_id=v[ 2 ]
00142                                     , trigger_type=v[ 3 ]
00143                                     , script_name=v[ 4 ]
00144                                     , after_script_name=v[ 5 ]
00145                                     , actbox_name=v[ 6 ]
00146                                     , actbox_url=v[ 7 ]
00147                                     , actbox_category=v[ 8 ]
00148                                     , props=self._genGuardProps( *v[ -4: ] )
00149                                     )
00150 
00151             for k, v in v[ 9 ].items():
00152                 transition.addVariable( k, v )
00153 
00154     def _initWorklists( self, dcworkflow ):
00155 
00156         for k, v in _WF_WORKLISTS.items():
00157 
00158             dcworkflow.worklists.addWorklist( k )
00159             worklist = dcworkflow.worklists._getOb( k )
00160 
00161             worklist.title = v[ 0 ]
00162 
00163             props=self._genGuardProps( *v[ -4: ] )
00164 
00165             for var_id, matches in v[ 2 ].items():
00166                 props[ 'var_match_%s' % var_id ] = ';'.join( matches )
00167 
00168             worklist.setProperties( description=v[ 1 ]
00169                                   , actbox_name=v[ 3 ]
00170                                   , actbox_url=v[ 4 ]
00171                                   , actbox_category=v[ 5 ]
00172                                   , props=props
00173                                   )
00174 
00175     def _initScripts( self, dcworkflow ):
00176 
00177         for k, v in _WF_SCRIPTS.items():
00178 
00179             if v[ 0 ] == PythonScript.meta_type:
00180                 script = PythonScript( k )
00181                 script.write( v[ 1 ] )
00182 
00183             elif v[ 0 ] == ExternalMethod.meta_type:
00184                 script = ExternalMethod(k,'', v[3], v[4])
00185 
00186             else:
00187                 raise ValueError, 'Unknown script type: %s' % v[ 0 ]
00188 
00189             dcworkflow.scripts._setObject( k, script )
00190 
00191 
00192 class WorkflowDefinitionConfiguratorTests( _WorkflowSetup, _GuardChecker ):
00193 
00194     layer = ExportImportZCMLLayer
00195 
00196     def _getTargetClass( self ):
00197         from Products.DCWorkflow.exportimport \
00198                 import WorkflowDefinitionConfigurator
00199 
00200         return WorkflowDefinitionConfigurator
00201 
00202     def test_getWorkflowInfo_dcworkflow_defaults( self ):
00203 
00204         WF_ID = 'dcworkflow_defaults'
00205 
00206         site = self._initSite()
00207         dcworkflow = self._initDCWorkflow( WF_ID )
00208 
00209         configurator = self._makeOne(dcworkflow).__of__(site)
00210         info = configurator.getWorkflowInfo( WF_ID )
00211 
00212         self.assertEqual( info[ 'id' ], WF_ID )
00213         self.assertEqual( info[ 'meta_type' ], DCWorkflowDefinition.meta_type )
00214         self.assertEqual( info[ 'title' ], dcworkflow.title )
00215         self.assertEqual( info['description'], dcworkflow.description )
00216 
00217         self.assertEqual( info[ 'state_variable' ], dcworkflow.state_var )
00218 
00219         self.assertEqual( len( info[ 'permissions' ] ), 0 )
00220         self.assertEqual( len( info[ 'variable_info' ] ), 0 )
00221         self.assertEqual( len( info[ 'state_info' ] ), 0 )
00222         self.assertEqual( len( info[ 'transition_info' ] ), 0 )
00223 
00224     def test_getWorkflowInfo_dcworkflow_permissions( self ):
00225 
00226         WF_ID = 'dcworkflow_permissions'
00227 
00228         site = self._initSite()
00229         dcworkflow = self._initDCWorkflow( WF_ID )
00230         dcworkflow.permissions = _WF_PERMISSIONS
00231 
00232         configurator = self._makeOne(dcworkflow).__of__(site)
00233         info = configurator.getWorkflowInfo( WF_ID )
00234 
00235         permissions = info[ 'permissions' ]
00236         self.assertEqual( len( permissions ), len( _WF_PERMISSIONS ) )
00237 
00238         for permission in _WF_PERMISSIONS:
00239             self.failUnless( permission in permissions )
00240 
00241     def test_getWorkflowInfo_dcworkflow_variables( self ):
00242 
00243         WF_ID = 'dcworkflow_variables'
00244 
00245         site = self._initSite()
00246         dcworkflow = self._initDCWorkflow( WF_ID )
00247         self._initVariables( dcworkflow )
00248 
00249         configurator = self._makeOne(dcworkflow).__of__(site)
00250         info = configurator.getWorkflowInfo( WF_ID )
00251 
00252         variable_info = info[ 'variable_info' ]
00253         self.assertEqual( len( variable_info ), len( _WF_VARIABLES ) )
00254 
00255         ids = [ x[ 'id' ] for x in variable_info ]
00256 
00257         for k in _WF_VARIABLES.keys():
00258             self.failUnless( k in ids )
00259 
00260         for info in variable_info:
00261 
00262             expected = _WF_VARIABLES[ info[ 'id' ] ]
00263 
00264             self.assertEqual( info[ 'description' ], expected[ 0 ] )
00265             self.assertEqual( info[ 'default_value' ], expected[ 1 ] )
00266             self.assertEqual( info[ 'default_expr' ], expected[ 2 ] )
00267             self.assertEqual( info[ 'for_catalog' ], expected[ 3 ] )
00268             self.assertEqual( info[ 'for_status' ], expected[ 4 ] )
00269             self.assertEqual( info[ 'update_always' ], expected[ 5 ] )
00270 
00271             self._assertGuard( info, *expected[ -4: ] )
00272 
00273     def test_getWorkflowInfo_dcworkflow_states( self ):
00274 
00275         WF_ID = 'dcworkflow_states'
00276         WF_INITIAL_STATE = 'closed'
00277 
00278         site = self._initSite()
00279         dcworkflow = self._initDCWorkflow( WF_ID )
00280         dcworkflow.initial_state = WF_INITIAL_STATE
00281         self._initStates( dcworkflow )
00282 
00283         configurator = self._makeOne(dcworkflow).__of__(site)
00284         info = configurator.getWorkflowInfo( WF_ID )
00285 
00286         self.assertEqual( info[ 'state_variable' ], dcworkflow.state_var )
00287         self.assertEqual( info[ 'initial_state' ], dcworkflow.initial_state )
00288 
00289         state_info = info[ 'state_info' ]
00290         self.assertEqual( len( state_info ), len( _WF_STATES ) )
00291 
00292         ids = [ x[ 'id' ] for x in state_info ]
00293 
00294         for k in _WF_STATES.keys():
00295             self.failUnless( k in ids )
00296 
00297         for info in state_info:
00298 
00299             expected = _WF_STATES[ info[ 'id' ] ]
00300 
00301             self.assertEqual( info[ 'title' ], expected[ 0 ] )
00302             self.assertEqual( info[ 'description' ], expected[ 1 ] )
00303             self.assertEqual( info[ 'transitions' ], expected[ 2 ] )
00304 
00305             permissions = info[ 'permissions' ]
00306 
00307             self.assertEqual( len( permissions ), len( expected[ 3 ] ) )
00308 
00309             for ep_id, ep_roles in expected[ 3 ].items():
00310 
00311                 fp = [ x for x in permissions if x[ 'name' ] == ep_id ][ 0 ]
00312 
00313                 self.assertEqual( fp[ 'acquired' ]
00314                                 , not isinstance( ep_roles, tuple ) )
00315 
00316                 self.assertEqual( len( fp[ 'roles' ] ), len( ep_roles ) )
00317 
00318                 for ep_role in ep_roles:
00319                     self.failUnless( ep_role in fp[ 'roles' ] )
00320 
00321             groups = info[ 'groups' ]
00322             self.assertEqual( len( groups ), len( expected[ 4 ] ) )
00323 
00324             for i in range( len( groups ) ):
00325                 self.assertEqual( groups[ i ], expected[ 4 ][ i ] )
00326 
00327             variables = info[ 'variables' ]
00328             self.assertEqual( len( variables ), len( expected[ 5 ] ) )
00329 
00330             for v_info in variables:
00331 
00332                 name, type, value = ( v_info[ 'name' ]
00333                                     , v_info[ 'type' ], v_info[ 'value' ] )
00334 
00335                 self.assertEqual( value, expected[ 5 ][ name ] )
00336 
00337                 if isinstance( value, bool ):
00338                     self.assertEqual( type, 'bool' )
00339                 elif isinstance( value, int ):
00340                     self.assertEqual( type, 'int' )
00341                 elif isinstance( value, float ):
00342                     self.assertEqual( type, 'float' )
00343                 elif isinstance( value, basestring ):
00344                     self.assertEqual( type, 'string' )
00345 
00346     def test_getWorkflowInfo_dcworkflow_transitions( self ):
00347 
00348         from Products.DCWorkflow.exportimport import TRIGGER_TYPES
00349 
00350         WF_ID = 'dcworkflow_transitions'
00351 
00352         site = self._initSite()
00353         dcworkflow = self._initDCWorkflow( WF_ID )
00354         self._initTransitions( dcworkflow )
00355 
00356         configurator = self._makeOne(dcworkflow).__of__(site)
00357         info = configurator.getWorkflowInfo( WF_ID )
00358 
00359         transition_info = info[ 'transition_info' ]
00360         self.assertEqual( len( transition_info ), len( _WF_TRANSITIONS ) )
00361 
00362         ids = [ x[ 'id' ] for x in transition_info ]
00363 
00364         for k in _WF_TRANSITIONS.keys():
00365             self.failUnless( k in ids )
00366 
00367         for info in transition_info:
00368 
00369             expected = _WF_TRANSITIONS[ info[ 'id' ] ]
00370 
00371             self.assertEqual( info[ 'title' ], expected[ 0 ] )
00372             self.assertEqual( info[ 'description' ], expected[ 1 ] )
00373             self.assertEqual( info[ 'new_state_id' ], expected[ 2 ] )
00374             self.assertEqual( info[ 'trigger_type' ]
00375                             , TRIGGER_TYPES[ expected[ 3 ] ] )
00376             self.assertEqual( info[ 'script_name' ], expected[ 4 ] )
00377             self.assertEqual( info[ 'after_script_name' ], expected[ 5 ] )
00378             self.assertEqual( info[ 'actbox_name' ], expected[ 6 ] )
00379             self.assertEqual( info[ 'actbox_url' ], expected[ 7 ] )
00380             self.assertEqual( info[ 'actbox_category' ], expected[ 8 ] )
00381 
00382             variables = info[ 'variables' ]
00383             self.assertEqual( len( variables ), len( expected[ 9 ] ) )
00384 
00385             for v_info in variables:
00386                 self.assertEqual( v_info[ 'expr' ]
00387                                 , expected[ 9 ][ v_info[ 'name' ] ] )
00388 
00389             self._assertGuard( info, *expected[ -4: ] )
00390 
00391     def test_getWorkflowInfo_dcworkflow_worklists( self ):
00392 
00393         WF_ID = 'dcworkflow_worklists'
00394 
00395         site = self._initSite()
00396         dcworkflow = self._initDCWorkflow( WF_ID )
00397         self._initWorklists( dcworkflow )
00398 
00399         configurator = self._makeOne(dcworkflow).__of__(site)
00400         info = configurator.getWorkflowInfo( WF_ID )
00401 
00402         worklist_info = info[ 'worklist_info' ]
00403         self.assertEqual( len( worklist_info ), len( _WF_WORKLISTS ) )
00404 
00405         ids = [ x[ 'id' ] for x in worklist_info ]
00406 
00407         for k in _WF_WORKLISTS.keys():
00408             self.failUnless( k in ids )
00409 
00410         for info in worklist_info:
00411 
00412             expected = _WF_WORKLISTS[ info[ 'id' ] ]
00413 
00414             self.assertEqual( info[ 'title' ], expected[ 0 ] )
00415             self.assertEqual( info[ 'description' ], expected[ 1 ] )
00416             self.assertEqual( info[ 'actbox_name' ], expected[ 3 ] )
00417             self.assertEqual( info[ 'actbox_url' ], expected[ 4 ] )
00418             self.assertEqual( info[ 'actbox_category' ], expected[ 5 ] )
00419 
00420             var_match = info[ 'var_match' ]
00421             self.assertEqual( len( var_match ), len( expected[ 2 ] ) )
00422 
00423             for var_id, values_txt in var_match:
00424 
00425                 values = [ x.strip() for x in values_txt.split( ';' ) ]
00426                 e_values = expected[ 2 ][ var_id ]
00427                 self.assertEqual( len( values ), len( e_values ) )
00428 
00429                 for e_value in e_values:
00430                     self.failUnless( e_value in values )
00431 
00432             self._assertGuard( info, *expected[ -4: ] )
00433 
00434     def test_getWorkflowInfo_dcworkflow_scripts( self ):
00435 
00436         WF_ID = 'dcworkflow_scripts'
00437 
00438         site = self._initSite()
00439         dcworkflow = self._initDCWorkflow( WF_ID )
00440         self._initScripts( dcworkflow )
00441 
00442         configurator = self._makeOne(dcworkflow).__of__(site)
00443         info = configurator.getWorkflowInfo( WF_ID )
00444 
00445         script_info = info[ 'script_info' ]
00446         self.assertEqual( len( script_info ), len( _WF_SCRIPTS ) )
00447 
00448         ids = [ x[ 'id' ] for x in script_info ]
00449 
00450         for k in _WF_SCRIPTS.keys():
00451             self.failUnless( k in ids )
00452 
00453         for info in script_info:
00454 
00455             expected = _WF_SCRIPTS[ info[ 'id' ] ]
00456 
00457             self.assertEqual( info[ 'meta_type' ], expected[ 0 ] )
00458 
00459             if info[ 'meta_type' ] == PythonScript.meta_type:
00460                 self.assertEqual( info[ 'filename' ]
00461                                 , expected[ 2 ] % WF_ID )
00462             else:
00463                 self.assertEqual( info[ 'filename' ], expected[ 2 ] )
00464 
00465     def test_generateXML_empty( self ):
00466 
00467         WF_ID = 'empty'
00468         WF_TITLE = 'Empty DCWorkflow'
00469         WF_DESCRIPTION = 'This is a empty workflow'
00470         WF_INITIAL_STATE = 'initial'
00471 
00472         site = self._initSite()
00473         dcworkflow = self._initDCWorkflow( WF_ID )
00474         dcworkflow.title = WF_TITLE
00475         dcworkflow.description = WF_DESCRIPTION
00476         dcworkflow.initial_state = WF_INITIAL_STATE
00477 
00478         configurator = self._makeOne(dcworkflow).__of__(site)
00479 
00480         self._compareDOM( configurator.generateWorkflowXML()
00481                         , _EMPTY_WORKFLOW_EXPORT % ( WF_ID
00482                                                    , WF_TITLE
00483                                                    , WF_DESCRIPTION
00484                                                    , WF_INITIAL_STATE
00485                                                    ) )
00486 
00487     def test_generateWorkflowXML_normal( self ):
00488 
00489         WF_ID = 'normal'
00490         WF_TITLE = 'Normal DCWorkflow'
00491         WF_DESCRIPTION = 'Normal Workflow'
00492         WF_INITIAL_STATE = 'closed'
00493 
00494         site = self._initSite()
00495         dcworkflow = self._initDCWorkflow( WF_ID )
00496         dcworkflow.title = WF_TITLE
00497         dcworkflow.description = WF_DESCRIPTION
00498         dcworkflow.initial_state = WF_INITIAL_STATE
00499         dcworkflow.permissions = _WF_PERMISSIONS
00500         self._initVariables( dcworkflow )
00501         self._initStates( dcworkflow )
00502         self._initTransitions( dcworkflow )
00503         self._initWorklists( dcworkflow )
00504         self._initScripts( dcworkflow )
00505 
00506         configurator = self._makeOne(dcworkflow).__of__(site)
00507 
00508         self._compareDOM( configurator.generateWorkflowXML()
00509                         , _NORMAL_WORKFLOW_EXPORT
00510                           % { 'workflow_id' : WF_ID
00511                             , 'title' : WF_TITLE
00512                             , 'description' : WF_DESCRIPTION
00513                             , 'initial_state' : WF_INITIAL_STATE
00514                             , 'workflow_filename' : WF_ID.replace(' ', '_')
00515                             } )
00516 
00517     def test_generateWorkflowXML_multiple( self ):
00518 
00519         WF_ID_1 = 'dc1'
00520         WF_TITLE_1 = 'Normal DCWorkflow #1'
00521         WF_DESCRIPTION_1 = 'Normal Number 1'
00522         WF_ID_2 = 'dc2'
00523         WF_TITLE_2 = 'Normal DCWorkflow #2'
00524         WF_DESCRIPTION_2 = 'Normal Numer 2'
00525         WF_INITIAL_STATE = 'closed'
00526 
00527         site = self._initSite()
00528 
00529         dcworkflow_1 = self._initDCWorkflow( WF_ID_1 )
00530         dcworkflow_1.title = WF_TITLE_1
00531         dcworkflow_1.description = WF_DESCRIPTION_1
00532         dcworkflow_1.initial_state = WF_INITIAL_STATE
00533         dcworkflow_1.permissions = _WF_PERMISSIONS
00534         self._initVariables( dcworkflow_1 )
00535         self._initStates( dcworkflow_1 )
00536         self._initTransitions( dcworkflow_1 )
00537         self._initWorklists( dcworkflow_1 )
00538         self._initScripts( dcworkflow_1 )
00539 
00540         dcworkflow_2 = self._initDCWorkflow( WF_ID_2 )
00541         dcworkflow_2.title = WF_TITLE_2
00542         dcworkflow_2.description = WF_DESCRIPTION_2
00543         dcworkflow_2.initial_state = WF_INITIAL_STATE
00544         dcworkflow_2.permissions = _WF_PERMISSIONS
00545         self._initVariables( dcworkflow_2 )
00546         self._initStates( dcworkflow_2 )
00547         self._initTransitions( dcworkflow_2 )
00548         self._initWorklists( dcworkflow_2 )
00549         self._initScripts( dcworkflow_2 )
00550 
00551         configurator = self._makeOne(dcworkflow_1).__of__(site)
00552 
00553         self._compareDOM( configurator.generateWorkflowXML()
00554                         , _NORMAL_WORKFLOW_EXPORT
00555                           % { 'workflow_id' : WF_ID_1
00556                             , 'title' : WF_TITLE_1
00557                             , 'description' : WF_DESCRIPTION_1
00558                             , 'initial_state' : WF_INITIAL_STATE
00559                             , 'workflow_filename' : WF_ID_1.replace(' ', '_')
00560                             } )
00561 
00562         configurator = self._makeOne(dcworkflow_2).__of__(site)
00563 
00564         self._compareDOM( configurator.generateWorkflowXML()
00565                         , _NORMAL_WORKFLOW_EXPORT
00566                           % { 'workflow_id' : WF_ID_2
00567                             , 'title' : WF_TITLE_2
00568                             , 'description' : WF_DESCRIPTION_2
00569                             , 'initial_state' : WF_INITIAL_STATE
00570                             , 'workflow_filename' : WF_ID_2.replace(' ', '_')
00571                             } )
00572 
00573     def test_parseWorkflowXML_empty( self ):
00574 
00575         WF_ID = 'empty'
00576         WF_TITLE = 'Empty DCWorkflow'
00577         WF_DESCRIPTION = 'This is an empty workflow'
00578         WF_INITIAL_STATE = 'initial'
00579 
00580         site = self._initSite()
00581 
00582         configurator = self._makeOne( site ).__of__( site )
00583 
00584         ( workflow_id
00585         , title
00586         , state_variable
00587         , initial_state
00588         , states
00589         , transitions
00590         , variables
00591         , worklists
00592         , permissions
00593         , scripts
00594         , description
00595         ) = configurator.parseWorkflowXML( _EMPTY_WORKFLOW_EXPORT
00596                                          % ( WF_ID
00597                                            , WF_TITLE
00598                                            , WF_DESCRIPTION
00599                                            , WF_INITIAL_STATE
00600                                            ) )
00601 
00602         self.assertEqual(description, WF_DESCRIPTION)
00603         self.assertEqual( len( states ), 0 )
00604         self.assertEqual( len( transitions ), 0 )
00605         self.assertEqual( len( variables ), 0 )
00606         self.assertEqual( len( worklists ), 0 )
00607         self.assertEqual( len( permissions ), 0 )
00608         self.assertEqual( len( scripts ), 0 )
00609 
00610     def test_parseWorkflowXML_normal_attribs( self ):
00611 
00612         WF_ID = 'normal'
00613         WF_TITLE = 'Normal DCWorkflow'
00614         WF_DESCRIPTION = 'This is a normal DCWorkflow'
00615         WF_INITIAL_STATE = 'closed'
00616 
00617         site = self._initSite()
00618 
00619         configurator = self._makeOne( site ).__of__( site )
00620 
00621         ( workflow_id
00622         , title
00623         , state_variable
00624         , initial_state
00625         , states
00626         , transitions
00627         , variables
00628         , worklists
00629         , permissions
00630         , scripts
00631         , description
00632         ) = configurator.parseWorkflowXML(
00633                           _NORMAL_WORKFLOW_EXPORT
00634                           % { 'workflow_id' : WF_ID
00635                             , 'title' : WF_TITLE
00636                             , 'description' : WF_DESCRIPTION
00637                             , 'initial_state' : WF_INITIAL_STATE
00638                             , 'workflow_filename' : WF_ID.replace(' ', '_')
00639                             } )
00640 
00641         self.assertEqual( workflow_id, WF_ID )
00642         self.assertEqual( title, WF_TITLE )
00643         self.assertEqual( description, WF_DESCRIPTION )
00644         self.assertEqual( state_variable, 'state' )
00645         self.assertEqual( initial_state, WF_INITIAL_STATE )
00646 
00647     def test_parseWorkflowXML_normal_states( self ):
00648 
00649         WF_ID = 'normal'
00650         WF_TITLE = 'Normal DCWorkflow'
00651         WF_DESCRIPTION = 'Normal workflow'
00652         WF_INITIAL_STATE = 'closed'
00653 
00654         site = self._initSite()
00655 
00656         configurator = self._makeOne( site ).__of__( site )
00657 
00658         ( workflow_id
00659         , title
00660         , state_variable
00661         , initial_state
00662         , states
00663         , transitions
00664         , variables
00665         , worklists
00666         , permissions
00667         , scripts
00668         , description
00669         ) = configurator.parseWorkflowXML(
00670                           _NORMAL_WORKFLOW_EXPORT
00671                           % { 'workflow_id' : WF_ID
00672                             , 'title' : WF_TITLE
00673                             , 'description' : WF_DESCRIPTION
00674                             , 'initial_state' : WF_INITIAL_STATE
00675                             , 'workflow_filename' : WF_ID.replace(' ', '_')
00676                             } )
00677 
00678         self.assertEqual( len( states ), len( _WF_STATES ) )
00679 
00680         for state in states:
00681 
00682             state_id = state[ 'state_id' ]
00683             self.failUnless( state_id in _WF_STATES )
00684 
00685             expected = _WF_STATES[ state_id ]
00686 
00687             self.assertEqual( state[ 'title' ], expected[ 0 ] )
00688 
00689             description = ''.join( state[ 'description' ] )
00690             self.failUnless( expected[ 1 ] in description )
00691 
00692             self.assertEqual( tuple( state[ 'transitions' ] ), expected[ 2 ] )
00693             self.assertEqual( state[ 'permissions' ], expected[ 3 ] )
00694             self.assertEqual( tuple( state[ 'groups' ] )
00695                             , tuple( expected[ 4 ] ) )
00696 
00697             for k, v_info in state[ 'variables' ].items():
00698 
00699                 exp_value = expected[ 5 ][ k ]
00700                 self.assertEqual( v_info[ 'value' ], str( exp_value ) )
00701 
00702                 if isinstance( exp_value, bool ):
00703                     self.assertEqual( v_info[ 'type' ], 'bool' )
00704                 elif isinstance( exp_value, int ):
00705                     self.assertEqual( v_info[ 'type' ], 'int' )
00706                 elif isinstance( exp_value, float ):
00707                     self.assertEqual( v_info[ 'type' ], 'float' )
00708                 elif isinstance( exp_value, basestring ):
00709                     self.assertEqual( v_info[ 'type' ], 'string' )
00710 
00711     def test_parseWorkflowXML_normal_transitions( self ):
00712 
00713         from Products.DCWorkflow.exportimport import TRIGGER_TYPES
00714 
00715         WF_ID = 'normal'
00716         WF_TITLE = 'Normal DCWorkflow'
00717         WF_DESCRIPTION = 'Normal workflow'
00718         WF_INITIAL_STATE = 'closed'
00719 
00720         site = self._initSite()
00721 
00722         configurator = self._makeOne( site ).__of__( site )
00723 
00724         ( workflow_id
00725         , title
00726         , state_variable
00727         , initial_state
00728         , states
00729         , transitions
00730         , variables
00731         , worklists
00732         , permissions
00733         , scripts
00734         , description 
00735         ) = configurator.parseWorkflowXML(
00736                           _NORMAL_WORKFLOW_EXPORT
00737                           % { 'workflow_id' : WF_ID
00738                             , 'title' : WF_TITLE
00739                             , 'description' : WF_DESCRIPTION
00740                             , 'initial_state' : WF_INITIAL_STATE
00741                             , 'workflow_filename' : WF_ID.replace(' ', '_')
00742                             } )
00743 
00744         self.assertEqual( len( transitions ), len( _WF_TRANSITIONS ) )
00745 
00746         for transition in transitions:
00747 
00748             transition_id = transition[ 'transition_id' ]
00749             self.failUnless( transition_id in _WF_TRANSITIONS )
00750 
00751             expected = _WF_TRANSITIONS[ transition_id ]
00752 
00753             self.assertEqual( transition[ 'title' ], expected[ 0 ] )
00754 
00755             description = ''.join( transition[ 'description' ] )
00756             self.failUnless( expected[ 1 ] in description )
00757 
00758             self.assertEqual( transition[ 'new_state' ], expected[ 2 ] )
00759             self.assertEqual( transition[ 'trigger' ]
00760                             , TRIGGER_TYPES[ expected[ 3 ] ] )
00761             self.assertEqual( transition[ 'before_script' ], expected[ 4 ] )
00762             self.assertEqual( transition[ 'after_script' ]
00763                             , expected[ 5 ] )
00764 
00765             action = transition[ 'action' ]
00766             self.assertEqual( action.get( 'name', '' ), expected[ 6 ] )
00767             self.assertEqual( action.get( 'url', '' ), expected[ 7 ] )
00768             self.assertEqual( action.get( 'category', '' ), expected[ 8 ] )
00769 
00770             self.assertEqual( transition[ 'variables' ], expected[ 9 ] )
00771 
00772             guard = transition[ 'guard' ]
00773             self.assertEqual( tuple( guard.get( 'permissions', () ) )
00774                             , expected[ 10 ] )
00775             self.assertEqual( tuple( guard.get( 'roles', () ) )
00776                             , expected[ 11 ] )
00777             self.assertEqual( tuple( guard.get( 'groups', () ) )
00778                             , expected[ 12 ] )
00779             self.assertEqual( guard.get( 'expression', '' ), expected[ 13 ] )
00780 
00781     def test_parseWorkflowXML_normal_variables( self ):
00782 
00783         WF_ID = 'normal'
00784         WF_TITLE = 'Normal DCWorkflow'
00785         WF_DESCRIPTION = 'Normal workflow'
00786         WF_INITIAL_STATE = 'closed'
00787 
00788         site = self._initSite()
00789 
00790         configurator = self._makeOne( site ).__of__( site )
00791 
00792         ( workflow_id
00793         , title
00794         , state_variable
00795         , initial_state
00796         , states
00797         , transitions
00798         , variables
00799         , worklists
00800         , permissions
00801         , scripts
00802         , description
00803         ) = configurator.parseWorkflowXML(
00804                           _NORMAL_WORKFLOW_EXPORT
00805                           % { 'workflow_id' : WF_ID
00806                             , 'title' : WF_TITLE
00807                             , 'description' : WF_DESCRIPTION
00808                             , 'initial_state' : WF_INITIAL_STATE
00809                             , 'workflow_filename' : WF_ID.replace(' ', '_')
00810                             } )
00811 
00812         self.assertEqual( len( variables ), len( _WF_VARIABLES ) )
00813 
00814         for variable in variables:
00815 
00816             variable_id = variable[ 'variable_id' ]
00817             self.failUnless( variable_id in _WF_VARIABLES )
00818 
00819             expected = _WF_VARIABLES[ variable_id ]
00820 
00821             description = ''.join( variable[ 'description' ] )
00822             self.failUnless( expected[ 0 ] in description )
00823 
00824             default = variable[ 'default' ]
00825             self.assertEqual( default[ 'value' ], expected[ 1 ] )
00826 
00827             exp_type = 'n/a'
00828 
00829             if expected[ 1 ]:
00830                 exp_value = expected[ 1 ]
00831 
00832                 if isinstance( exp_value, bool ):
00833                     exp_type = 'bool'
00834                 elif isinstance( exp_value, int ):
00835                     exp_type = 'int'
00836                 elif isinstance( exp_value, float ):
00837                     exp_type = 'float'
00838                 elif isinstance( exp_value, basestring ):
00839                     exp_type = 'string'
00840                 else:
00841                     exp_type = 'XXX'
00842 
00843             self.assertEqual( default[ 'type' ], exp_type )
00844             self.assertEqual( default[ 'expression' ], expected[ 2 ] )
00845 
00846             self.assertEqual( variable[ 'for_catalog' ], expected[ 3 ] )
00847             self.assertEqual( variable[ 'for_status' ], expected[ 4 ] )
00848             self.assertEqual( variable[ 'update_always' ], expected[ 5 ] )
00849 
00850             guard = variable[ 'guard' ]
00851             self.assertEqual( tuple( guard.get( 'permissions', () ) )
00852                             , expected[ 6 ] )
00853             self.assertEqual( tuple( guard.get( 'roles', () ) )
00854                             , expected[ 7 ] )
00855             self.assertEqual( tuple( guard.get( 'groups', () ) )
00856                             , expected[ 8 ] )
00857             self.assertEqual( guard.get( 'expression', '' ), expected[ 9 ] )
00858 
00859     def test_parseWorkflowXML_normal_worklists( self ):
00860 
00861         WF_ID = 'normal'
00862         WF_TITLE = 'Normal DCWorkflow'
00863         WF_DESCRIPTION = 'Normal workflow'
00864         WF_INITIAL_STATE = 'closed'
00865 
00866         site = self._initSite()
00867 
00868         configurator = self._makeOne( site ).__of__( site )
00869 
00870         ( workflow_id
00871         , title
00872         , state_variable
00873         , initial_state
00874         , states
00875         , transitions
00876         , variables
00877         , worklists
00878         , permissions
00879         , scripts
00880         , description
00881         ) = configurator.parseWorkflowXML(
00882                           _NORMAL_WORKFLOW_EXPORT
00883                           % { 'workflow_id' : WF_ID
00884                             , 'title' : WF_TITLE
00885                             , 'description' : WF_DESCRIPTION
00886                             , 'initial_state' : WF_INITIAL_STATE
00887                             , 'workflow_filename' : WF_ID.replace(' ', '_')
00888                             } )
00889 
00890         self.assertEqual( len( worklists ), len( _WF_WORKLISTS ) )
00891 
00892         for worklist in worklists:
00893 
00894             worklist_id = worklist[ 'worklist_id' ]
00895             self.failUnless( worklist_id in _WF_WORKLISTS )
00896 
00897             expected = _WF_WORKLISTS[ worklist_id ]
00898 
00899             self.assertEqual( worklist[ 'title' ], expected[ 0 ] )
00900 
00901             description = ''.join( worklist[ 'description' ] )
00902             self.failUnless( expected[ 1 ] in description )
00903 
00904             self.assertEqual( tuple( worklist[ 'match' ] )
00905                             , tuple( expected[ 2 ] ) )
00906 
00907             action = worklist[ 'action' ]
00908             self.assertEqual( action.get( 'name', '' ), expected[ 3 ] )
00909             self.assertEqual( action.get( 'url', '' ), expected[ 4 ] )
00910             self.assertEqual( action.get( 'category', '' ), expected[ 5 ] )
00911 
00912             guard = worklist[ 'guard' ]
00913             self.assertEqual( tuple( guard.get( 'permissions', () ) )
00914                             , expected[ 6 ] )
00915             self.assertEqual( tuple( guard.get( 'roles', () ) )
00916                             , expected[ 7 ] )
00917             self.assertEqual( tuple( guard.get( 'groups', () ) )
00918                             , expected[ 8 ] )
00919             self.assertEqual( guard.get( 'expression', '' ), expected[ 9 ] )
00920 
00921     def test_parseWorkflowXML_normal_permissions( self ):
00922 
00923         WF_ID = 'normal'
00924         WF_TITLE = 'Normal DCWorkflow'
00925         WF_DESCRIPTION = 'Normal workflow'
00926         WF_INITIAL_STATE = 'closed'
00927 
00928         site = self._initSite()
00929 
00930         configurator = self._makeOne( site ).__of__( site )
00931 
00932         ( workflow_id
00933         , title
00934         , state_variable
00935         , initial_state
00936         , states
00937         , transitions
00938         , variables
00939         , worklists
00940         , permissions
00941         , scripts
00942         , description
00943         ) = configurator.parseWorkflowXML(
00944                           _NORMAL_WORKFLOW_EXPORT
00945                           % { 'workflow_id' : WF_ID
00946                             , 'title' : WF_TITLE
00947                             , 'description' : WF_DESCRIPTION
00948                             , 'initial_state' : WF_INITIAL_STATE
00949                             , 'workflow_filename' : WF_ID.replace(' ', '_')
00950                             } )
00951 
00952         self.assertEqual( len( permissions ), len( _WF_PERMISSIONS ) )
00953 
00954         for permission in permissions:
00955 
00956             self.failUnless( permission in _WF_PERMISSIONS )
00957 
00958     def test_parseWorkflowXML_normal_scripts( self ):
00959 
00960         WF_ID = 'normal'
00961         WF_TITLE = 'Normal DCWorkflow'
00962         WF_DESCRIPTION = 'Normal workflow'
00963         WF_INITIAL_STATE = 'closed'
00964 
00965         site = self._initSite()
00966 
00967         configurator = self._makeOne( site ).__of__( site )
00968 
00969         ( workflow_id
00970         , title
00971         , state_variable
00972         , initial_state
00973         , states
00974         , transitions
00975         , variables
00976         , worklists
00977         , permissions
00978         , scripts
00979         , description
00980         ) = configurator.parseWorkflowXML(
00981                           _NORMAL_WORKFLOW_EXPORT
00982                           % { 'workflow_id' : WF_ID
00983                             , 'title' : WF_TITLE
00984                             , 'description': WF_DESCRIPTION
00985                             , 'initial_state' : WF_INITIAL_STATE
00986                             , 'workflow_filename' : WF_ID.replace(' ', '_')
00987                             } )
00988 
00989         self.assertEqual( len( scripts ), len( _WF_SCRIPTS ) )
00990 
00991         for script in scripts:
00992 
00993             script_id = script[ 'script_id' ]
00994             self.failUnless( script_id in _WF_SCRIPTS )
00995 
00996             expected = _WF_SCRIPTS[ script_id ]
00997 
00998             self.assertEqual( script[ 'meta_type' ], expected[ 0 ] )
00999 
01000             # Body is not kept as part of the workflow XML
01001 
01002             if script[ 'meta_type' ] == PythonScript.meta_type:
01003                 self.assertEqual( script[ 'filename' ]
01004                                 , expected[ 2 ] % workflow_id )
01005             else:
01006                 self.assertEqual( script[ 'filename' ], expected[ 2 ] )
01007 
01008 
01009 _WF_PERMISSIONS = \
01010 ( 'Open content for modifications'
01011 , 'Modify content'
01012 , 'Query history'
01013 , 'Restore expired content'
01014 )
01015 
01016 _WF_GROUPS = \
01017 ( 'Content_owners'
01018 , 'Content_assassins'
01019 )
01020 
01021 _WF_VARIABLES = \
01022 { 'when_opened':  ( 'Opened when'
01023                   , ''
01024                   , "python:None"
01025                   , True
01026                   , False
01027                   , True
01028                   , ( 'Query history', 'Open content for modifications' )
01029                   , ()
01030                   , ()
01031                   , ""
01032                   )
01033 , 'when_expired': ( 'Expired when'
01034                   , ''
01035                   , "nothing"
01036                   , True
01037                   , False
01038                   , True
01039                   , ( 'Query history', 'Open content for modifications' )
01040                   , ()
01041                   , ()
01042                   , ""
01043                   )
01044 , 'killed_by':    ( 'Killed by'
01045                   , 'n/a'
01046                   , ""
01047                   , True
01048                   , False
01049                   , True
01050                   , ()
01051                   , ( 'Hangman', 'Sherrif' )
01052                   , ()
01053                   , ""
01054                   )
01055 }
01056 
01057 _WF_STATES = \
01058 { 'closed':  ( 'Closed'
01059              , 'Closed for modifications'
01060              , ( 'open', 'kill', 'expire' )
01061              , { 'Modify content':  () }
01062              , ()
01063              , { 'is_opened':  False, 'is_closed':  True }
01064              )
01065 , 'opened':  ( 'Opened'
01066              , 'Open for modifications'
01067              , ( 'close', 'kill', 'expire' )
01068              , { 'Modify content':  [ 'Owner', 'Manager' ] }
01069              , [ ( 'Content_owners', ( 'Owner', ) ) ]
01070              , { 'is_opened':  True, 'is_closed':  False }
01071              )
01072 , 'killed':  ( 'Killed'
01073              , 'Permanently unavailable'
01074              , ()
01075              , {}
01076              , ()
01077              , {}
01078              )
01079 , 'expired': ( 'Expired'
01080              , 'Expiration date has passed'
01081              , ( 'open', )
01082              , { 'Modify content':  [ 'Owner', 'Manager' ] }
01083              , ()
01084              , { 'is_opened':  False, 'is_closed':  False }
01085              )
01086 }
01087 
01088 _WF_TRANSITIONS = \
01089 { 'open':    ( 'Open'
01090              , 'Open the object for modifications'
01091              , 'opened'
01092              , TRIGGER_USER_ACTION
01093              , 'before_open'
01094              , ''
01095              , 'Open'
01096              , 'string:${object_url}/open_for_modifications'
01097              , 'workflow'
01098              , { 'when_opened' : 'object/ZopeTime' }
01099              , ( 'Open content for modifications', )
01100              , ()
01101              , ()
01102              , ""
01103              )
01104 , 'close':   ( 'Close'
01105              , 'Close the object for modifications'
01106              , 'closed'
01107              , TRIGGER_USER_ACTION
01108              , ''
01109              , 'after_close'
01110              , 'Close'
01111              , 'string:${object_url}/close_for_modifications'
01112              , 'workflow'
01113              , {}
01114              , ()
01115              , ( 'Owner', 'Manager' )
01116              , ()
01117              , ""
01118              )
01119 , 'kill':    ( 'Kill'
01120              , 'Make the object permanently unavailable.'
01121              , 'killed'
01122              , TRIGGER_USER_ACTION
01123              , ''
01124              , 'after_kill'
01125              , 'Kill'
01126              , 'string:${object_url}/kill_object'
01127              , 'workflow'
01128              , { 'killed_by' : 'string:${user/getId}' }
01129              , ()
01130              , ()
01131              , ( 'Content_assassins', )
01132              , ""
01133              )
01134 , 'expire':  ( 'Expire'
01135              , 'Retire objects whose expiration is past.'
01136              , 'expired'
01137              , TRIGGER_AUTOMATIC
01138              , 'before_expire'
01139              , ''
01140              , ''
01141              , ''
01142              , ''
01143              , { 'when_expired' : 'object/ZopeTime' }
01144              , ()
01145              , ()
01146              , ()
01147              , "python: object.expiration() <= object.ZopeTime()"
01148              )
01149 }
01150 
01151 _WF_WORKLISTS = \
01152 { 'expired_list':   ( 'Expired'
01153                     , 'Worklist for expired content'
01154                     , { 'state' : ( 'expired', ) }
01155                     , 'Expired items'
01156                     , 'string:${portal_url}/expired_items'
01157                     , 'workflow'
01158                     , ( 'Restore expired content', )
01159                     , ()
01160                     , ()
01161                     , ""
01162                     )
01163 , 'alive_list':     ( 'Alive'
01164                     , 'Worklist for content not yet expired / killed'
01165                     , { 'state' : ( 'open',  'closed' ) }
01166                     , 'Expired items'
01167                     , 'string:${portal_url}/expired_items'
01168                     , 'workflow'
01169                     , ( 'Restore expired content', )
01170                     , ()
01171                     , ()
01172                     , ""
01173                     )
01174 }
01175 
01176 _BEFORE_OPEN_SCRIPT = """\
01177 ## Script (Python) "before_open"
01178 ##bind container=container
01179 ##bind context=context
01180 ##bind namespace=
01181 ##bind script=script
01182 ##bind subpath=traverse_subpath
01183 ##parameters=
01184 ##title=
01185 ##
01186 return 'before_open'
01187 """
01188 
01189 _AFTER_CLOSE_SCRIPT = """\
01190 ## Script (Python) "after_close"
01191 ##bind container=container
01192 ##bind context=context
01193 ##bind namespace=
01194 ##bind script=script
01195 ##bind subpath=traverse_subpath
01196 ##parameters=
01197 ##title=
01198 ##
01199 return 'after_close'
01200 """
01201 
01202 _AFTER_KILL_SCRIPT = """\
01203 ## Script (Python) "after_kill"
01204 ##bind container=container
01205 ##bind context=context
01206 ##bind namespace=
01207 ##bind script=script
01208 ##bind subpath=traverse_subpath
01209 ##parameters=
01210 ##title=
01211 ##
01212 return 'after_kill'
01213 """
01214 
01215 _WF_SCRIPTS = \
01216 { 'before_open':    ( PythonScript.meta_type
01217                     , _BEFORE_OPEN_SCRIPT
01218                     , 'workflows/%s/scripts/before_open.py'
01219                     , None
01220                     , None
01221                     )
01222 , 'after_close':    ( PythonScript.meta_type
01223                     , _AFTER_CLOSE_SCRIPT
01224                     , 'workflows/%s/scripts/after_close.py'
01225                     , None
01226                     , None
01227                     )
01228 , 'after_kill':     ( PythonScript.meta_type
01229                     , _AFTER_KILL_SCRIPT
01230                     , 'workflows/%s/scripts/after_kill.py'
01231                     , None
01232                     , None
01233                     )
01234 , 'before_expire': ( ExternalMethod.meta_type
01235                    , ''
01236                    , ''
01237                    , 'DCWorkflow.test_method'
01238                    , 'test'
01239                    )
01240 }
01241 
01242 _NORMAL_TOOL_EXPORT = """\
01243 <?xml version="1.0"?>
01244 <object name="portal_workflow" meta_type="Dummy Workflow Tool">
01245  <property name="title"></property>
01246  <object name="Non-DCWorkflow" meta_type="Dummy Workflow"/>
01247  <object name="dcworkflow" meta_type="Workflow"/>
01248  <bindings>
01249   <default/>
01250  </bindings>
01251 </object>
01252 """
01253 
01254 _NORMAL_TOOL_EXPORT_WITH_FILENAME = """\
01255 <?xml version="1.0"?>
01256 <object name="portal_workflow" meta_type="Dummy Workflow Tool">
01257  <property name="title"></property>
01258  <object name="Non-DCWorkflow" meta_type="Dummy Workflow"/>
01259  <object name="%(workflow_id)s" meta_type="Workflow"/>
01260  <bindings>
01261   <default/>
01262  </bindings>
01263 </object>
01264 """
01265 
01266 _FILENAME_TOOL_EXPORT = """\
01267 <?xml version="1.0"?>
01268 <object name="portal_workflow" meta_type="Dummy Workflow Tool">
01269  <property name="title"></property>
01270  <object name="name with spaces" meta_type="Workflow"/>
01271  <bindings>
01272   <default/>
01273  </bindings>
01274 </object>
01275 """
01276 
01277 _EMPTY_WORKFLOW_EXPORT = """\
01278 <?xml version="1.0"?>
01279 <dc-workflow
01280     workflow_id="%s"
01281     title="%s"
01282     description="%s"
01283     state_variable="state"
01284     initial_state="%s">
01285 </dc-workflow>
01286 """
01287 
01288 # Make sure old exports are still imported well. Changes:
01289 # - scripts are now in in a 'scripts' subdirectory
01290 _OLD_WORKFLOW_EXPORT = """\
01291 <?xml version="1.0"?>
01292 <dc-workflow
01293     workflow_id="%(workflow_id)s"
01294     title="%(title)s"
01295     state_variable="state"
01296     initial_state="%(initial_state)s">
01297  <permission>Open content for modifications</permission>
01298  <permission>Modify content</permission>
01299  <permission>Query history</permission>
01300  <permission>Restore expired content</permission>
01301  <state
01302     state_id="closed"
01303     title="Closed">
01304   <description>Closed for modifications</description>
01305   <exit-transition
01306     transition_id="open"/>
01307   <exit-transition
01308     transition_id="kill"/>
01309   <exit-transition
01310     transition_id="expire"/>
01311   <permission-map
01312     acquired="False"
01313     name="Modify content">
01314   </permission-map>
01315   <assignment
01316     name="is_closed"
01317     type="bool">True</assignment>
01318   <assignment
01319     name="is_opened"
01320     type="bool">False</assignment>
01321  </state>
01322  <state
01323     state_id="expired"
01324     title="Expired">
01325   <description>Expiration date has passed</description>
01326   <exit-transition
01327     transition_id="open"/>
01328   <permission-map
01329     acquired="True"
01330     name="Modify content">
01331    <permission-role>Owner</permission-role>
01332    <permission-role>Manager</permission-role>
01333   </permission-map>
01334   <assignment
01335     name="is_closed"
01336     type="bool">False</assignment>
01337   <assignment
01338     name="is_opened"
01339     type="bool">False</assignment>
01340  </state>
01341  <state
01342     state_id="killed"
01343     title="Killed">
01344   <description>Permanently unavailable</description>
01345  </state>
01346  <state
01347     state_id="opened"
01348     title="Opened">
01349   <description>Open for modifications</description>
01350   <exit-transition
01351     transition_id="close"/>
01352   <exit-transition
01353     transition_id="kill"/>
01354   <exit-transition
01355     transition_id="expire"/>
01356   <permission-map
01357     acquired="True"
01358     name="Modify content">
01359    <permission-role>Owner</permission-role>
01360    <permission-role>Manager</permission-role>
01361   </permission-map>
01362   <group-map name="Content_owners">
01363    <group-role>Owner</group-role>
01364   </group-map>
01365   <assignment
01366     name="is_closed"
01367     type="bool">False</assignment>
01368   <assignment
01369     name="is_opened"
01370     type="bool">True</assignment>
01371  </state>
01372  <transition
01373     transition_id="close"
01374     title="Close"
01375     trigger="USER"
01376     new_state="closed"
01377     before_script=""
01378     after_script="after_close">
01379   <description>Close the object for modifications</description>
01380   <action
01381     category="workflow"
01382     url="string:${object_url}/close_for_modifications">Close</action>
01383   <guard>
01384    <guard-role>Owner</guard-role>
01385    <guard-role>Manager</guard-role>
01386   </guard>
01387  </transition>
01388  <transition
01389     transition_id="expire"
01390     title="Expire"
01391     trigger="AUTOMATIC"
01392     new_state="expired"
01393     before_script="before_expire"
01394     after_script="">
01395   <description>Retire objects whose expiration is past.</description>
01396   <guard>
01397    <guard-expression>python: object.expiration() &lt;= object.ZopeTime()</guard-expression>
01398   </guard>
01399   <assignment
01400     name="when_expired">object/ZopeTime</assignment>
01401  </transition>
01402  <transition
01403     transition_id="kill"
01404     title="Kill"
01405     trigger="USER"
01406     new_state="killed"
01407     before_script=""
01408     after_script="after_kill">
01409   <description>Make the object permanently unavailable.</description>
01410   <action
01411     category="workflow"
01412     url="string:${object_url}/kill_object">Kill</action>
01413   <guard>
01414    <guard-group>Content_assassins</guard-group>
01415   </guard>
01416   <assignment
01417     name="killed_by">string:${user/getId}</assignment>
01418  </transition>
01419  <transition
01420     transition_id="open"
01421     title="Open"
01422     trigger="USER"
01423     new_state="opened"
01424     before_script="before_open"
01425     after_script="">
01426   <description>Open the object for modifications</description>
01427   <action
01428     category="workflow"
01429     url="string:${object_url}/open_for_modifications">Open</action>
01430   <guard>
01431    <guard-permission>Open content for modifications</guard-permission>
01432   </guard>
01433   <assignment
01434     name="when_opened">object/ZopeTime</assignment>
01435  </transition>
01436  <worklist
01437     worklist_id="alive_list"
01438     title="Alive">
01439   <description>Worklist for content not yet expired / killed</description>
01440   <action
01441     category="workflow"
01442     url="string:${portal_url}/expired_items">Expired items</action>
01443   <guard>
01444    <guard-permission>Restore expired content</guard-permission>
01445   </guard>
01446   <match name="state" values="open; closed"/>
01447  </worklist>
01448  <worklist
01449     worklist_id="expired_list"
01450     title="Expired">
01451   <description>Worklist for expired content</description>
01452   <action
01453     category="workflow"
01454     url="string:${portal_url}/expired_items">Expired items</action>
01455   <guard>
01456    <guard-permission>Restore expired content</guard-permission>
01457   </guard>
01458   <match name="state" values="expired"/>
01459  </worklist>
01460  <variable
01461     variable_id="killed_by"
01462     for_catalog="True"
01463     for_status="False"
01464     update_always="True">
01465    <description>Killed by</description>
01466    <default>
01467     <value type="string">n/a</value>
01468    </default>
01469    <guard>
01470     <guard-role>Hangman</guard-role>
01471     <guard-role>Sherrif</guard-role>
01472    </guard>
01473  </variable>
01474  <variable
01475     variable_id="when_expired"
01476     for_catalog="True"
01477     for_status="False"
01478     update_always="True">
01479    <description>Expired when</description>
01480    <default>
01481     <expression>nothing</expression>
01482    </default>
01483    <guard>
01484     <guard-permission>Query history</guard-permission>
01485     <guard-permission>Open content for modifications</guard-permission>
01486    </guard>
01487  </variable>
01488  <variable
01489     variable_id="when_opened"
01490     for_catalog="True"
01491     for_status="False"
01492     update_always="True">
01493    <description>Opened when</description>
01494    <default>
01495     <expression>python:None</expression>
01496    </default>
01497    <guard>
01498     <guard-permission>Query history</guard-permission>
01499     <guard-permission>Open content for modifications</guard-permission>
01500    </guard>
01501  </variable>
01502  <script
01503     script_id="after_close"
01504     type="Script (Python)"
01505     filename="workflows/%(workflow_filename)s/after_close.py"
01506     module=""
01507     function=""
01508     />
01509  <script
01510     script_id="after_kill"
01511     type="Script (Python)"
01512     filename="workflows/%(workflow_filename)s/after_kill.py"
01513     module=""
01514     function=""
01515     />
01516  <script
01517     script_id="before_expire"
01518     type="External Method"
01519     filename=""
01520     module="DCWorkflow.test_method"
01521     function="test"
01522     />
01523  <script
01524     script_id="before_open"
01525     type="Script (Python)"
01526     filename="workflows/%(workflow_filename)s/before_open.py"
01527     module=""
01528     function=""
01529     />
01530 </dc-workflow>
01531 """
01532 
01533 _NORMAL_WORKFLOW_EXPORT = """\
01534 <?xml version="1.0"?>
01535 <dc-workflow
01536     workflow_id="%(workflow_id)s"
01537     title="%(title)s"
01538     description="%(description)s"
01539     state_variable="state"
01540     initial_state="%(initial_state)s">
01541  <permission>Open content for modifications</permission>
01542  <permission>Modify content</permission>
01543  <permission>Query history</permission>
01544  <permission>Restore expired content</permission>
01545  <state
01546     state_id="closed"
01547     title="Closed">
01548   <description>Closed for modifications</description>
01549   <exit-transition
01550     transition_id="open"/>
01551   <exit-transition
01552     transition_id="kill"/>
01553   <exit-transition
01554     transition_id="expire"/>
01555   <permission-map
01556     acquired="False"
01557     name="Modify content">
01558   </permission-map>
01559   <assignment
01560     name="is_closed"
01561     type="bool">True</assignment>
01562   <assignment
01563     name="is_opened"
01564     type="bool">False</assignment>
01565  </state>
01566  <state
01567     state_id="expired"
01568     title="Expired">
01569   <description>Expiration date has passed</description>
01570   <exit-transition
01571     transition_id="open"/>
01572   <permission-map
01573     acquired="True"
01574     name="Modify content">
01575    <permission-role>Owner</permission-role>
01576    <permission-role>Manager</permission-role>
01577   </permission-map>
01578   <assignment
01579     name="is_closed"
01580     type="bool">False</assignment>
01581   <assignment
01582     name="is_opened"
01583     type="bool">False</assignment>
01584  </state>
01585  <state
01586     state_id="killed"
01587     title="Killed">
01588   <description>Permanently unavailable</description>
01589  </state>
01590  <state
01591     state_id="opened"
01592     title="Opened">
01593   <description>Open for modifications</description>
01594   <exit-transition
01595     transition_id="close"/>
01596   <exit-transition
01597     transition_id="kill"/>
01598   <exit-transition
01599     transition_id="expire"/>
01600   <permission-map
01601     acquired="True"
01602     name="Modify content">
01603    <permission-role>Owner</permission-role>
01604    <permission-role>Manager</permission-role>
01605   </permission-map>
01606   <group-map name="Content_owners">
01607    <group-role>Owner</group-role>
01608   </group-map>
01609   <assignment
01610     name="is_closed"
01611     type="bool">False</assignment>
01612   <assignment
01613     name="is_opened"
01614     type="bool">True</assignment>
01615  </state>
01616  <transition
01617     transition_id="close"
01618     title="Close"
01619     trigger="USER"
01620     new_state="closed"
01621     before_script=""
01622     after_script="after_close">
01623   <description>Close the object for modifications</description>
01624   <action
01625     category="workflow"
01626     url="string:${object_url}/close_for_modifications">Close</action>
01627   <guard>
01628    <guard-role>Owner</guard-role>
01629    <guard-role>Manager</guard-role>
01630   </guard>
01631  </transition>
01632  <transition
01633     transition_id="expire"
01634     title="Expire"
01635     trigger="AUTOMATIC"
01636     new_state="expired"
01637     before_script="before_expire"
01638     after_script="">
01639   <description>Retire objects whose expiration is past.</description>
01640   <guard>
01641    <guard-expression>python: object.expiration() &lt;= object.ZopeTime()</guard-expression>
01642   </guard>
01643   <assignment
01644     name="when_expired">object/ZopeTime</assignment>
01645  </transition>
01646  <transition
01647     transition_id="kill"
01648     title="Kill"
01649     trigger="USER"
01650     new_state="killed"
01651     before_script=""
01652     after_script="after_kill">
01653   <description>Make the object permanently unavailable.</description>
01654   <action
01655     category="workflow"
01656     url="string:${object_url}/kill_object">Kill</action>
01657   <guard>
01658    <guard-group>Content_assassins</guard-group>
01659   </guard>
01660   <assignment
01661     name="killed_by">string:${user/getId}</assignment>
01662  </transition>
01663  <transition
01664     transition_id="open"
01665     title="Open"
01666     trigger="USER"
01667     new_state="opened"
01668     before_script="before_open"
01669     after_script="">
01670   <description>Open the object for modifications</description>
01671   <action
01672     category="workflow"
01673     url="string:${object_url}/open_for_modifications">Open</action>
01674   <guard>
01675    <guard-permission>Open content for modifications</guard-permission>
01676   </guard>
01677   <assignment
01678     name="when_opened">object/ZopeTime</assignment>
01679  </transition>
01680  <worklist
01681     worklist_id="alive_list"
01682     title="Alive">
01683   <description>Worklist for content not yet expired / killed</description>
01684   <action
01685     category="workflow"
01686     url="string:${portal_url}/expired_items">Expired items</action>
01687   <guard>
01688    <guard-permission>Restore expired content</guard-permission>
01689   </guard>
01690   <match name="state" values="open; closed"/>
01691  </worklist>
01692  <worklist
01693     worklist_id="expired_list"
01694     title="Expired">
01695   <description>Worklist for expired content</description>
01696   <action
01697     category="workflow"
01698     url="string:${portal_url}/expired_items">Expired items</action>
01699   <guard>
01700    <guard-permission>Restore expired content</guard-permission>
01701   </guard>
01702   <match name="state" values="expired"/>
01703  </worklist>
01704  <variable
01705     variable_id="killed_by"
01706     for_catalog="True"
01707     for_status="False"
01708     update_always="True">
01709    <description>Killed by</description>
01710    <default>
01711     <value type="string">n/a</value>
01712    </default>
01713    <guard>
01714     <guard-role>Hangman</guard-role>
01715     <guard-role>Sherrif</guard-role>
01716    </guard>
01717  </variable>
01718  <variable
01719     variable_id="when_expired"
01720     for_catalog="True"
01721     for_status="False"
01722     update_always="True">
01723    <description>Expired when</description>
01724    <default>
01725     <expression>nothing</expression>
01726    </default>
01727    <guard>
01728     <guard-permission>Query history</guard-permission>
01729     <guard-permission>Open content for modifications</guard-permission>
01730    </guard>
01731  </variable>
01732  <variable
01733     variable_id="when_opened"
01734     for_catalog="True"
01735     for_status="False"
01736     update_always="True">
01737    <description>Opened when</description>
01738    <default>
01739     <expression>python:None</expression>
01740    </default>
01741    <guard>
01742     <guard-permission>Query history</guard-permission>
01743     <guard-permission>Open content for modifications</guard-permission>
01744    </guard>
01745  </variable>
01746  <script
01747     script_id="after_close"
01748     type="Script (Python)"
01749     filename="workflows/%(workflow_filename)s/scripts/after_close.py"
01750     module=""
01751     function=""
01752     />
01753  <script
01754     script_id="after_kill"
01755     type="Script (Python)"
01756     filename="workflows/%(workflow_filename)s/scripts/after_kill.py"
01757     module=""
01758     function=""
01759     />
01760  <script
01761     script_id="before_expire"
01762     type="External Method"
01763     filename=""
01764     module="DCWorkflow.test_method"
01765     function="test"
01766     />
01767  <script
01768     script_id="before_open"
01769     type="Script (Python)"
01770     filename="workflows/%(workflow_filename)s/scripts/before_open.py"
01771     module=""
01772     function=""
01773     />
01774 </dc-workflow>
01775 """
01776 
01777 
01778 class Test_exportWorkflow(_WorkflowSetup, _GuardChecker):
01779 
01780     layer = ExportImportZCMLLayer
01781 
01782     def test_empty( self ):
01783         from Products.CMFCore.exportimport.workflow import exportWorkflowTool
01784 
01785         site = self._initSite()
01786         context = DummyExportContext( site )
01787         exportWorkflowTool( context )
01788 
01789         self.assertEqual( len( context._wrote ), 1 )
01790         filename, text, content_type = context._wrote[ 0 ]
01791         self.assertEqual( filename, 'workflows.xml' )
01792         self._compareDOM( text, _EMPTY_TOOL_EXPORT )
01793         self.assertEqual( content_type, 'text/xml' )
01794 
01795     def test_normal( self ):
01796         from Products.CMFCore.exportimport.workflow import exportWorkflowTool
01797 
01798         WF_ID_NON = 'non_dcworkflow'
01799         WF_TITLE_NON = 'Non-DCWorkflow'
01800         WF_DESCRIPTION_NON = 'Not a DCWorkflow'
01801         WF_ID_DC = 'dcworkflow'
01802         WF_TITLE_DC = 'DCWorkflow'
01803         WF_DESCRIPTION_DC = 'I am a DCWorkflow'
01804         WF_INITIAL_STATE = 'closed'
01805 
01806         site = self._initSite()
01807 
01808         wf_tool = site.portal_workflow
01809         nondcworkflow = DummyWorkflow( WF_TITLE_NON )
01810         nondcworkflow.title = WF_TITLE_NON
01811         nondcworkflow.description = WF_DESCRIPTION_NON
01812         wf_tool._setObject( WF_ID_NON, nondcworkflow )
01813 
01814         dcworkflow = self._initDCWorkflow( WF_ID_DC )
01815         dcworkflow.title = WF_TITLE_DC
01816         dcworkflow.description = WF_DESCRIPTION_DC
01817         dcworkflow.initial_state = WF_INITIAL_STATE
01818         dcworkflow.permissions = _WF_PERMISSIONS
01819         self._initVariables( dcworkflow )
01820         self._initStates( dcworkflow )
01821         self._initTransitions( dcworkflow )
01822         self._initWorklists( dcworkflow )
01823         self._initScripts( dcworkflow )
01824 
01825         context = DummyExportContext( site )
01826         exportWorkflowTool( context )
01827 
01828         # workflows list, wf defintion and 3 scripts
01829         self.assertEqual( len( context._wrote ), 6 )
01830 
01831         filename, text, content_type = context._wrote[ 0 ]
01832         self.assertEqual( filename, 'workflows.xml' )
01833         self._compareDOM( text, _NORMAL_TOOL_EXPORT )
01834         self.assertEqual( content_type, 'text/xml' )
01835 
01836         filename, text, content_type = context._wrote[ 2 ]
01837         self.assertEqual( filename, 'workflows/%s/definition.xml' % WF_ID_DC )
01838         self._compareDOM( text
01839                         , _NORMAL_WORKFLOW_EXPORT
01840                           % { 'workflow_id' : WF_ID_DC
01841                             , 'title' : WF_TITLE_DC
01842                             , 'description' : WF_DESCRIPTION_DC
01843                             , 'initial_state' : WF_INITIAL_STATE
01844                             , 'workflow_filename' : WF_ID_DC.replace(' ', '_')
01845                             } )
01846         self.assertEqual( content_type, 'text/xml' )
01847 
01848         # just testing first script
01849         filename, text, content_type = context._wrote[ 3 ]
01850         self.assertEqual( filename, 'workflows/%s/scripts/after_close.py' % WF_ID_DC )
01851         self.assertEqual( text, _AFTER_CLOSE_SCRIPT)
01852         self.assertEqual( content_type, 'text/plain' )
01853 
01854     def test_with_filenames( self ):
01855         from Products.CMFCore.exportimport.workflow import exportWorkflowTool
01856 
01857         WF_ID_DC = 'name with spaces'
01858         WF_TITLE_DC = 'DCWorkflow with spaces'
01859         WF_DESCRIPTION_DC = 'Workflow w/spaces'
01860         WF_INITIAL_STATE = 'closed'
01861 
01862         site = self._initSite()
01863 
01864         dcworkflow = self._initDCWorkflow( WF_ID_DC )
01865         dcworkflow.title = WF_TITLE_DC
01866         dcworkflow.description = WF_DESCRIPTION_DC
01867         dcworkflow.initial_state = WF_INITIAL_STATE
01868         dcworkflow.permissions = _WF_PERMISSIONS
01869         self._initVariables( dcworkflow )
01870         self._initStates( dcworkflow )
01871         self._initTransitions( dcworkflow )
01872         self._initWorklists( dcworkflow )
01873         self._initScripts( dcworkflow )
01874 
01875         context = DummyExportContext( site )
01876         exportWorkflowTool( context )
01877 
01878         # workflows list, wf defintion and 3 scripts
01879         self.assertEqual( len( context._wrote ), 5 )
01880 
01881         filename, text, content_type = context._wrote[ 0 ]
01882         self.assertEqual( filename, 'workflows.xml' )
01883         self._compareDOM( text, _FILENAME_TOOL_EXPORT )
01884         self.assertEqual( content_type, 'text/xml' )
01885 
01886         filename, text, content_type = context._wrote[ 1 ]
01887         self.assertEqual( filename
01888                         , 'workflows/name_with_spaces/definition.xml' )
01889         self._compareDOM( text
01890                         , _NORMAL_WORKFLOW_EXPORT
01891                           % { 'workflow_id' : WF_ID_DC
01892                             , 'title' : WF_TITLE_DC
01893                             , 'description' : WF_DESCRIPTION_DC
01894                             , 'initial_state' : WF_INITIAL_STATE
01895                             , 'workflow_filename' : WF_ID_DC.replace(' ', '_')
01896                             } )
01897         self.assertEqual( content_type, 'text/xml' )
01898 
01899         # just testing first script
01900         filename, text, content_type = context._wrote[ 2 ]
01901         self.assertEqual( filename, 'workflows/%s/scripts/after_close.py' %
01902                           WF_ID_DC.replace(' ', '_'))
01903         self.assertEqual( text, _AFTER_CLOSE_SCRIPT)
01904         self.assertEqual( content_type, 'text/plain' )
01905 
01906 
01907 class Test_importWorkflow(_WorkflowSetup, _GuardChecker):
01908 
01909     layer = ExportImportZCMLLayer
01910 
01911     def _importNormalWorkflow( self, wf_id, wf_title, 
01912                                wf_description, wf_initial_state ):
01913         from Products.CMFCore.exportimport.workflow import importWorkflowTool
01914 
01915         site, context = self._prepareImportNormalWorkflow(
01916             wf_id, wf_title, wf_description, wf_initial_state)
01917 
01918         importWorkflowTool(context)
01919 
01920         return site.portal_workflow
01921 
01922     def _prepareImportNormalWorkflow(self, wf_id, wf_title, wf_description, 
01923                                      wf_initial_state, site=None, purge=True):
01924         if site is None:
01925             site = self._initSite()
01926         workflow_filename = wf_id.replace(' ', '_')
01927 
01928         context = DummyImportContext(site, purge=purge)
01929         context._files[ 'workflows.xml'
01930                       ] = (_NORMAL_TOOL_EXPORT_WITH_FILENAME
01931                             % { 'workflow_id' : wf_id
01932                               }
01933                           )
01934 
01935         context._files[ 'workflows/%s/definition.xml' % workflow_filename
01936                       ] = ( _NORMAL_WORKFLOW_EXPORT
01937                             % { 'workflow_id' : wf_id
01938                               , 'title' : wf_title
01939                               , 'description' : wf_description
01940                               , 'initial_state' : wf_initial_state
01941                               , 'workflow_filename' : workflow_filename
01942                               }
01943                           )
01944 
01945         context._files[ 'workflows/%s/scripts/after_close.py' % workflow_filename
01946                       ] = _AFTER_CLOSE_SCRIPT
01947 
01948         context._files[ 'workflows/%s/scripts/after_kill.py' % workflow_filename
01949                       ] = _AFTER_KILL_SCRIPT
01950 
01951         context._files[ 'workflows/%s/scripts/before_open.py' % workflow_filename
01952                       ] = _BEFORE_OPEN_SCRIPT
01953 
01954         return site, context
01955 
01956 
01957     def _importOldWorkflow( self, wf_id, wf_title, wf_initial_state ):
01958         from Products.CMFCore.exportimport.workflow import importWorkflowTool
01959 
01960         site = self._initSite()
01961         wf_tool = site.portal_workflow
01962         workflow_filename = wf_id.replace(' ', '_')
01963 
01964         context = DummyImportContext( site )
01965         context._files[ 'workflows.xml'
01966                       ] = (_NORMAL_TOOL_EXPORT_WITH_FILENAME
01967                             % { 'workflow_id' : wf_id
01968                               }
01969                           )
01970 
01971         context._files[ 'workflows/%s/definition.xml' % workflow_filename
01972                       ] = ( _OLD_WORKFLOW_EXPORT
01973                             % { 'workflow_id' : wf_id
01974                               , 'title' : wf_title
01975                               , 'initial_state' : wf_initial_state
01976                               , 'workflow_filename' : workflow_filename
01977                               }
01978                           )
01979 
01980         context._files[ 'workflows/%s/after_close.py' % workflow_filename
01981                       ] = _AFTER_CLOSE_SCRIPT
01982 
01983         context._files[ 'workflows/%s/after_kill.py' % workflow_filename
01984                       ] = _AFTER_KILL_SCRIPT
01985 
01986         context._files[ 'workflows/%s/before_open.py' % workflow_filename
01987                       ] = _BEFORE_OPEN_SCRIPT
01988 
01989         importWorkflowTool( context )
01990 
01991         return wf_tool
01992 
01993     def test_empty_default_purge( self ):
01994         from Products.CMFCore.exportimport.workflow import importWorkflowTool
01995 
01996         WF_ID_NON = 'non_dcworkflow_%s'
01997         WF_TITLE_NON = 'Non-DCWorkflow #%s'
01998 
01999         site = self._initSite()
02000         wf_tool = site.portal_workflow
02001 
02002         for i in range( 4 ):
02003             nondcworkflow = DummyWorkflow( WF_TITLE_NON % i )
02004             nondcworkflow.title = WF_TITLE_NON % i
02005             wf_tool._setObject( WF_ID_NON % i, nondcworkflow )
02006 
02007         wf_tool._default_chain = ( WF_ID_NON % 1, )
02008         wf_tool._chains_by_type[ 'sometype' ] = ( WF_ID_NON % 2, )
02009         self.assertEqual( len( wf_tool.objectIds() ), 4 )
02010 
02011         context = DummyImportContext( site )
02012         context._files[ 'workflows.xml' ] = _EMPTY_TOOL_EXPORT
02013         importWorkflowTool( context )
02014 
02015         self.assertEqual( len( wf_tool.objectIds() ), 0 )
02016         self.assertEqual( len( wf_tool._default_chain ), 0 )
02017         self.assertEqual( len( wf_tool._chains_by_type ), 0 )
02018 
02019     def test_empty_explicit_purge( self ):
02020         from Products.CMFCore.exportimport.workflow import importWorkflowTool
02021 
02022         WF_ID_NON = 'non_dcworkflow_%s'
02023         WF_TITLE_NON = 'Non-DCWorkflow #%s'
02024 
02025         site = self._initSite()
02026         wf_tool = site.portal_workflow
02027 
02028         for i in range( 4 ):
02029             nondcworkflow = DummyWorkflow( WF_TITLE_NON % i )
02030             nondcworkflow.title = WF_TITLE_NON % i
02031             wf_tool._setObject( WF_ID_NON % i, nondcworkflow )
02032 
02033         wf_tool._default_chain = ( WF_ID_NON % 1, )
02034         wf_tool._chains_by_type[ 'sometype' ] = ( WF_ID_NON % 2, )
02035         self.assertEqual( len( wf_tool.objectIds() ), 4 )
02036 
02037         context = DummyImportContext( site, True )
02038         context._files[ 'workflows.xml' ] = _EMPTY_TOOL_EXPORT
02039         importWorkflowTool( context )
02040 
02041         self.assertEqual( len( wf_tool.objectIds() ), 0 )
02042         self.assertEqual( len( wf_tool._default_chain ), 0 )
02043         self.assertEqual( len( wf_tool._chains_by_type ), 0 )
02044 
02045     def test_empty_skip_purge( self ):
02046         from Products.CMFCore.exportimport.workflow import importWorkflowTool
02047 
02048         WF_ID_NON = 'non_dcworkflow_%s'
02049         WF_TITLE_NON = 'Non-DCWorkflow #%s'
02050 
02051         site = self._initSite()
02052         wf_tool = site.portal_workflow
02053 
02054         for i in range( 4 ):
02055             nondcworkflow = DummyWorkflow( WF_TITLE_NON % i )
02056             nondcworkflow.title = WF_TITLE_NON % i
02057             wf_tool._setObject( WF_ID_NON % i, nondcworkflow )
02058 
02059         wf_tool._default_chain = ( WF_ID_NON % 1, )
02060         wf_tool._chains_by_type[ 'sometype' ] = ( WF_ID_NON % 2, )
02061         self.assertEqual( len( wf_tool.objectIds() ), 4 )
02062 
02063         context = DummyImportContext( site, False )
02064         context._files[ 'typestool.xml' ] = _EMPTY_TOOL_EXPORT
02065         importWorkflowTool( context )
02066 
02067         self.assertEqual( len( wf_tool.objectIds() ), 4 )
02068         self.assertEqual( len( wf_tool._default_chain ), 1 )
02069         self.assertEqual( wf_tool._default_chain[ 0 ], WF_ID_NON % 1 )
02070         self.assertEqual( len( wf_tool._chains_by_type ), 1 )
02071         self.assertEqual( wf_tool._chains_by_type[ 'sometype' ]
02072                         , ( WF_ID_NON % 2, )
02073                         )
02074 
02075     def test_bindings_skip_purge( self ):
02076         from Products.CMFCore.exportimport.workflow import importWorkflowTool
02077 
02078         WF_ID_NON = 'non_dcworkflow_%s'
02079         WF_TITLE_NON = 'Non-DCWorkflow #%s'
02080 
02081         site = self._initSite()
02082         wf_tool = site.portal_workflow
02083 
02084         for i in range( 4 ):
02085             nondcworkflow = DummyWorkflow( WF_TITLE_NON % i )
02086             nondcworkflow.title = WF_TITLE_NON % i
02087             wf_tool._setObject( WF_ID_NON % i, nondcworkflow )
02088 
02089         wf_tool._default_chain = ( WF_ID_NON % 1, )
02090         wf_tool._chains_by_type[ 'sometype' ] = ( WF_ID_NON % 2, )
02091         self.assertEqual( len( wf_tool.objectIds() ), 4 )
02092 
02093         context = DummyImportContext( site, False )
02094         context._files[ 'workflows.xml' ] = _BINDINGS_TOOL_EXPORT
02095         importWorkflowTool( context )
02096 
02097         self.assertEqual( len( wf_tool.objectIds() ), 4 )
02098         self.assertEqual( len( wf_tool._default_chain ), 2 )
02099         self.assertEqual( wf_tool._default_chain[ 0 ], WF_ID_NON % 0 )
02100         self.assertEqual( wf_tool._default_chain[ 1 ], WF_ID_NON % 1 )
02101         self.assertEqual( len( wf_tool._chains_by_type ), 2 )
02102         self.assertEqual( wf_tool._chains_by_type[ 'sometype' ]
02103                         , ( WF_ID_NON % 2, )
02104                         )
02105         self.assertEqual( wf_tool._chains_by_type[ 'anothertype' ]
02106                         , ( WF_ID_NON % 3, )
02107                         )
02108 
02109     def test_import_twice_nopurge(self):
02110         from Products.CMFCore.exportimport.workflow import importWorkflowTool
02111 
02112         WF_ID = 'dcworkflow_purge'
02113         WF_TITLE = 'DC Workflow testing purge'
02114         WF_DESCRIPTION = 'Test Purge'
02115         WF_INITIAL_STATE = 'closed'
02116 
02117         # Import a first time
02118         site, context = self._prepareImportNormalWorkflow(
02119             WF_ID, WF_TITLE, WF_DESCRIPTION, WF_INITIAL_STATE)
02120         importWorkflowTool(context)
02121 
02122         # Now reimport without purge
02123         site, context = self._prepareImportNormalWorkflow(
02124             WF_ID, WF_TITLE, WF_DESCRIPTION, WF_INITIAL_STATE, 
02125             site=site, purge=False)
02126         importWorkflowTool(context)
02127         workflow = site.portal_workflow.objectValues()[1]
02128 
02129         self.assertEqual(workflow.getId(), WF_ID)
02130         self.assertEqual(workflow.meta_type, DCWorkflowDefinition.meta_type)
02131         self.assertEqual(workflow.title, WF_TITLE)
02132         self.assertEqual(workflow.state_var, 'state')
02133         self.assertEqual(workflow.initial_state, WF_INITIAL_STATE)
02134         self.assertEqual(len(workflow.variables.objectItems()),
02135                          len(_WF_VARIABLES))
02136         self.assertEqual(len(workflow.states.objectItems()),
02137                          len(_WF_STATES))
02138         self.assertEqual(len(workflow.transitions.objectItems()),
02139                          len(_WF_TRANSITIONS))
02140         self.assertEqual(len(workflow.permissions),
02141                          len(_WF_PERMISSIONS))
02142         self.assertEqual(len(workflow.scripts.objectItems()),
02143                          len(_WF_SCRIPTS))
02144         self.assertEqual(len(workflow.worklists.objectItems()),
02145                          len(_WF_WORKLISTS))
02146 
02147 
02148     def test_from_empty_dcworkflow_top_level( self ):
02149 
02150         WF_ID = 'dcworkflow_tool'
02151         WF_TITLE = 'DC Workflow testing tool'
02152         WF_DESCRIPTION = 'Testing Tool'
02153         WF_INITIAL_STATE = 'closed'
02154 
02155         tool = self._importNormalWorkflow( WF_ID, WF_TITLE,
02156                                            WF_DESCRIPTION, WF_INITIAL_STATE )
02157 
02158         self.assertEqual( len( tool.objectIds() ), 2 )
02159         self.assertEqual( tool.objectIds()[ 1 ], WF_ID )
02160 
02161     def test_from_empty_dcworkflow_workflow_attrs( self ):
02162 
02163         WF_ID = 'dcworkflow_attrs'
02164         WF_TITLE = 'DC Workflow testing attrs'
02165         WF_DESCRIPTION = 'Testing Attributes'
02166         WF_INITIAL_STATE = 'closed'
02167 
02168         tool = self._importNormalWorkflow( WF_ID, WF_TITLE,
02169                                            WF_DESCRIPTION, WF_INITIAL_STATE )
02170 
02171         workflow = tool.objectValues()[ 1 ]
02172         self.assertEqual( workflow.meta_type, DCWorkflowDefinition.meta_type )
02173         self.assertEqual( workflow.title, WF_TITLE )
02174         self.assertEqual( workflow.state_var, 'state' )
02175         self.assertEqual( workflow.initial_state, WF_INITIAL_STATE )
02176 
02177     def test_from_empty_dcworkflow_workflow_permissions( self ):
02178 
02179         WF_ID = 'dcworkflow_permissions'
02180         WF_TITLE = 'DC Workflow testing permissions'
02181         WF_DESCRIPTION = 'Testing Permissions'
02182         WF_INITIAL_STATE = 'closed'
02183 
02184         tool = self._importNormalWorkflow( WF_ID, WF_TITLE,
02185                                            WF_DESCRIPTION, WF_INITIAL_STATE )
02186 
02187         workflow = tool.objectValues()[ 1 ]
02188 
02189         permissions = workflow.permissions
02190         self.assertEqual( len( permissions ), len( _WF_PERMISSIONS ) )
02191 
02192         for permission in permissions:
02193             self.failUnless( permission in _WF_PERMISSIONS )
02194 
02195     def test_from_empty_dcworkflow_workflow_variables( self ):
02196 
02197         WF_ID = 'dcworkflow_variables'
02198         WF_TITLE = 'DC Workflow testing variables'
02199         WF_DESCRIPTION = 'Testing Variables'
02200         WF_INITIAL_STATE = 'closed'
02201 
02202         tool = self._importNormalWorkflow( WF_ID, WF_TITLE,
02203                                            WF_DESCRIPTION, WF_INITIAL_STATE )
02204 
02205         workflow = tool.objectValues()[ 1 ]
02206 
02207         variables = workflow.variables
02208 
02209         self.assertEqual( len( variables.objectItems() )
02210                         , len( _WF_VARIABLES ) )
02211 
02212         for id, variable in variables.objectItems():
02213 
02214             expected = _WF_VARIABLES[ variable.getId() ]
02215             self.failUnless( expected[ 0 ] in variable.description )
02216             self.assertEqual( variable.default_value, expected[ 1 ] )
02217             self.assertEqual( variable.getDefaultExprText(), expected[ 2 ] )
02218             self.assertEqual( variable.for_catalog, expected[ 3 ] )
02219             self.assertEqual( variable.for_status, expected[ 4 ] )
02220             self.assertEqual( variable.update_always, expected[ 5 ] )
02221 
02222             guard = variable.getInfoGuard()
02223 
02224             self.assertEqual( guard.permissions, expected[ 6 ] )
02225             self.assertEqual( guard.roles, expected[ 7 ] )
02226             self.assertEqual( guard.groups, expected[ 8 ] )
02227             self.assertEqual( guard.getExprText(), expected[ 9 ] )
02228 
02229     def test_from_empty_dcworkflow_workflow_states( self ):
02230 
02231         WF_ID = 'dcworkflow_states'
02232         WF_TITLE = 'DC Workflow testing states'
02233         WF_DESCRIPTION = 'Testing States'
02234         WF_INITIAL_STATE = 'closed'
02235 
02236         tool = self._importNormalWorkflow( WF_ID, WF_TITLE,
02237                                            WF_DESCRIPTION, WF_INITIAL_STATE )
02238 
02239         workflow = tool.objectValues()[ 1 ]
02240 
02241         states = workflow.states
02242 
02243         self.assertEqual( len( states.objectItems() )
02244                         , len( _WF_STATES ) )
02245 
02246         for id, state in states.objectItems():
02247 
02248             expected = _WF_STATES[ state.getId() ]
02249             self.assertEqual( state.title, expected[ 0 ] )
02250             self.failUnless( expected[ 1 ] in state.description )
02251 
02252             self.assertEqual( len( state.transitions ), len( expected[ 2 ] ) )
02253 
02254             for transition_id in state.transitions:
02255                 self.failUnless( transition_id in expected[ 2 ] )
02256 
02257             for permission in state.getManagedPermissions():
02258 
02259                 p_info = state.getPermissionInfo( permission )
02260                 p_expected = expected[ 3 ].get( permission, [] )
02261 
02262                 self.assertEqual( bool( p_info[ 'acquired' ] )
02263                                 , isinstance(p_expected, list) )
02264 
02265                 self.assertEqual( len( p_info[ 'roles' ] ), len( p_expected ) )
02266 
02267                 for role in p_info[ 'roles' ]:
02268                     self.failIf( role not in p_expected )
02269 
02270             group_roles = state.group_roles or {}
02271             self.assertEqual( len( group_roles ), len( expected[ 4 ] ) )
02272 
02273             for group_id, exp_roles in expected[ 4 ]:
02274 
02275                 self.assertEqual( len( state.getGroupInfo( group_id ) )
02276                                 , len( exp_roles ) )
02277 
02278                 for role in state.getGroupInfo( group_id ):
02279                     self.failUnless( role in exp_roles )
02280 
02281             self.assertEqual( len( state.getVariableValues() )
02282                             , len( expected[ 5 ] ) )
02283 
02284             for var_id, value in state.getVariableValues():
02285 
02286                 self.assertEqual( value, expected[ 5 ][ var_id ] )
02287 
02288     def test_from_empty_dcworkflow_workflow_transitions( self ):
02289 
02290         WF_ID = 'dcworkflow_transitions'
02291         WF_TITLE = 'DC Workflow testing transitions'
02292         WF_DESCRIPTION = 'Testing Transitions'
02293         WF_INITIAL_STATE = 'closed'
02294 
02295         tool = self._importNormalWorkflow( WF_ID, WF_TITLE,
02296                                            WF_DESCRIPTION, WF_INITIAL_STATE )
02297 
02298         workflow = tool.objectValues()[ 1 ]
02299 
02300         transitions = workflow.transitions
02301 
02302         self.assertEqual( len( transitions.objectItems() )
02303                         , len( _WF_TRANSITIONS ) )
02304 
02305         for id, transition in transitions.objectItems():
02306 
02307             expected = _WF_TRANSITIONS[ transition.getId() ]
02308             self.assertEqual( transition.title, expected[ 0 ] )
02309             self.failUnless( expected[ 1 ] in transition.description )
02310             self.assertEqual( transition.new_state_id, expected[ 2 ] )
02311             self.assertEqual( transition.trigger_type, expected[ 3 ] )
02312             self.assertEqual( transition.script_name, expected[ 4 ] )
02313             self.assertEqual( transition.after_script_name, expected[ 5 ] )
02314             self.assertEqual( transition.actbox_name, expected[ 6 ] )
02315             self.assertEqual( transition.actbox_url, expected[ 7 ] )
02316             self.assertEqual( transition.actbox_category, expected[ 8 ] )
02317 
02318             var_exprs = transition.var_exprs
02319 
02320             self.assertEqual( len( var_exprs ), len( expected[ 9 ] ) )
02321 
02322             for var_id, expr in var_exprs.items():
02323                 self.assertEqual( expr.text, expected[ 9 ][ var_id ] )
02324 
02325             guard = transition.getGuard()
02326 
02327             self.assertEqual( guard.permissions, expected[ 10 ] )
02328             self.assertEqual( guard.roles, expected[ 11 ] )
02329             self.assertEqual( guard.groups, expected[ 12 ] )
02330             self.assertEqual( guard.getExprText(), expected[ 13 ] )
02331 
02332     def test_from_empty_dcworkflow_workflow_worklists( self ):
02333 
02334         WF_ID = 'dcworkflow_worklists'
02335         WF_TITLE = 'DC Workflow testing worklists'
02336         WF_DESCRIPTION = 'Testing Worklists'
02337         WF_INITIAL_STATE = 'closed'
02338 
02339         tool = self._importNormalWorkflow( WF_ID, WF_TITLE, 
02340                                            WF_DESCRIPTION, WF_INITIAL_STATE )
02341 
02342         workflow = tool.objectValues()[ 1 ]
02343 
02344         worklists = workflow.worklists
02345 
02346         self.assertEqual( len( worklists.objectItems() )
02347                         , len( _WF_WORKLISTS ) )
02348 
02349         for id, worklist in worklists.objectItems():
02350 
02351             expected = _WF_WORKLISTS[ worklist.getId() ]
02352             self.failUnless( expected[ 1 ] in worklist.description )
02353 
02354             var_matches = worklist.var_matches
02355 
02356             self.assertEqual( len( var_matches ), len( expected[ 2 ] ) )
02357 
02358             for var_id, values in var_matches.items():
02359                 exp_values = expected[ 2 ][ var_id ]
02360                 self.assertEqual( len( values ), len( exp_values ) )
02361 
02362                 for value in values:
02363                     self.failUnless( value in exp_values, values )
02364 
02365             self.assertEqual( worklist.actbox_name, expected[ 3 ] )
02366             self.assertEqual( worklist.actbox_url, expected[ 4 ] )
02367             self.assertEqual( worklist.actbox_category, expected[ 5 ] )
02368 
02369             guard = worklist.getGuard()
02370 
02371             self.assertEqual( guard.permissions, expected[ 6 ] )
02372             self.assertEqual( guard.roles, expected[ 7 ] )
02373             self.assertEqual( guard.groups, expected[ 8 ] )
02374             self.assertEqual( guard.getExprText(), expected[ 9 ] )
02375 
02376     def test_from_old_dcworkflow_workflow_scripts( self ):
02377 
02378         WF_ID = 'old_dcworkflow_scripts'
02379         WF_TITLE = 'Old DC Workflow testing scripts'
02380         WF_INITIAL_STATE = 'closed'
02381 
02382         tool = self._importOldWorkflow( WF_ID, WF_TITLE, WF_INITIAL_STATE )
02383 
02384         workflow = tool.objectValues()[ 1 ]
02385 
02386         scripts = workflow.scripts
02387 
02388         self.assertEqual( len( scripts.objectItems() )
02389                         , len( _WF_SCRIPTS ) )
02390 
02391         for script_id, script in scripts.objectItems():
02392 
02393             expected = _WF_SCRIPTS[ script_id ]
02394 
02395             self.assertEqual( script.meta_type, expected[ 0 ] )
02396 
02397             if script.meta_type == PythonScript.meta_type:
02398                 self.assertEqual( script.manage_FTPget(), expected[ 1 ] )
02399 
02400     def test_from_empty_dcworkflow_workflow_scripts( self ):
02401 
02402         WF_ID = 'dcworkflow_scripts'
02403         WF_TITLE = 'DC Workflow testing scripts'
02404         WF_DESCRIPTION = 'Testing Scripts'
02405         WF_INITIAL_STATE = 'closed'
02406 
02407         tool = self._importNormalWorkflow( WF_ID, WF_TITLE, 
02408                                            WF_DESCRIPTION, WF_INITIAL_STATE )
02409 
02410         workflow = tool.objectValues()[ 1 ]
02411 
02412         scripts = workflow.scripts
02413 
02414         self.assertEqual( len( scripts.objectItems() )
02415                         , len( _WF_SCRIPTS ) )
02416 
02417         for script_id, script in scripts.objectItems():
02418 
02419             expected = _WF_SCRIPTS[ script_id ]
02420 
02421             self.assertEqual( script.meta_type, expected[ 0 ] )
02422 
02423             if script.meta_type == PythonScript.meta_type:
02424                 self.assertEqual( script.manage_FTPget(), expected[ 1 ] )
02425 
02426     def test_scripts_with_invalid_meta_type(self):
02427         """
02428         A script with an invalid meta_type should raise an error.
02429 
02430         Otherwise the previous script will be added for that script.
02431         """
02432         from Products.DCWorkflow import exportimport
02433 
02434         tool = self._importNormalWorkflow(
02435             'dcworkflow_scripts', 'DC Workflow testing scripts',
02436             'Testing Scripts', 'closed')
02437         workflow = tool.objectValues()[1]
02438         scripts = workflow.scripts
02439 
02440         s_infos = [
02441             dict(script_id='invalid', meta_type='invalid',
02442                  filename='')]
02443         self.assertRaises(ValueError, 
02444                           exportimport._initDCWorkflowScripts,
02445                           workflow, s_infos, None)
02446 
02447     def test_scripts_by_meta_type(self):
02448         """
02449         Constructors for meta_types other than those hard coded should
02450         be looked up.
02451         """
02452         from Products.DCWorkflow import exportimport
02453 
02454         tool = self._importNormalWorkflow(
02455             'dcworkflow_scripts', 'DC Workflow testing scripts',
02456             'Testing Scripts', 'closed')
02457         workflow = tool.objectValues()[1]
02458         scripts = workflow.scripts
02459 
02460         scripts.all_meta_types = scripts.all_meta_types() + [
02461             dict(instance=PythonScript, name='Foo Script')]
02462 
02463         s_infos = [
02464             dict(script_id='doc', meta_type='DTML Document',
02465                  filename=''),
02466             dict(script_id='bar', meta_type='Foo Script',
02467                  filename='')]
02468         exportimport._initDCWorkflowScripts(workflow, s_infos, None)
02469         
02470         self.assertEqual(scripts['doc'].meta_type, 'DTML Document')
02471         self.assertEqual(scripts['bar'].meta_type, 'Script (Python)')
02472 
02473 def test_suite():
02474     return unittest.TestSuite((
02475         unittest.makeSuite( WorkflowDefinitionConfiguratorTests ),
02476         unittest.makeSuite( Test_exportWorkflow ),
02477         unittest.makeSuite( Test_importWorkflow ),
02478         ))
02479 
02480 if __name__ == '__main__':
02481     from Products.CMFCore.testing import run
02482     run(test_suite())