ERP5Workflow: improving testWorklist

bt5/erp5_workflow_test/TestTemplateItem/portal_components/test.erp5.testWorkflowMixin.py

 from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
 
 class testWorkflowMixin(ERP5TypeTestCase):
-  def getWorklistDocumentCountFromActionName(self, action_name):
+  def countFromActionName(self, action_name):
+    # action_name look like: "Documents to validate (3)"
     self.assertEqual(action_name[-1], ')')
     left_parenthesis_offset = action_name.rfind('(')
     self.assertNotEquals(left_parenthesis_offset, -1)
     return int(action_name[left_parenthesis_offset + 1:-1])
 
-  def checkWorklist(self, action_list, name, count, url_parameter_dict=None, workflow_id=None):
+  def checkWorklist(self, action_list, name, count, url_parameter_dict=None, workflow_id=None,
+                   selection_name=None):
     entry_list = [
       x for x in action_list if x['name'].startswith(name)
                              and (
@@ -16,19 +18,22 @@ class testWorkflowMixin(ERP5TypeTestCase):
                                   and x['workflow_id'] == workflow_id
                              )
     ]
+
+    # ensure there is a single entry in action list
     self.assertEqual(len(entry_list), count and 1)
     if count:
       self.assertEqual(count,
-        self.getWorklistDocumentCountFromActionName(entry_list[0]['name']))
+        self.countFromActionName(entry_list[0]['name']))
     if entry_list and url_parameter_dict:
       url = entry_list[0].get('url')
       self.assertTrue(url, 'Can not check url parameters without url')
       url = '%s%s' % (self.portal.getId(), url[len(self.portal.absolute_url()):])
       # Touch URL to save worklist parameters in listbox selection
-      self.publish(url, 'manager:') # XXX: troubles running live test, returns HTTP error 500
+      publish_response = self.publish(url, 'manager:') # XXX: troubles running live test, returns HTTP error 500
+      self.assertEqual(publish_response.status, 200)
       self.commit()
       selection_parameter_dict = self.portal.portal_selections.getSelectionParamsFor(
-                                                    self.module_selection_name)
+                                                    selection_name)
       for parameter, value in url_parameter_dict.iteritems():
         self.assertIn(parameter, selection_parameter_dict)
         self.assertEqual(value, selection_parameter_dict[parameter])

bt5/erp5_workflow_test/TestTemplateItem/portal_components/test.erp5.testWorklist.py

@@ -6,32 +6,27 @@ class TestWorklist(testWorkflowMixin):
   run_all_test = 1
   quiet = 1
 
-  checked_portal_type = 'Organisation'
-  module_selection_name = 'organisation_module_selection'
-  checked_validation_state = 'draft'
-  not_checked_validation_state = 'not_draft'
-  checked_workflow = 'validation_workflow'
-  worklist_assignor_id = 'assignor_worklist'
-  actbox_assignor_name = 'assignor_todo'
-  worklist_owner_id = 'owner_worklist'
-  actbox_owner_name = 'owner_todo'
-  worklist_assignor_owner_id = 'assignor_owner_worklist'
-  actbox_assignor_owner_name = 'assignor_owner_todo'
-  worklist_desactivated_id = '%s_desactivated' % worklist_owner_id
-  actbox_desactivated_by_expression = '%s_desactivated' % actbox_owner_name
-  worklist_wrong_state_id = '%s_wrong_state' % worklist_owner_id
-  actbox_wrong_state = '%s_wrong_state' % actbox_owner_name
-
-  worklist_int_variable_id = 'int_value_worklist'
-  actbox_int_variable_name = 'int_value_todo'
-  int_catalogued_variable_id = 'int_index'
-  int_value = 1
-
   user_dict = {
     'foo': [None, None],
     'bar': [None, None],
   }
 
+  test_worklist_dict = {
+    'assignor_worklist':
+      {'actbox_name': 'assignor_todo', 'role': 'Assignor', 'expr': None, 'state': 'draft', 'int_variable': None},
+    'owner_worklist':
+      {'actbox_name': 'owner_todo', 'role': 'Owner', 'expr': None, 'state': 'draft', 'int_variable': None},
+    'owner_worklist_desactivated':
+      {'actbox_name':'owner_todo_desactivated', 'role': 'Owner', 'expr': 'python: 0', 'state': 'draft', 'int_variable': None},
+    'owner_worklist_wrong_state':
+      {'actbox_name':'owner_todo_wrong_state', 'role': 'Owner', 'expr': None, 'state': 'not_draft', 'int_variable': None},
+    'assignor_owner_worklist':
+      {'actbox_name':'assignor_owner_todo', 'role': 'Assignor ; Owner','expr':  None, 'state': 'draft', 'int_variable': None},
+    'int_value_worklist':
+      {'actbox_name':'int_value_todo', 'role': None, 'expr': None, 'state': None, 'int_variable': str(1)}
+  }
+
+
   def getTitle(self):
     return "Worklist"
 
@@ -91,21 +86,20 @@ class TestWorklist(testWorkflowMixin):
         group = "%s" % user_data[0],
         function = "%s" % user_data[1],
         start_date = '01/01/1900',
-        stop_date = '01/01/2900',
       )
       assignment.open()
     # Reindexing is required for the security to work
     self.tic()
 
   def createDocument(self, **kw):
-    module = self.getPortal().getDefaultModule(self.checked_portal_type)
-    result = module.newContent(portal_type=self.checked_portal_type, **kw)
-    result.setProperty(self.int_catalogued_variable_id, self.int_value)
-    assert result.getValidationState() == self.checked_validation_state
+    module = self.getPortal().getDefaultModule('Organisation')
+    result = module.newContent(portal_type='Organisation', **kw)
+    result.setProperty('int_index', 1)
+    assert result.getValidationState() == 'draft'
     return result
 
   def associatePropertySheet(self):
-    self._addPropertySheet(self.checked_portal_type, 'SortIndex')
+    self._addPropertySheet('Organisation', 'SortIndex')
 
   def addWorkflowCataloguedVariable(self, workflow_id, variable_id):
     # Add new workflow compatibility
@@ -118,65 +112,56 @@ class TestWorklist(testWorkflowMixin):
       variable_value = variables[variable_id]
       assert variable_value.for_catalog == 1
 
-  def createWorklist(self, workflow_id, worklist_id, actbox_name,
-                     actbox_url=None, **kw):
+  def createWorklist(self, workflow_id, *args, **kw):
     workflow_value = self.getWorkflowTool()[workflow_id]
-    actbox_name='%s (%%(count)s)' % actbox_name
+    if workflow_value.__class__.__name__ == 'Workflow':
+      self.createERP5Worklist(workflow_value,*args, **kw)
+    else:
+      self.createDCWorklist(workflow_value, *args, **kw)
 
-    # add new workflow compatibility
+  def createERP5Worklist(self, workflow_value, worklist_id, actbox_name,
+                     actbox_url=None, portal_type=None, validation_state=None,
+                     guard_roles='', guard_expr=None, **kw):
+    actbox_name='%s (%%(count)s)' % actbox_name
     if workflow_value.__class__.__name__ == 'Workflow':
       if getattr(workflow_value, worklist_id, None):
         workflow_value.manage_delObjects([worklist_id])
       worklist_value = workflow_value.newContent(portal_type='Worklist')
-      worklist_value.setReference(worklist_id)
-
-      objectAndMethodToApplyArgumentTo = [
-        (worklist_value, 'setActboxName', actbox_name),
-        (worklist_value, 'setActboxUrl', str(actbox_url)),
-        (worklist_value, 'setActboxCategory', 'global'),
-        (worklist_value, 'setMatchedPortalTypeList', kw.get('portal_type')),
-        (worklist_value, 'setMatchedValidationState', '' if not kw.get('validation_state') \
-                                                      else 'state_' + kw.get('validation_state'))
-      ]
-
-      for (kw_key, method) in [(self.int_catalogued_variable_id, 'setInitialValue'),
-                               ('region_uid', 'setDefaultExpr'),
-                               ('base_category_id', 'setInitialValue')]:
-        var_value = kw.get(kw_key, None)
-        var_object = None
-        # Create variable for worklist if not exists:
-        if var_value is not None:
-          var_object = worklist_value._getOb(kw_key, None)
-          if var_object is None:
-            var_object = worklist_value.newContent(portal_type='Worklist Variable')
-            var_object.setReference(kw_key)
-        objectAndMethodToApplyArgumentTo.append(var_object, method, var_value)
-
-      guard_roles_string = kw.get('guard_roles', '')
-      guard_role_list = [ var.strip() for var in guard_roles_string.split(';') ]
-      objectAndMethodToApplyArgumentTo += [
-        (worklist_value, 'setRoleList', guard_role_list),
-        (worklist_value, 'setExpression', kw.get('guard_expr'))
-      ]
-
-      # Configure new worklist:
-      for (obj, method_name, argument) in objectAndMethodToApplyArgumentTo:
-        method = (getattr(obj, method_name, None))
-        if method is not None and argument not in [None, '']:
-          method(argument)
-
-      worklist_value.getGuard()
-    else:
-      worklists = workflow_value.worklists
-      if worklists._getOb(worklist_id, None):
-        worklists.deleteWorklists([worklist_id])
-      worklists.addWorklist(worklist_id)
-      worklist_value = worklists._getOb(worklist_id)
-      worklist_value.setProperties('',
-        actbox_name=actbox_name, actbox_url=actbox_url,
-        props={k if k.startswith('guard_') else 'var_match_' + k: v
-               for k, v in kw.iteritems()})
+      guard_roles = [] if not guard_roles else [role.strip() for role in guard_roles.split(';')]
+      # XXX(WORKFLOW), it would be much nicer to use categories instead of ids like 'state_draft',
+      # this way we would be consistent with what is used in transitions
+      validation_state = None if not validation_state else 'state_' + validation_state
+
+      worklist_value.edit(
+        reference=worklist_id,
+        actbox_name=actbox_name,
+        actbox_url=actbox_url,
+        actbox_category='global',
+        matched_validation_state=validation_state,
+        matched_portal_type_list=portal_type,
+        role_list=guard_roles,
+        expression=guard_expr
+      )
+
+      for _, value in kw.iteritems():
+        if isinstance(value, str) and value.startswith('python'):
+          # XXX(WORKFLOW), default expr does not follow naming conventions
+          worklist_value.newContent(portal_type='Worklist Variable', default_expr=value)
+        else:
+          worklist_value.newContent(portal_type='Worklist Variable', intial_value=value)
+
+  def createDCWorklist(self, workflow_value, worklist_id, actbox_name,
+                       actbox_url=None, **kw):
+    actbox_name='%s (%%(count)s)' % actbox_name
 
+    worklists = workflow_value.worklists
+    if worklists._getOb(worklist_id, None):
+      worklists.deleteWorklists([worklist_id])
+    worklists.addWorklist(worklist_id)
+    worklist_value = worklists._getOb(worklist_id)
+    worklist_value.setProperties('', actbox_name=actbox_name, actbox_url=actbox_url,
+                                 props={k if k.startswith('guard_') else 'var_match_' + k: v
+                                          for k, v in kw.iteritems()})
 
   def removeWorklist(self, workflow_id, worklist_id_list):
     # add new workflow compatibility
@@ -188,37 +173,6 @@ class TestWorklist(testWorkflowMixin):
       worklists = self.getWorkflowTool()[workflow_id].worklists
       worklists.deleteWorklists(worklist_id_list)
 
-  def createWorklists(self):
-    for worklist_id, actbox_name, role, expr, state, int_variable in [
-          (self.worklist_assignor_id, self.actbox_assignor_name,
-           'Assignor', None, self.checked_validation_state, None),
-          (self.worklist_owner_id, self.actbox_owner_name,
-           'Owner', None, self.checked_validation_state, None),
-          (self.worklist_desactivated_id, self.actbox_desactivated_by_expression,
-           'Owner', 'python: 0', self.checked_validation_state, None),
-          (self.worklist_wrong_state_id, self.actbox_wrong_state,
-           'Owner', None, self.not_checked_validation_state, None),
-          (self.worklist_assignor_owner_id, self.actbox_assignor_owner_name,
-           'Assignor; Owner', None, self.checked_validation_state, None),
-          (self.worklist_int_variable_id, self.actbox_int_variable_name,
-           None, None, None, str(self.int_value)),
-    ]:
-      self.createWorklist(self.checked_workflow, worklist_id, actbox_name,
-                          guard_roles=role, guard_expr=expr,
-                          portal_type=self.checked_portal_type,
-                          validation_state=state,
-                          **{self.int_catalogued_variable_id: int_variable})
-
-  def removeWorklists(self):
-    self.removeWorklist(self.checked_workflow, [
-          self.worklist_assignor_id,
-          self.worklist_owner_id,
-          self.worklist_desactivated_id,
-          self.worklist_wrong_state_id,
-          self.worklist_assignor_owner_id,
-          self.worklist_int_variable_id,
-    ])
-
   def createCategories(self):
     category_tool = self.getCategoryTool()
     for base_category, category_list in (
@@ -229,7 +183,6 @@ class TestWorklist(testWorkflowMixin):
         if not getattr(category_tool[base_category], category, None):
           newContent(portal_type='Category', id=category)
 
-
   def test_01_permission(self, quiet=0, run=run_all_test):
     """
     Test the permission of the building module.
@@ -243,9 +196,19 @@ class TestWorklist(testWorkflowMixin):
 
     self.logMessage("Create worklists")
     self.associatePropertySheet()
-    self.addWorkflowCataloguedVariable(self.checked_workflow,
-                                       self.int_catalogued_variable_id)
-    self.createWorklists()
+    self.addWorkflowCataloguedVariable('validation_workflow',
+                                       'int_index')
+    for worklist_id in self.test_worklist_dict.keys():
+      worklist = self.test_worklist_dict[worklist_id]
+      self.createWorklist('validation_workflow',
+                          worklist_id,
+                          worklist['actbox_name'],
+                          guard_roles=worklist['role'],
+                          guard_expr=worklist['expr'],
+                          portal_type='Organisation',
+                          validation_state=worklist['state'],
+                          int_index=worklist['int_variable'])
+    self.tic()
     try:
       self.logMessage("Create document as Manager")
       document = self.createDocument()
@@ -253,16 +216,15 @@ class TestWorklist(testWorkflowMixin):
       self.tic()
       self.clearCache()
 
-      result = workflow_tool.listActions(object=document)
-
       # Users can not see worklist as they are not Assignor
       for user_id in ('manager', ):
         self.login(user_id)
         result = workflow_tool.listActions(object=document)
+
         self.logMessage("Check %s worklist as Assignor" % user_id)
-        self.checkWorklist(result, self.actbox_assignor_name, 0)
+        self.checkWorklist(result, 'assignor_todo', 0)
         self.logMessage("Check %s worklist as Owner" % user_id)
-        self.checkWorklist(result, self.actbox_owner_name, 1)
+        self.checkWorklist(result, 'owner_todo', 1)
       for user_id in ('foo', 'bar'):
         self.logMessage("Check %s worklist" % user_id)
         self.login(user_id)
@@ -285,36 +247,35 @@ class TestWorklist(testWorkflowMixin):
           self.login(user_id)
           result = workflow_tool.listActions(object=document)
           self.logMessage("  Check %s worklist as Assignor" % user_id)
-          self.checkWorklist(result, self.actbox_assignor_name, assignor)
+          self.checkWorklist(result, 'assignor_todo', assignor)
           self.logMessage("  Check %s worklist as Owner" % user_id)
-          self.checkWorklist(result, self.actbox_owner_name, owner)
+          self.checkWorklist(result, 'owner_todo', owner)
           self.logMessage("  Check %s worklist as Owner and Assignor" % user_id)
-          self.checkWorklist(result, self.actbox_assignor_owner_name, both)
+          self.checkWorklist(result, 'assignor_owner_todo', both)
 
       # Check if int variable are managed by the worklist
       user_id = 'manager'
       self.login(user_id)
       result = workflow_tool.listActions(object=document)
       self.logMessage("Check %s worklist with int value as %s" % \
-                                     (user_id, self.int_value))
-      self.checkWorklist(result, self.actbox_int_variable_name, 1)
+                                     (user_id, 1))
+      self.checkWorklist(result, 'int_value_todo', 1)
 
       # Change int value on document
-      new_value = self.int_value + 1
-      document.setProperty(self.int_catalogued_variable_id, new_value)
+      document.setProperty('int_index', 2)
       self.tic()
       self.clearCache()
 
       result = workflow_tool.listActions(object=document)
       self.logMessage("Check %s worklist with int value as %s" % \
-                                     (user_id, new_value))
-      self.checkWorklist(result, self.actbox_int_variable_name, 0)
+                                     (user_id, 2))
+      self.checkWorklist(result, 'int_value_todo', 0)
 
       #
       # Check monovalued security role
       #
       self.login('manager')
-      module = self.getPortal().getDefaultModule(self.checked_portal_type)
+      module = self.getPortal().getDefaultModule('Organisation')
       module.manage_setLocalRoles('bar', ['Author'])
 
       self.login('bar')
@@ -343,7 +304,7 @@ class TestWorklist(testWorkflowMixin):
 
           result = workflow_tool.listActions(object=document)
           self.logMessage("Check %s worklist as Owner (%s)" % (user_id, count))
-          self.checkWorklist(result, self.actbox_owner_name, count)
+          self.checkWorklist(result, 'owner_todo', count)
 
       test(0, 0, 1)
 
@@ -362,7 +323,8 @@ class TestWorklist(testWorkflowMixin):
             current_sql_catalog_local_role_keys
         self.commit()
     finally:
-      self.removeWorklists()
+      self.removeWorklist('validation_workflow', self.test_worklist_dict.keys())
+      self.commit()
 
   def test_02_related_key(self, quiet=0, run=run_all_test):
     """
@@ -379,13 +341,13 @@ class TestWorklist(testWorkflowMixin):
 
     self.logMessage("Create worklists using 'base_category_id' related key")
 
-    self.addWorkflowCataloguedVariable(self.checked_workflow,
+    self.addWorkflowCataloguedVariable('validation_workflow',
                                        'base_category_id')
-    self.createWorklist(self.checked_workflow, 'region_worklist', 'has_region',
-                        portal_type=self.checked_portal_type,
+    self.createWorklist('validation_workflow', 'region_worklist', 'has_region',
+                        portal_type='Organisation',
                         base_category_id='region')
-    self.createWorklist(self.checked_workflow, 'role_worklist', 'has_role',
-                        portal_type=self.checked_portal_type,
+    self.createWorklist('validation_workflow', 'role_worklist', 'has_role',
+                        portal_type='Organisation',
                         base_category_id='role')
 
     try:
@@ -410,8 +372,10 @@ class TestWorklist(testWorkflowMixin):
       self.checkWorklist(result, 'has_region', 2)
       self.checkWorklist(result, 'has_role', 1)
     finally:
-      self.removeWorklist(self.checked_workflow,
+      self.removeWorklist('validation_workflow',
                           ['region_worklist', 'role_worklist'])
+      self.commit()
+
 
   def test_03_worklist_guard(self, quiet=0, run=run_all_test):
     """
@@ -424,11 +388,11 @@ class TestWorklist(testWorkflowMixin):
     self.createManagerAndLogin()
 
     self.logMessage("Create worklists with guard expression")
-    self.createWorklist(self.checked_workflow, 'guard_expression_worklist',
+    self.createWorklist('validation_workflow', 'guard_expression_worklist',
                         'valid_guard_expression',
-                        portal_type=self.checked_portal_type,
+                        portal_type='Organisation',
                         validation_state='validated',
-                        guard_roles="Associate",
+                        guard_roles='Associate',
                         guard_expr='python: user.getId() == "bar"')
 
     try:
@@ -454,8 +418,10 @@ class TestWorklist(testWorkflowMixin):
       result = workflow_tool.listActions(object=document)
       self.checkWorklist(result, 'valid_guard_expression', 0)
     finally:
-      self.removeWorklist(self.checked_workflow,
+      self.removeWorklist('validation_workflow',
                           ['guard_expression_worklist'])
+      self.commit()
+
 
   def test_04_dynamic_variables(self):
     """
@@ -470,25 +436,24 @@ class TestWorklist(testWorkflowMixin):
 
     self.logMessage("Create worklists using 'region_uid' related key"\
                     " and TALES Expression")
-    self.addWorkflowCataloguedVariable(self.checked_workflow,
+    self.addWorkflowCataloguedVariable('validation_workflow',
                                        'region_uid')
-    self.createWorklist(self.checked_workflow, 'region_worklist',
+    self.createWorklist('validation_workflow', 'region_worklist',
                         'has_semewhere_region',
-                        portal_type=self.checked_portal_type,
+                        portal_type='Organisation',
                         actbox_url='organisation_module?'\
                         'region_uid:list=%(region_uid)s&'\
                         'portal_type:list=%(portal_type)s&reset:int=1',
-                        region_uid='python:object.getPortalObject().'\
-                        'portal_categories.getCategoryUid("somewhere",'\
-                        ' base_category="region")')
+                        region_uid='python:[str(object.getPortalObject().'\
+                        'portal_categories.region.somewhere.getUid())]')
 
     try:
       document = self.createDocument()
       self.tic()
       self.clearCache()
       self.logMessage("  Check no document has region categories defined")
-      list_action = workflow_tool.listActions(object=document)
-      self.checkWorklist(list_action, 'has_semewhere_region', 0)
+      action_list = workflow_tool.listActions(object=document)
+      self.checkWorklist(action_list, 'has_semewhere_region', 0)
 
       self.logMessage("  Creates documents with region categories defined")
 
@@ -501,13 +466,14 @@ class TestWorklist(testWorkflowMixin):
       self.logMessage("  Check there are documents with region categories defined")
       action_list = workflow_tool.listActions(object=document)
       url_parameter_dict = {'region_uid': [str(self.portal.portal_categories.\
-                                          getCategoryUid("region/somewhere"))],
-                            'portal_type': [self.checked_portal_type]}
+                                          region.somewhere.getUid())],
+                            'portal_type': ['Organisation']}
       self.checkWorklist(action_list, 'has_semewhere_region', 2,
-                         url_parameter_dict=url_parameter_dict)
+                         url_parameter_dict=url_parameter_dict, selection_name='organisation_module_selection')
 
     finally:
-      self.removeWorklist(self.checked_workflow, ['region_worklist'])
+      self.removeWorklist('validation_workflow', ['region_worklist'])
+      self.commit()
 
 
 def test_suite():

product/ERP5Workflow/Document/Workflow.py

@@ -343,14 +343,12 @@ class Workflow(IdAsReferenceMixin("", "prefix"), XMLObject):
 
     portal = self.getPortalObject()
     def getPortalTypeListForWorkflow(workflow_id):
-        workflow_tool = portal.portal_workflow
-        result = []
-        append = result.append
+        portal_type_list = []
         for type_info in portal.portal_types.objectValues():
           portal_type = type_info.id
           if workflow_id in type_info.getTypeWorkflowList():
-            append(portal_type)
-        return result
+            portal_type_list.append(portal_type)
+        return portal_type_list
 
     _getPortalTypeListForWorkflow = CachingMethod(getPortalTypeListForWorkflow,
                               id='_getPortalTypeListForWorkflow', cache_factory = 'erp5_ui_long')

product/ERP5Workflow/Document/Worklist.py

@@ -200,7 +200,7 @@ class Worklist(IdAsReferenceMixin("worklist_", "prefix"), XMLObject):
           matches_id = self.getMatchedCausalityState()
           matches_ref_list.append(matches_id)
           matches = tuple(matches_ref_list)
-        else:
+        elif id:
           # Local dynamic variable:
           dynamic_varible = self._getOb('variable_'+id)
           if dynamic_varible.getInitialValue():