--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/cubicweb/test/unittest_wfutils.py Mon May 11 13:57:34 2015 +0200
@@ -0,0 +1,115 @@
+# copyright 2017 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
+#
+# This file is part of CubicWeb.
+#
+# CubicWeb is free software: you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation, either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# CubicWeb 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 Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with CubicWeb. If not, see <http://www.gnu.org/licenses/>.
+
+import copy
+
+from cubicweb.devtools import testlib
+from cubicweb.wfutils import setup_workflow
+
+
+class TestWFUtils(testlib.CubicWebTC):
+
+ defs = {
+ 'group': {
+ 'etypes': 'CWGroup',
+ 'default': True,
+ 'initial_state': u'draft',
+ 'states': [u'draft', u'published'],
+ 'transitions': {
+ u'publish': {
+ 'fromstates': u'draft',
+ 'tostate': u'published',
+ 'requiredgroups': u'managers'
+ }
+ }
+ }
+ }
+
+ def test_create_workflow(self):
+ with self.admin_access.cnx() as cnx:
+ wf = setup_workflow(cnx, 'group', self.defs['group'])
+ self.assertEqual(wf.name, 'group')
+ self.assertEqual(wf.initial.name, u'draft')
+
+ draft = wf.state_by_name(u'draft')
+ self.assertIsNotNone(draft)
+
+ published = wf.state_by_name(u'published')
+ self.assertIsNotNone(published)
+
+ publish = wf.transition_by_name(u'publish')
+ self.assertIsNotNone(publish)
+
+ self.assertEqual(publish.destination_state, (published, ))
+ self.assertEqual(draft.allowed_transition, (publish, ))
+
+ self.assertEqual(
+ {g.name for g in publish.require_group},
+ {'managers'})
+
+ def test_update(self):
+ with self.admin_access.cnx() as cnx:
+ wf = setup_workflow(cnx, 'group', self.defs['group'])
+ eid = wf.eid
+
+ with self.admin_access.cnx() as cnx:
+ wfdef = copy.deepcopy(self.defs['group'])
+ wfdef['states'].append('new')
+ wfdef['initial_state'] = 'new'
+ wfdef['transitions'][u'publish']['fromstates'] = ('draft', 'new')
+ wfdef['transitions'][u'publish']['requiredgroups'] = (
+ u'managers', u'users')
+ wfdef['transitions'][u'todraft'] = {
+ 'fromstates': ('new', 'published'),
+ 'tostate': 'draft',
+ }
+
+ wf = setup_workflow(cnx, 'group', wfdef)
+ self.assertEqual(wf.eid, eid)
+ self.assertEqual(wf.name, 'group')
+ self.assertEqual(wf.initial.name, u'new')
+
+ new = wf.state_by_name(u'new')
+ self.assertIsNotNone(new)
+
+ draft = wf.state_by_name(u'draft')
+ self.assertIsNotNone(draft)
+
+ published = wf.state_by_name(u'published')
+ self.assertIsNotNone(published)
+
+ publish = wf.transition_by_name(u'publish')
+ self.assertIsNotNone(publish)
+
+ todraft = wf.transition_by_name(u'todraft')
+ self.assertIsNotNone(todraft)
+
+ self.assertEqual(
+ {g.name for g in publish.require_group},
+ {'managers', 'users'})
+
+ self.assertEqual(publish.destination_state, (published, ))
+ self.assertEqual(draft.allowed_transition, (publish, ))
+ self.assertEqual(todraft.destination_state, (draft, ))
+ self.assertEqual(published.allowed_transition, (todraft, ))
+ self.assertCountEqual(new.allowed_transition, (publish, todraft))
+
+
+if __name__ == '__main__':
+ import unittest
+ unittest.main()
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/cubicweb/wfutils.py Mon May 11 13:57:34 2015 +0200
@@ -0,0 +1,149 @@
+# copyright 2017 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
+#
+# This file is part of CubicWeb.
+#
+# CubicWeb is free software: you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation, either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# CubicWeb 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 Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with CubicWeb. If not, see <http://www.gnu.org/licenses/>.
+"""Workflow setup utilities.
+
+These functions work with a declarative workflow definition:
+
+.. code-block:: python
+
+ {
+ 'etypes': 'CWGroup',
+ 'default': True,
+ 'initial_state': u'draft',
+ 'states': [u'draft', u'published'],
+ 'transitions': {
+ u'publish': {
+ 'fromstates': u'draft',
+ 'tostate': u'published',
+ 'requiredgroups': u'managers'
+ 'conditions': (
+ 'U in_group X',
+ 'X owned_by U'
+ )
+ }
+ }
+ }
+
+.. autofunction:: setup_workflow
+.. autofunction:: cleanupworkflow
+"""
+
+import collections
+
+from six import text_type
+
+from cubicweb import NoResultError
+
+
+def get_tuple_or_list(value):
+ if value is None:
+ return None
+ if not isinstance(value, (tuple, list)):
+ value = (value,)
+ return value
+
+
+def cleanupworkflow(cnx, wf, wfdef):
+ """Cleanup an existing workflow by removing the states and transitions that
+ do not exist in the given definition.
+
+ :param cnx: A connexion with enough permissions to define a workflow
+ :param wf: A `Workflow` entity
+ :param wfdef: A workflow definition
+ """
+ cnx.execute(
+ 'DELETE State S WHERE S state_of WF, WF eid %%(wf)s, '
+ 'NOT S name IN (%s)' % (
+ ', '.join('"%s"' % s for s in wfdef['states'])),
+ {'wf': wf.eid})
+
+ cnx.execute(
+ 'DELETE Transition T WHERE T transition_of WF, WF eid %%(wf)s, '
+ 'NOT T name IN (%s)' % (
+ ', '.join('"%s"' % s for s in wfdef['transitions'])),
+ {'wf': wf.eid})
+
+
+def setup_workflow(cnx, name, wfdef, cleanup=True):
+ """Create or update a workflow definition so it matches the given
+ definition.
+
+ :param cnx: A connexion with enough permissions to define a workflow
+ :param name: The workflow name. Used to create the `Workflow` entity, or
+ to find an existing one.
+ :param wfdef: A workflow definition.
+ :param cleanup: Remove extra states and transitions. Can be done separatly
+ by calling :func:`cleanupworkflow`.
+ :return: The created/updated workflow entity
+ """
+ name = text_type(name)
+ try:
+ wf = cnx.find('Workflow', name=name).one()
+ except NoResultError:
+ wf = cnx.create_entity('Workflow', name=name)
+
+ etypes = get_tuple_or_list(wfdef['etypes'])
+ cnx.execute('DELETE WF workflow_of ETYPE WHERE WF eid %%(wf)s, '
+ 'NOT ETYPE name IN (%s)' % ','.join('"%s"' for e in etypes),
+ {'wf': wf.eid})
+ cnx.execute('SET WF workflow_of ETYPE WHERE'
+ ' NOT WF workflow_of ETYPE, WF eid %%(wf)s, ETYPE name IN (%s)'
+ % ','.join('"%s"' % e for e in etypes),
+ {'wf': wf.eid})
+ if wfdef['default']:
+ cnx.execute(
+ 'SET ETYPE default_workflow X '
+ 'WHERE '
+ 'NOT ETYPE default_workflow X, '
+ 'X eid %%(x)s, ETYPE name IN (%s)' % ','.join(
+ '"%s"' % e for e in etypes),
+ {'x': wf.eid})
+
+ states = {}
+ states_transitions = collections.defaultdict(list)
+ for state in wfdef['states']:
+ st = wf.state_by_name(state) or wf.add_state(state)
+ states[state] = st
+
+ if 'initial_state' in wfdef:
+ wf.cw_set(initial_state=states[wfdef['initial_state']])
+
+ for trname, trdef in wfdef['transitions'].items():
+ tr = (wf.transition_by_name(trname) or
+ cnx.create_entity('Transition', name=trname))
+ tr.cw_set(transition_of=wf)
+ if trdef.get('tostate'):
+ tr.cw_set(destination_state=states[trdef['tostate']])
+ fromstates = get_tuple_or_list(trdef.get('fromstates', ()))
+ for stname in fromstates:
+ states_transitions[stname].append(tr)
+
+ requiredgroups = get_tuple_or_list(trdef.get('requiredgroups', ()))
+ conditions = get_tuple_or_list(trdef.get('conditions', ()))
+
+ tr.set_permissions(requiredgroups, conditions, reset=True)
+
+ for stname, transitions in states_transitions.items():
+ state = states[stname]
+ state.cw_set(allowed_transition=None)
+ state.cw_set(allowed_transition=transitions)
+
+ if cleanup:
+ cleanupworkflow(cnx, wf, wfdef)
+
+ return wf
--- a/doc/book/devrepo/datamodel/define-workflows.rst Mon Feb 20 15:56:07 2017 +0100
+++ b/doc/book/devrepo/datamodel/define-workflows.rst Mon May 11 13:57:34 2015 +0200
@@ -158,3 +158,11 @@
re-create all the workflow entities. The user interface should only be
a reference for you to view the states and transitions, but is not the
appropriate interface to define your application workflow.
+
+
+Alternative way to declare workflows
+------------------------------------
+
+.. automodule:: cubicweb.wfutils
+
+
--- a/doc/changes/3.25.rst Mon Feb 20 15:56:07 2017 +0100
+++ b/doc/changes/3.25.rst Mon May 11 13:57:34 2015 +0200
@@ -8,6 +8,9 @@
allow to switch off internal connection pooling for use with others poolers
such as pgbouncer_.
-
.. _pgbouncer: https://pgbouncer.github.io/
+
+* A new way to declare workflows as simple data structure (dict/list) has been
+ introduced. Respective utility functions live in ``cubicweb.wfutils``
+ module. This handles both the creation and migration of workflows.
--- a/flake8-ok-files.txt Mon Feb 20 15:56:07 2017 +0100
+++ b/flake8-ok-files.txt Mon May 11 13:57:34 2015 +0200
@@ -80,6 +80,7 @@
cubicweb/test/unittest_rtags.py
cubicweb/test/unittest_schema.py
cubicweb/test/unittest_toolsutils.py
+cubicweb/test/unittest_wfutils.py
cubicweb/web/formwidgets.py
cubicweb/web/test/data/entities.py
cubicweb/web/test/unittest_http_headers.py
@@ -112,3 +113,4 @@
cubicweb/pyramid/test/test_rest_api.py
cubicweb/pyramid/test/test_tools.py
cubicweb/pyramid/pyramidctl.py
+cubicweb/wfutils.py