Commit 81784db4 authored by Rafael Monnerat's avatar Rafael Monnerat

Initial Import of ERP5Workflow porduct.

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@39260 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent 6ba051e8
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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 AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
from Products.ERP5Type.XMLObject import XMLObject
from zLOG import LOG, ERROR, DEBUG, WARNING
class StateError(Exception):
"""
Must call only an available transition
"""
pass
class State(XMLObject):
"""
A ERP5 State.
"""
meta_type = 'ERP5 State'
portal_type = 'State'
add_permission = Permissions.AddPortalContent
isPortalContent = 1
isRADContent = 1
# Declarative security
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.AccessContentsInformation)
# Declarative properties
property_sheets = (
PropertySheet.Base,
PropertySheet.XMLObject,
PropertySheet.CategoryCore,
PropertySheet.DublinCore,
PropertySheet.State,)
def getAvailableTransitionList(self, document):
"""
Return available transitions only if they are accessible for document.
"""
transition_list = self.getDestinationValueList(portal_type = 'Transition')
result_list = []
for transition in transition_list:
value = transition._checkPermission(document)
if value:
result_list.append(transition)
return result_list
def executeTransition(self, transition, document, form_kw=None):
"""
Execute transition on the object.
"""
if transition not in self.getAvailableTransitionList(document):
raise StateError
else:
transition.execute(document, form_kw=form_kw)
def undoTransition(self, document):
"""
Reverse previous transition
"""
wh = self.getWorkflowHistory(document, remove_undo=1)
status_dict = wh[-2]
# Update workflow state
state_bc_id = self.getParentValue().getStateBaseCategory()
document.setCategoryMembership(state_bc_id, status_dict[state_bc_id])
# Update workflow history
status_dict['undo'] = 1
self.getParentValue()._updateWorkflowHistory(document, status_dict)
# XXX
LOG("State, undo", ERROR, "Variable (like DateTime) need to be updated!")
def getWorkflowHistory(self, document, remove_undo=0, remove_not_displayed=0):
"""
Return history tuple
"""
wh = document.workflow_history[self.getParentValue()._generateHistoryKey()]
result = []
# Remove undo
if not remove_undo:
result = [x.copy() for x in wh]
else:
result = []
for x in wh:
if x.has_key('undo') and x['undo'] == 1:
result.pop()
else:
result.append(x.copy())
return result
def getVariableValue(self, document, variable_name):
"""
Get current value of the variable from the object
"""
status_dict = self.getParentValue().getCurrentStatusDict(document)
return status_dict[variable_name]
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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 AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
from Products.ERP5Type.XMLObject import XMLObject
from zLOG import LOG, ERROR, DEBUG, WARNING
from Products.PageTemplates.Expressions import getEngine
from Products.ERP5Type.Accessor.Base import _evaluateTales
class Transition(XMLObject):
"""
A ERP5 Transition.
"""
meta_type = 'ERP5 Transition'
portal_type = 'Transition'
add_permission = Permissions.AddPortalContent
isPortalContent = 1
isRADContent = 1
# Declarative security
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.AccessContentsInformation)
# Declarative properties
property_sheets = (
PropertySheet.Base,
PropertySheet.XMLObject,
PropertySheet.CategoryCore,
PropertySheet.DublinCore,
PropertySheet.Transition,
)
def execute(self, document, form_kw=None):
"""
Execute transition.
"""
# Call the before script
#self._executeBeforeScript(document)
# Modify the state
self._changeState(document)
# Get variable values
status_dict = self.getParentValue().getCurrentStatusDict(document)
status_dict['undo'] = 0
# Modify workflow history
state_bc_id = self.getParentValue().getStateBaseCategory()
status_dict[state_bc_id] = document.getCategoryMembershipList(state_bc_id)[0]
state_object = document.unrestrictedTraverse(status_dict[state_bc_id])
object = self.getParentValue().getStateChangeInformation(document, state_object, transition=self)
# Update all variables
variable_list = self.getParentValue().contentValues(portal_type='Variable')
for variable in variable_list:
if variable.getAutomaticUpdate():
# if we have it in form get it from there
# otherwise use default
variable_title = variable.getTitle()
if form_kw.has_key(variable_title):
status_dict[variable_title] = form_kw[variable_title]
else:
status_dict[variable_title] = variable.getInitialValue(object=object)
# Update all transition variables
if form_kw is not None:
object.REQUEST.other.update(form_kw)
variable_list = self.contentValues(portal_type='Transition Variable')
for variable in variable_list:
status_dict[variable.getCausalityTitle()] = variable.getInitialValue(object=object)
self.getParentValue()._updateWorkflowHistory(document, status_dict)
# Call the after script
self._executeAfterScript(document, form_kw=form_kw)
def _changeState(self, document):
"""
Change the state of the object.
"""
state = self.getDestination()
if state is not None:
# Some transitions don't update the state
state_bc_id = self.getParentValue().getStateBaseCategory()
document.setCategoryMembership(state_bc_id, state)
def _executeAfterScript(self, document, form_kw=None):
"""
Execute post transition script.
"""
if form_kw is None:
form_kw = {}
script_id = self.getAfterScriptId()
if script_id is not None:
script = getattr(document, script_id)
script(**form_kw)
def _checkPermission(self, document):
"""
Check if transition is allowed.
"""
expr_value = self.getGuardExpression(evaluate=0)
if expr_value is not None:
# do not use 'getGuardExpression' to calculate tales because
# it caches value which is bad. Instead do it manually
value = _evaluateTales(document, expr_value)
else:
value = True
#print "CALC", expr_value, '-->', value
return value
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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 AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
from Products.ERP5Type.XMLObject import XMLObject
class Variable(XMLObject):
"""
A ERP5 Variable.
"""
meta_type = 'ERP5 Variable'
portal_type = 'Variable'
add_permission = Permissions.AddPortalContent
isPortalContent = 1
isRADContent = 1
# Declarative security
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.AccessContentsInformation)
# Declarative properties
property_sheets = (
PropertySheet.Base,
PropertySheet.XMLObject,
PropertySheet.CategoryCore,
PropertySheet.DublinCore,
PropertySheet.Variable,
)
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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 AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
from Products.ERP5Type.XMLObject import XMLObject
from zLOG import LOG, ERROR, DEBUG, WARNING
from tempfile import mktemp
import os
import sys
from os.path import basename, splitext, join
from Products.DCWorkflowGraph.config import bin_search_path, DOT_EXE
from Products.DCWorkflowGraph.DCWorkflowGraph import bin_search
from Globals import PersistentMapping
from Acquisition import aq_base
from DateTime import DateTime
class Workflow(XMLObject):
"""
A ERP5 Workflow.
"""
meta_type = 'ERP5 Workflow'
portal_type = 'Workflow'
add_permission = Permissions.AddPortalContent
isPortalContent = 1
isRADContent = 1
# Declarative security
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.AccessContentsInformation)
# Declarative properties
property_sheets = (
PropertySheet.Base,
PropertySheet.XMLObject,
PropertySheet.CategoryCore,
PropertySheet.DublinCore,
PropertySheet.Workflow,
)
def initializeDocument(self, document):
"""
Set initial state on the Document
"""
state_bc_id = self.getStateBaseCategory()
document.setCategoryMembership(state_bc_id, self.getSource())
object = self.getStateChangeInformation(document, self.getSourceValue())
# Initialize workflow history
status_dict = {state_bc_id: self.getSource()}
variable_list = self.contentValues(portal_type='Variable')
for variable in variable_list:
status_dict[variable.getTitle()] = variable.getInitialValue(object=object)
self._updateWorkflowHistory(document, status_dict)
def _generateHistoryKey(self):
"""
Generate a key used in the workflow history.
"""
return self.getRelativeUrl()
def _updateWorkflowHistory(self, document, status_dict):
"""
Change the state of the object.
"""
# Create history attributes if needed
if getattr(aq_base(document), 'workflow_history', None) is None:
document.workflow_history = PersistentMapping()
# XXX this _p_changed is apparently not necessary
document._p_changed = 1
# Add an entry for the workflow in the history
workflow_key = self._generateHistoryKey()
if not document.workflow_history.has_key(workflow_key):
document.workflow_history[workflow_key] = ()
# Update history
document.workflow_history[workflow_key] += (status_dict, )
# XXX this _p_changed marks the document modified, but the
# only the PersistentMapping is modified
document._p_changed = 1
# XXX this _p_changed is apparently not necessary
document.workflow_history._p_changed = 1
def getCurrentStatusDict(self, document):
"""
Get the current status dict.
"""
workflow_key = self._generateHistoryKey()
# Copy is requested
result = document.workflow_history[workflow_key][-1].copy()
return result
def getDateTime(self):
"""
Return current date time.
"""
return DateTime()
def getStateChangeInformation(self, document, state, transition=None):
"""
Return an object used for variable tales expression.
"""
if transition is None:
transition_url = None
else:
transition_url = transition.getRelativeUrl()
return self.asContext(document=document,
transition=transition,
transition_url=transition_url,
state=state)
###########
## Graph ##
############
def getGraph(self, format="gif", REQUEST=None, *args, **kw):
"""
show a workflow as a graph, copy from:
"OpenFlowEditor":http://www.openflow.it/wwwopenflow/Download/OpenFlowEditor_0_4.tgz
"""
pot = self.getPOT()
infile = mktemp('.dot')
f = open(infile, 'w')
f.write(pot)
f.close()
outfile = mktemp('.%s' % format)
os.system('%s -T%s -o %s %s' % (bin_search(DOT_EXE), format, outfile, infile))
out = open(outfile, 'rb')
result = out.read()
out.close()
os.remove(infile)
os.remove(outfile)
return result
def getPOT(self):
"""
get the pot, copy from:
"dcworkfow2dot.py":http://awkly.org/Members/sidnei/weblog_storage/blog_27014
and Sidnei da Silva owns the copyright of the this function
"""
out = []
transition_dict = {}
out.append('digraph "%s" {' % self.getTitle())
transition_with_init_state_list = []
for state in self.contentValues(portal_type='State'):
out.append('%s [shape=box,label="%s",' \
'style="filled",fillcolor="#ffcc99"];' % \
(state.getId(), state.getTitle()))
# XXX Use API instead of getDestinationValueList
for available_transition in state.getDestinationValueList():
transition_with_init_state_list.append(available_transition.getId())
destination_state = available_transition.getDestinationValue()
if destination_state is None:
# take care of 'remain in state' transitions
destination_state = state
#
key = (state.getId(), destination_state.getId())
value = transition_dict.get(key, [])
value.append(available_transition.getTitle())
transition_dict[key] = value
# iterate also on transitions, and add transitions with no initial state
for transition in self.contentValues(portal_type='Transition'):
trans_id = transition.getId()
if trans_id not in transition_with_init_state_list:
destination_state = transition.getDestinationValue()
if destination_state is None:
dest_state_id = None
else:
dest_state_id = destination_state.getId()
key = (None, dest_state_id)
value = transition_dict.get(key, [])
value.append(transition.getTitle())
transition_dict[key] = value
for k, v in transition_dict.items():
out.append('%s -> %s [label="%s"];' % (k[0], k[1],
',\\n'.join(v)))
out.append('}')
return '\n'.join(out)
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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 AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
from Products.ERP5Type.XMLObject import XMLObject
class Worklist(XMLObject):
"""
A ERP5 Worklist.
"""
meta_type = 'ERP5 Worklist'
portal_type = 'Worklist'
add_permission = Permissions.AddPortalContent
isPortalContent = 1
isRADContent = 1
# Declarative security
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.AccessContentsInformation)
# Declarative properties
property_sheets = (
PropertySheet.Base,
PropertySheet.XMLObject,
PropertySheet.CategoryCore,
PropertySheet.DublinCore,
)
This diff is collapsed.
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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.
#
##############################################################################
class State:
"""
State properties and categories
"""
_properties = (
{
'id': 'is_initial_state',
'description': 'Define the initial state of the workflow',
'type': 'boolean',
'mode': 'w',
'default': 0
},
)
# XXX Can not use because getDestinationTitleList acquire on Node
_categories = ('destination',)
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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.
#
##############################################################################
class Transition:
"""
Transition properties and categories
"""
_properties = (
{ 'id' : 'transition_form_id',
'description' : 'Defines the form use to display question to the user.',
'type' : 'string',
'mode' : 'w' },
{ 'id' : 'before_script_id',
'description' : 'Defines the script called before a transition.',
'type' : 'string',
'mode' : 'w' },
{ 'id' : 'after_script_id',
'description' : 'Defines the script called after a transition.',
'type' : 'string',
'mode' : 'w' },
{ 'id' : 'guard_expression',
'description' : 'Tales expression use to disable transition',
'type' : 'tales',
'mode' : 'w' },
)
# XXX Can not use because getDestinationTitleList acquires on Node
_categories = ('destination',)
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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.
#
##############################################################################
class Variable:
"""
Variable properties and categories
"""
_properties = (
{ 'id' : 'initial_value',
'description' : 'Defines the default value.',
'type' : 'tales',
'mode' : 'w' },
{ 'id' : 'automatic_update',
'description' : 'Do we update the value in each transition ?',
'type' : 'boolean',
'mode' : 'w' },
)
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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.
#
##############################################################################
class Workflow:
"""
Workflow properties and categories
"""
_properties = (
{
'id' : 'state_base_category',
'description' : 'Defines the base category used to save the current' \
'state',
'type' : 'string',
'mode' : 'w'
},
)
# XXX Can not use because getDestinationTitleList acquire on Node
_categories = ('source',)
##############################################################################
#
# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# 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.
#
##############################################################################
"""
ERP5Workflow is a product containing Document to create
workflow in the ERP5 way.
"""
# Update ERP5 Globals
from Products.ERP5Type.Utils import initializeProduct, updateGlobals
import sys, Permissions
this_module = sys.modules[ __name__ ]
document_classes = updateGlobals(this_module, globals(),
permissions_module=Permissions)
# Define object classes and tools
object_classes = ()
portal_tools = ()
content_classes = ()
content_constructors = ()
# Finish installation
def initialize(context):
import Document
from zLOG import LOG, INFO
initializeProduct(context, this_module, globals(),
document_module=Document,
document_classes=document_classes,
object_classes=object_classes,
portal_tools=portal_tools,
content_constructors=content_constructors,
content_classes=content_classes)
##############################################################################
#
# Copyright (c) 2007 Nexedi SA and Contributors. All Rights Reserved.
# Jerome Perrin <jerome@nexedi.com>
#
# 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.
#
##############################################################################
import unittest
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from AccessControl.SecurityManagement import newSecurityManager
from AccessControl import Unauthorized
from AccessControl import SpecialUsers
try:
from transaction import get as get_transaction
except ImportError:
pass
class TestERP5Workflow(ERP5TypeTestCase):
"""
Tests ERP5 Workflow.
"""
def getBusinessTemplateList(self):
"""Returns list of BT to be installed."""
return ('erp5_workflow',)
def changeToAnonymous(self):
"""
Change the current user to Anonymous
"""
newSecurityManager(None, SpecialUsers.nobody)
def afterSetUp(self):
self.portal = self.getPortal()
self.workflow_module = self.portal.workflow_module
self.login() # as Manager
def test_SimpleWorkflow(self):
workflow = self.workflow_module.newContent(
portal_type='Workflow')
s1 = workflow.newContent(portal_type='State',
title='State 1')
s2 = workflow.newContent(portal_type='State',
title='State 2')
t1 = workflow.newContent(portal_type='Transition',
title='Transition 1')
s1.setDestinationValue(t1)
t1.setDestinationValue(s2)
# set initial state
workflow.setSourceValue(s1)
# state variable
workflow.setStateBaseCategory('current_state')
# create a document and associate it to this workflow
doc = self.portal.newContent(portal_type='Folder', id='test_doc')
workflow.initializeDocument(doc)
self.assertEquals(s1.getRelativeUrl(),
doc._getDefaultAcquiredCategoryMembership('current_state'))
# pass a transition
t1.execute(doc)
self.assertEquals(s2.getRelativeUrl(),
doc._getDefaultAcquiredCategoryMembership('current_state'))
def test_getAvailableTransitionList(self):
workflow = self.workflow_module.newContent(
portal_type='Workflow',
state_base_category='current_state')
s1 = workflow.newContent(portal_type='State',
title='State 1')
workflow.setSourceValue(s1)
t1 = workflow.newContent(portal_type='Transition',
title='Transition 1')
t2 = workflow.newContent(portal_type='Transition',
title='Transition 2',
guard_expression='python: False')
s1.setDestinationValueList([t1, t2])
doc = self.portal.newContent(portal_type='Folder', id='test_doc')
workflow.initializeDocument(doc)
self.assertEquals([t1], s1.getAvailableTransitionList(doc))
def test_WorkflowVariables(self):
workflow = self.workflow_module.newContent(
portal_type='Workflow',
state_base_category='current_state')
s1 = workflow.newContent(portal_type='State',
title='State 1')
workflow.setSourceValue(s1)
t1 = workflow.newContent(portal_type='Transition',
title='Transition 1',
destination_value=s1)
s1.setDestinationValue(t1)
v1 = workflow.newContent(portal_type='Variable',
title='actor',
initial_value='member/getUserName')
doc = self.portal.newContent(portal_type='Folder', id='test_doc')
workflow.initializeDocument(doc)
t1.execute(doc)
current_state = workflow.getCurrentStatusDict(doc)
self.failUnless(isinstance(current_state, dict))
self.assertEquals(s1.getRelativeUrl(), current_state.get('current_state'))
self.assertEquals('ERP5TypeTestCase', current_state.get('actor'))
self.assertEquals(0, current_state.get('undo'))
# XXX workflow history is a method on State ?
history = s1.getWorkflowHistory(doc)
self.assertEquals(len(history), 2)
def test_afterScript(self):
workflow = self.workflow_module.newContent(
portal_type='Workflow',
state_base_category='current_state')
s1 = workflow.newContent(portal_type='State',
title='State 1')
s2 = workflow.newContent(portal_type='State',
title='State 2')
t1 = workflow.newContent(portal_type='Transition',
title='Transition 1',
after_script_id='Document_testAfterScript'
)
s1.setDestinationValue(t1)
t1.setDestinationValue(s2)
workflow.setSourceValue(s1)
doc = self.portal.newContent(portal_type='Folder', id='test_doc')
called = []
def Document_testAfterScript(**kw):
called.append('called %s' % kw)
doc.Document_testAfterScript = Document_testAfterScript
workflow.initializeDocument(doc)
t1.execute(doc)
self.assertEquals(['called {}'], called)
# FIXME: not passing parameter to an after script is probably too
# restrictive
def test_WorkflowSecurity(self):
"""
Test workflow security.
"""
workflow_module = self.portal.workflow_module
def createWorkflowInstance():
return workflow_module.newContent(portal_type='Workflow')
workflow_instance = createWorkflowInstance()
# Anonymous User must not be able to access workflow module
# or workflow instances
self.changeToAnonymous()
self.assertRaises(Unauthorized, workflow_module.view)
self.assertRaises(Unauthorized, createWorkflowInstance)
self.assertRaises(Unauthorized, lambda: workflow_instance.view())
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(TestERP5Workflow))
return suite
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