Commit aed3f283 authored by Arnaud Fontaine's avatar Arnaud Fontaine

Accessor: Generate Portal Type accessors for transitions date of Workflow History.

Such as:
  - getStartTransitionDateList() and getStartTransitionDate().
  - getDeliverTransitionDateList() and getDeliveryTransitionDate().
  - ...

Where the non-List accessors returned the last element.

Use Case: Set AT.{start,stop}_date from PPL stop transition date.
parent f523edc4
##############################################################################
#
# Copyright (c) 2018 Nexedi SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from Products.ERP5Type.PsycoWrapper import psyco
from Base import Getter as BaseGetter
from Acquisition import aq_base
class ListGetter(BaseGetter):
"""
Get all 'key' properties from Workflow History Transitions. A
default value can be provided if needed.
"""
def __init__(self, id, workflow_id, transition_id, key):
self._id = id
self.__name__ = id
self._workflow_id = workflow_id
self._transition_id = transition_id
self._key = key
def __call__(self, instance, *args):
try:
default = args[0]
except IndexError:
default = []
instance = aq_base(instance)
try:
workflow_history_dict_list = instance.workflow_history[self._workflow_id]
except (AttributeError, KeyError):
return default
else:
result_list = []
for workflow_history_dict in workflow_history_dict_list:
if workflow_history_dict['action'] == self._transition_id:
result_list.append(workflow_history_dict.get(self._key, default))
return result_list if result_list else default
psyco.bind(__call__)
class Getter(BaseGetter):
"""
Get 'key' property from Workflow History latest Transition. A
default value can be provided if needed.
"""
def __init__(self, id, list_accessor_name):
self._id = id
self.__name__ = id
self._list_accessor_name = list_accessor_name
def __call__(self, instance, default=None):
value_list = getattr(instance, self._list_accessor_name)()
try:
return value_list[-1]
except IndexError:
return default
psyco.bind(__call__)
...@@ -482,6 +482,7 @@ def getClassPropertyList(klass): ...@@ -482,6 +482,7 @@ def getClassPropertyList(klass):
if p not in ps_list]) if p not in ps_list])
return ps_list return ps_list
from Products.ERP5Type.Accessor import WorkflowHistory as WorkflowHistoryAccessor
def initializePortalTypeDynamicWorkflowMethods(ptype_klass, portal_workflow): def initializePortalTypeDynamicWorkflowMethods(ptype_klass, portal_workflow):
"""We should now make sure workflow methods are defined """We should now make sure workflow methods are defined
and also make sure simulation state is defined.""" and also make sure simulation state is defined."""
...@@ -519,6 +520,21 @@ def initializePortalTypeDynamicWorkflowMethods(ptype_klass, portal_workflow): ...@@ -519,6 +520,21 @@ def initializePortalTypeDynamicWorkflowMethods(ptype_klass, portal_workflow):
storage = dc_workflow_dict storage = dc_workflow_dict
transitions = wf.transitions transitions = wf.transitions
for transition in transitions.objectValues():
transition_id = transition.getId()
list_method_id = 'get%sTransitionDateList' % UpperCase(transition_id)
if not hasattr(ptype_klass, list_method_id):
method = WorkflowHistoryAccessor.ListGetter(list_method_id, wf_id, transition_id, 'time')
ptype_klass.registerAccessor(method,
Permissions.AccessContentsInformation)
method_id = 'get%sTransitionDate' % UpperCase(transition_id)
if not hasattr(ptype_klass, method_id):
method = WorkflowHistoryAccessor.Getter(method_id, list_method_id)
ptype_klass.registerAccessor(method,
Permissions.AccessContentsInformation)
elif wf_type == "InteractionWorkflowDefinition": elif wf_type == "InteractionWorkflowDefinition":
storage = interaction_workflow_dict storage = interaction_workflow_dict
transitions = wf.interactions transitions = wf.interactions
......
...@@ -323,6 +323,57 @@ class TestPortalTypeClass(ERP5TypeTestCase): ...@@ -323,6 +323,57 @@ class TestPortalTypeClass(ERP5TypeTestCase):
self.assertTrue(not erp5.portal_type.Person.__isghost__) self.assertTrue(not erp5.portal_type.Person.__isghost__)
self.assertTrue('_categories' in erp5.portal_type.Person.__dict__) self.assertTrue('_categories' in erp5.portal_type.Person.__dict__)
def testWorkflowHistoryAccessor(self):
person = self.portal.person_module.newContent(portal_type='Person')
from Products.ERP5Type.Utils import UpperCase
for transition_id in ('delete',
'delete_action',
'invalidate',
'invalidate_action',
'validate',
'validate_action'):
method_id = 'get%sTransitionDate' % UpperCase(transition_id)
self.assertNotEqual(getattr(person, method_id, None), None)
self.assertEqual(getattr(person, method_id)('hoge'), 'hoge')
list_method_id = 'get%sTransitionDateList' % UpperCase(transition_id)
self.assertNotEqual(getattr(person, list_method_id, None), None)
self.assertEqual(getattr(person, list_method_id)('hoge'), 'hoge')
person.validate()
self.assertEqual(len(person.getInvalidateTransitionDateList()), 0)
validate_date_list = person.getValidateTransitionDateList()
self.assertEqual(len(validate_date_list), 1)
self.assertEqual(
sorted([ workflow_history_dict['time']
for workflow_history_dict in person.workflow_history['validation_workflow']
if workflow_history_dict['action'] == 'validate' ]),
validate_date_list)
self.assertEqual(person.getValidateTransitionDate(), validate_date_list[0])
person.invalidate()
invalidate_date_list = person.getInvalidateTransitionDateList()
self.assertEqual(len(invalidate_date_list), 1)
self.assertEqual(
sorted([ workflow_history_dict['time']
for workflow_history_dict in person.workflow_history['validation_workflow']
if workflow_history_dict['action'] == 'invalidate' ]),
invalidate_date_list)
self.assertEqual(person.getInvalidateTransitionDate(), invalidate_date_list[0])
self.assertEqual(person.getValidateTransitionDateList(), validate_date_list)
person.validate()
self.assertEqual(person.getInvalidateTransitionDateList(), invalidate_date_list)
validate_date_list = person.getValidateTransitionDateList()
self.assertEqual(len(validate_date_list), 2)
self.assertEqual(
sorted([ workflow_history_dict['time']
for workflow_history_dict in person.workflow_history['validation_workflow']
if workflow_history_dict['action'] == 'validate' ]),
validate_date_list)
self.assertEqual(person.getValidateTransitionDate(), validate_date_list[-1])
class TestZodbPropertySheet(ERP5TypeTestCase): class TestZodbPropertySheet(ERP5TypeTestCase):
""" """
XXX: WORK IN PROGRESS XXX: WORK IN PROGRESS
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment