diff --git a/CHANGES.rst b/CHANGES.rst
index db0f34f..8f8367b 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -4,6 +4,7 @@ repoze.workflow Changelog
1.2 (unreleased)
----------------
+- support for role guards in transitions
- TBD
1.1 (2020-07-01)
diff --git a/docs/configuration.rst b/docs/configuration.rst
index eed245e..e4ede34 100644
--- a/docs/configuration.rst
+++ b/docs/configuration.rst
@@ -95,6 +95,14 @@ workflow.
permission="moderate"
/>
+
+
+
+
+
@@ -164,6 +172,14 @@ attributes:
the current user implied by the request has the permission in the
``context``, ``False`` otherwise.
+``roles_checker``
+
+ A Python dotted-name referring to a permission checking function.
+ This function should accept three arguments: ``roles`` (a list of
+ string), ``context`` and ``request``. It should return ``True`` if
+ the current user implied by the request has at least one of the roles
+ in the ``context``, ``False`` otherwise.
+
A ``workflow`` tag may contain ``transition`` and ``state`` tags. A
workflow declared via ZCML is unique amongst all workflows defined if
the combination of its ``type``, its ``content_types`` and its
@@ -323,6 +339,13 @@ The ``alias`` tag may only be used within a ``state`` tag. The
aliases*, it will be considered to be in that state, according to
e.g. ``workflow.state_of``, etc.
+The ``role`` Tag
+----------------
+The ``role`` tag which may occur within the ``transition`` tag allows
+to specify role guards. The only attribute available is ``name``.
+The roles specified within a ``transition`` are passed as a list of strings
+to the ``roles_checker`` which can be specified for the workflow.
+
.. _callbacks:
Callbacks
diff --git a/docs/usage.rst b/docs/usage.rst
index a7fa5c4..d3aed91 100644
--- a/docs/usage.rst
+++ b/docs/usage.rst
@@ -109,7 +109,7 @@ use some of these APIs you need a "request" object. This object is
available in :mod:`repoze.bfg` views as the "request" parameter to the
view. Your web framework may have another kind of request object
obtained from another place. If none of your workflows use a
-``permission_checker``, you can pass ``None`` as the request object.
+``permission_checker`` or ``roles_checker``, you can pass ``None`` as the request object.
Here is how you initialize a piece of content to the initial workflow
state:
@@ -180,7 +180,7 @@ current
transitions
A sequence of transition dictionaries; if any of the transitions is
- not allowed due to a permission violation, it will not show up in
+ not allowed due to a permission violation or insuficient roles, it will not show up in
this list.
You can also obtain state information about a nonexistent object
@@ -193,9 +193,9 @@ content object) using ``state_info``:
state_info = workflow.state_info(None, request, context=someotherobject)
This will return the same list of dictionaries, except the ``current``
-flag will always be false. Permissions used to compute the allowed
+flag will always be false. Permissions and roles used to compute the allowed
transitions will be computed against the ``context`` (the ``context``
-will be passed to the permission checker instead of any particular
+will be passed to the permission checker resp. the roles checker instead of any particular
content object).
You can obtain transition information for a piece of content using the
diff --git a/repoze/workflow/meta.zcml b/repoze/workflow/meta.zcml
index ef6f15d..9765792 100644
--- a/repoze/workflow/meta.zcml
+++ b/repoze/workflow/meta.zcml
@@ -33,6 +33,14 @@
handler="repoze.workflow.zcml.guard_function"
/>
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/repoze/workflow/tests/fixtures/dummy.py b/repoze/workflow/tests/fixtures/dummy.py
index 1c03644..1836a6d 100644
--- a/repoze/workflow/tests/fixtures/dummy.py
+++ b/repoze/workflow/tests/fixtures/dummy.py
@@ -16,6 +16,9 @@ class Content(object):
def callback(context, transition):
""" """
+def callback_after(context, transition):
+ """"""
+
def never(context, transition): # pragma: NO COVER
raise WorkflowError("This is never allowed")
@@ -23,3 +26,6 @@ def elector(context): return True
def has_permission(permission, context, request):
""" """
+
+def has_role(roles, context, request):
+ """"""
\ No newline at end of file
diff --git a/repoze/workflow/tests/test_workflow.py b/repoze/workflow/tests/test_workflow.py
index 0d19ea8..c70d080 100644
--- a/repoze/workflow/tests/test_workflow.py
+++ b/repoze/workflow/tests/test_workflow.py
@@ -8,9 +8,9 @@ def _getTargetClass(self):
return Workflow
def _makeOne(self, attr='state', initial_state='pending',
- permission_checker=None):
+ permission_checker=None, roles_checker=None):
klass = self._getTargetClass()
- return klass(attr, initial_state, permission_checker)
+ return klass(attr, initial_state, permission_checker, roles_checker=roles_checker)
def _makePopulated(self, state_callback=None, transition_callback=None):
sm = self._makeOne()
@@ -38,9 +38,10 @@ def _makePopulated(self, state_callback=None, transition_callback=None):
def _makePopulatedOverlappingTransitions(
self, state_callback=None, transition_callback=None,
- permission_checker=None):
+ permission_checker=None, roles_checker=None):
sm = self._makePopulated(state_callback, transition_callback)
sm.permission_checker = permission_checker
+ sm.roles_checker = roles_checker
sm._transition_data['submit2'] = dict(
name='submit2',
@@ -100,6 +101,80 @@ def checker(permission, context, request):
self.assertEqual([a[0] for a in sorted(checker_args)],
['forbidden1', 'forbidden2'])
+ def test_transition_to_state_two_transitions_none_works_with_role_check(self):
+ callback_args = []
+ def dummy(content, info): #pragma NO COVER
+ callback_args.append((content, info))
+
+ checker_args = []
+ def checker(permission, context, request):
+ checker_args.append((permission, context, request))
+ return permission == 'allowed'
+
+ roles_checker_args = []
+ def roles_checker(roles, context, request):
+ roles_checker_args.append((roles, context, request))
+ return roles == ['admin', 'manager']
+
+ sm = self._makePopulatedOverlappingTransitions(
+ transition_callback=dummy,
+ permission_checker=checker,
+ roles_checker=roles_checker
+ )
+
+ sm._transition_data['submit']['permission'] = 'allowed'
+ sm._transition_data['submit']['roles'] = ['viewer', 'editor']
+ sm._transition_data['submit2']['permission'] = 'allowed'
+ sm._transition_data['submit2']['roles'] = ['viewer']
+
+ ob = DummyContent()
+ ob.state = 'private'
+ from repoze.workflow import WorkflowError
+ self.assertRaises(WorkflowError, sm.transition_to_state,
+ ob, object(), 'pending')
+ self.assertEqual(len(callback_args), 0)
+ self.assertEqual(len(checker_args), 2)
+ self.assertEqual([a[0] for a in sorted(checker_args)],
+ ['allowed', 'allowed'])
+ self.assertEqual([a[0] for a in sorted(roles_checker_args)],
+ [['viewer'], ['viewer', 'editor']])
+
+ def test_transition_to_state_two_transitions_second_works_with_role_check(self):
+ callback_args = []
+ def dummy(content, info): #pragma NO COVER
+ callback_args.append((content, info))
+
+ checker_args = []
+ def checker(permission, context, request):
+ checker_args.append((permission, context, request))
+ return permission == 'allowed'
+
+ roles_checker_args = []
+ def roles_checker(roles, context, request):
+ roles_checker_args.append((roles, context, request))
+ return 'viewer' in roles
+
+ sm = self._makePopulatedOverlappingTransitions(
+ transition_callback=dummy,
+ permission_checker=checker,
+ roles_checker=roles_checker
+ )
+
+ sm._transition_data['submit']['permission'] = 'allowed'
+ sm._transition_data['submit']['roles'] = ['admin', 'editor']
+ sm._transition_data['submit2']['permission'] = 'allowed'
+ sm._transition_data['submit2']['roles'] = ['viewer']
+
+ ob = DummyContent()
+ ob.state = 'private'
+ sm.transition_to_state(ob, object(), 'pending')
+ self.assertEqual(len(callback_args), 1)
+ self.assertEqual(len(checker_args), 2)
+ self.assertEqual([a[0] for a in sorted(checker_args)],
+ ['allowed', 'allowed'])
+ self.assertEqual([a[0] for a in sorted(roles_checker_args)],
+ [['admin', 'editor'], ['viewer']])
+
def test_class_conforms_to_IWorkflow(self):
from zope.interface.verify import verifyClass
from repoze.workflow.interfaces import IWorkflow
@@ -258,6 +333,14 @@ def test_add_transition_with_permission_no_permission_checker(self):
self.assertRaises(WorkflowError, sm.add_transition, 'make_public',
'private', 'public', permission='permission')
+ def test_add_transition_with_roles_no_roles_checker(self):
+ from repoze.workflow import WorkflowError
+ sm = self._makeOne()
+ sm.add_state('private')
+ sm.add_state('public')
+ self.assertRaises(WorkflowError, sm.add_transition, 'make_public',
+ 'private', 'public', roles=['somerole'])
+
def test_check_fails(self):
from repoze.workflow import WorkflowError
sm = self._makeOne()
@@ -762,6 +845,63 @@ def append(content, name, context=None, request=None, guards=()):
self.assertRaises(WorkflowError, permitted, None, info)
self.assertEqual(args, [('view', None, request)])
+ def test_transition_permissive_with_roles(self):
+ args = []
+ def checker(*arg):
+ args.append(arg)
+ return True
+ workflow = self._makeOne(roles_checker=checker)
+ transitioned = []
+ def append(content, name, context=None, request=None, guards=()):
+ D = {'content':content, 'name': name, 'request': request,
+ 'guards':guards, 'context':context }
+ transitioned.append(D)
+ workflow._transition = lambda *arg, **kw: append(*arg, **kw)
+ content = DummyContent()
+ content.state = 'pending'
+ request = object()
+ workflow.transition(content, request, 'publish')
+ self.assertEqual(len(transitioned), 1)
+ transitioned = transitioned[0]
+ self.assertEqual(transitioned['content'], content)
+ self.assertEqual(transitioned['name'], 'publish')
+ self.assertEqual(transitioned['request'], request)
+ self.assertEqual(transitioned['context'], None)
+ permitted = transitioned['guards'][0]
+ info = DummyCallbackInfo(transition = {'roles': ['manager', 'viewer']})
+ result = permitted(None, info)
+ self.assertEqual(result, None)
+ self.assertEqual(args, [(['manager', 'viewer'], None, request)])
+
+ def test_transition_not_permissive_with_roles(self):
+ args = []
+ def checker(*arg):
+ args.append(arg)
+ return False
+ from repoze.workflow import WorkflowError
+ workflow = self._makeOne(roles_checker=checker)
+ transitioned = []
+ def append(content, name, context=None, request=None, guards=()):
+ D = {'content': content, 'name': name, 'request': request,
+ 'guards': guards, 'context': context }
+ transitioned.append(D)
+ workflow._transition = lambda *arg, **kw: append(*arg, **kw)
+ request = object()
+ content = DummyContent()
+ content.state = 'pending'
+ workflow.transition(content, request, 'publish')
+ self.assertEqual(len(transitioned), 1)
+ transitioned = transitioned[0]
+ self.assertEqual(transitioned['content'], content)
+ self.assertEqual(transitioned['name'], 'publish')
+ self.assertEqual(transitioned['request'], request)
+ self.assertEqual(transitioned['context'], None)
+ permitted = transitioned['guards'][0]
+ info = DummyCallbackInfo(transition = {'roles': ['manager', 'viewer']})
+ self.assertRaises(WorkflowError, permitted, None, info)
+ self.assertEqual(args, [(['manager', 'viewer'], None, request)])
+
+
def test_transition_request_is_None(self):
def checker(*arg): raise NotImplementedError
workflow = self._makeOne(permission_checker=checker)
@@ -971,6 +1111,60 @@ def checker(*arg):
self.assertEqual(args, [('view', request, 'whatever'),
('view', request, 'whatever')])
+ def test_get_transitions_permissive_checking_roles(self):
+ args = []
+ def roles_checker(*arg):
+ args.append(arg)
+ return True
+ workflow = self._makeOne(roles_checker=roles_checker)
+ workflow._get_transitions=lambda *arg, **kw: [{'roles':['viewer', 'editor']}, {}]
+ transitions = workflow.get_transitions(None, None, None, 'private')
+ self.assertEqual(len(transitions), 2)
+ self.assertEqual(args, [(['viewer', 'editor'], None, None)])
+
+ def test_get_transitions_nonpermissive_checking_roles(self):
+ args = []
+ def roles_checker(*arg):
+ args.append(arg)
+ return False
+ workflow = self._makeOne(roles_checker=roles_checker)
+ workflow._get_transitions=lambda *arg, **kw: [{'roles':['viewer', 'editor']}, {}]
+ transitions = workflow.get_transitions(None, 'private')
+ self.assertEqual(len(transitions), 1)
+ self.assertEqual(args, [(['viewer', 'editor'], None, 'private')])
+
+ def test_state_info_permissive_checking_roles(self):
+ args = []
+ def checker(*arg):
+ args.append(arg)
+ return True
+ state_info = []
+ state_info.append({'transitions':[{'roles':['viewer', 'editor']}, {}]})
+ state_info.append({'transitions':[{'roles':['viewer', 'editor']}, {}]})
+ workflow = self._makeOne(roles_checker=checker)
+ workflow._state_info = lambda *arg, **kw: state_info
+ request = object()
+ result = workflow.state_info(request, 'whatever')
+ self.assertEqual(result, state_info)
+ self.assertEqual(args, [(['viewer', 'editor'], request, 'whatever'),
+ (['viewer', 'editor'], request, 'whatever')])
+
+ def test_state_info_nonpermissive_checking_roles(self):
+ args = []
+ def checker(*arg):
+ args.append(arg)
+ return False
+ state_info = []
+ state_info.append({'transitions':[{'roles':['viewer', 'editor']}, {}]})
+ state_info.append({'transitions':[{'roles':['viewer', 'editor']}, {}]})
+ workflow = self._makeOne(roles_checker=checker)
+ workflow._state_info = lambda *arg, **kw: state_info
+ request = object()
+ result = workflow.state_info(request, 'whatever')
+ self.assertEqual(result, [{'transitions': [{}]}, {'transitions': [{}]}])
+ self.assertEqual(args, [(['viewer', 'editor'], request, 'whatever'),
+ (['viewer', 'editor'], request, 'whatever')])
+
def test_callbackinfo_has_request(self):
def transition_cb(content, info):
self.assertEqual(info.request, request)
diff --git a/repoze/workflow/tests/test_zcml.py b/repoze/workflow/tests/test_zcml.py
index 25e67c4..62822ee 100644
--- a/repoze/workflow/tests/test_zcml.py
+++ b/repoze/workflow/tests/test_zcml.py
@@ -1,6 +1,9 @@
import unittest
from zope.testing.cleanup import cleanUp
+from repoze.workflow.tests.fixtures.dummy import callback_after
+
+
class TestWorkflowDirective(unittest.TestCase):
def setUp(self):
cleanUp()
@@ -82,12 +85,12 @@ class IDummy2(Interface):
{'from_state': 'private', 'callback': None,
'guards': [], 'name': 'make_public',
'to_state': 'public', 'permission':None,
- 'title': 'make_public'},
+ 'title': 'make_public', 'roles': []},
'make_private':
{'from_state': 'private', 'callback': None,
'guards': [], 'name': 'make_private',
'to_state': 'public', 'permission':None,
- 'title': 'Retract'},
+ 'title': 'Retract', 'roles': []},
})
self.assertEqual(workflow.initial_state, 'public')
@@ -122,12 +125,12 @@ class IDummy2(Interface):
{'from_state': 'private', 'callback': None,
'guards': [], 'name': 'make_public',
'to_state': 'public', 'permission':None,
- 'title': 'make_public'},
+ 'title': 'make_public', 'roles': []},
'make_private':
{'from_state': 'private', 'callback': None,
'guards': [], 'name': 'make_private',
'to_state': 'public', 'permission':None,
- 'title': 'Retract'},
+ 'title': 'Retract', 'roles': []},
}
)
self.assertEqual(workflow.initial_state, 'public')
@@ -203,19 +206,21 @@ def _getTargetClass(self):
return TransitionDirective
def _makeOne(self, context=None, name=None, from_state=None,
- to_state=None, callback=None, permission=None):
+ to_state=None, callback=None, permission=None, callback_after=None):
return self._getTargetClass()(context, name, from_state,
- to_state, callback, permission)
+ to_state, callback,
+ permission, callback_after=callback_after)
def test_ctor(self):
directive = self._makeOne('context', 'name', 'from_state',
- 'to_state', 'callback', 'permission')
+ 'to_state', 'callback', 'permission', callback_after='callback_after')
self.assertEqual(directive.context, 'context')
self.assertEqual(directive.name, 'name')
self.assertEqual(directive.callback, 'callback')
self.assertEqual(directive.from_state, 'from_state')
self.assertEqual(directive.to_state, 'to_state')
self.assertEqual(directive.permission, 'permission')
+ self.assertEqual(directive.callback_after, 'callback_after')
self.assertEqual(directive.extras, {})
def test_after(self):
@@ -301,6 +306,7 @@ def test_execute_actions(self):
from repoze.workflow.tests.fixtures.dummy import IContent
from repoze.workflow.tests.fixtures.dummy import elector
from repoze.workflow.tests.fixtures.dummy import has_permission
+ from repoze.workflow.tests.fixtures.dummy import has_role
from repoze.workflow._compat import text_ as _u
xmlconfig.file('configure.zcml', package, execute=True)
sm = getSiteManager()
@@ -315,6 +321,7 @@ def test_execute_actions(self):
self.assertEqual(workflow.description, 'The workflow which is of the '
'testing fixtures package')
self.assertEqual(workflow.permission_checker, has_permission)
+ self.assertEqual(workflow.roles_checker, has_role)
self.assertEqual(
workflow._state_aliases,
{'supersecret':'private'},
@@ -333,34 +340,51 @@ def test_execute_actions(self):
},
})
transitions = workflow._transition_data
- self.assertEqual(len(transitions), 3)
+ self.assertEqual(len(transitions), 4)
self.assertEqual(transitions['private_to_public'],
{'from_state': _u('private'),
'callback': callback,
+ 'callback_after': callback_after,
'guards': [],
'name': _u('private_to_public'),
'to_state': _u('public'),
- 'permission':'moderate',
+ 'roles': [],
+ 'permission': 'moderate',
'title': 'private_to_public',
}),
self.assertEqual(transitions['unavailable_public_to_private'],
{'from_state': _u('public'),
'callback': callback,
+ 'callback_after': None,
'guards': [dummy.never],
'name': _u('unavailable_public_to_private'),
'to_state': _u('private'),
'permission':_u('moderate'),
+ 'roles': [],
'title': _u('unavailable_public_to_private'),
}),
self.assertEqual(transitions['public_to_private'],
{'from_state': _u('public'),
'callback': callback,
+ 'callback_after': None,
'guards': [],
'name': 'public_to_private',
'to_state': _u('private'),
'permission':'moderate',
+ 'roles': [],
'title': 'public_to_private'}
)
+ self.assertEqual(transitions['private_to_public_by_role'],
+ {'from_state': _u('private'),
+ 'callback': None,
+ 'callback_after': None,
+ 'guards': [],
+ 'name': _u('private_to_public_by_role'),
+ 'to_state': _u('public'),
+ 'permission': None,
+ 'roles': ['admin', 'editor'],
+ 'title': 'private_to_public_by_role',
+ }),
class TestRegisterWorkflow(unittest.TestCase):
def setUp(self):
@@ -443,3 +467,4 @@ def __init__(self, name, from_state='private', to_state='public',
self.title = title
self.extras = extras
self.guards = []
+ self.roles = []
diff --git a/repoze/workflow/workflow.py b/repoze/workflow/workflow.py
index 712f95b..c724f9e 100644
--- a/repoze/workflow/workflow.py
+++ b/repoze/workflow/workflow.py
@@ -22,7 +22,7 @@ class Workflow(object):
"""
def __init__(self, state_attr, initial_state, permission_checker=None,
- name='', description=''):
+ name='', description='', roles_checker=None):
"""
o state_attr - attribute name where a given object's current
state will be stored (object is responsible for
@@ -37,6 +37,7 @@ def __init__(self, state_attr, initial_state, permission_checker=None,
self.permission_checker = permission_checker
self.name = name
self.description = description
+ self.roles_checker = roles_checker
def __call__(self, context):
return self # allow ourselves to act as an adapter
@@ -58,7 +59,7 @@ def add_state(self, state_name, callback=None, aliases=(),
self._state_aliases[alias] = state_name
def add_transition(self, transition_name, from_state, to_state,
- callback=None, permission=None, title=None, **kw):
+ callback=None, permission=None, title=None, roles=None, callback_after=None, **kw):
""" Add a transition to the FSM. ``**kw`` must not contain
any of the keys ``from_state``, ``name``, ``to_state``, or
``callback``; these are reserved for internal use."""
@@ -73,12 +74,19 @@ def add_transition(self, transition_name, from_state, to_state,
raise WorkflowError(
'Permission %r defined without permission checker on '
'workflow' % permission)
+ if roles and self.roles_checker is None:
+ raise WorkflowError(
+ 'Roles %r defined without roles checker on '
+ 'workflow' % roles)
+
transition = kw
transition['name'] = transition_name
transition['from_state'] = from_state
transition['to_state'] = to_state
transition['callback'] = callback
+ transition['callback_after'] = callback_after
transition['permission'] = permission
+ transition['roles'] = roles or []
if title is None:
title = transition_name
transition['title'] = title
@@ -140,6 +148,11 @@ def state_info(self, content, request, context=None, from_state=None):
if not self.permission_checker(permission, context,
request):
continue
+ roles = transition.get('roles')
+ if roles is not None:
+ if not self.roles_checker(roles, context,
+ request):
+ continue
L.append(transition)
state['transitions'] = L
return states
@@ -227,6 +240,10 @@ def _transition(self, content, transition_name, context, request, guards):
state_callback(content, info)
setattr(content, self.state_attr, to_state)
+ transition_callback_after = transition['callback_after']
+
+ if transition_callback_after is not None:
+ transition_callback_after(content, info)
def transition(self, content, request, transition_name, context=None,
guards=()):
@@ -235,6 +252,11 @@ def transition(self, content, request, transition_name, context=None,
permission_guard = PermissionGuard(request, transition_name,
self.permission_checker)
guards.append(permission_guard)
+ if self.roles_checker:
+ guards = list(guards)
+ roles_guard = RolesGuard(request, transition_name,
+ self.roles_checker)
+ guards.append(roles_guard)
self._transition(content, transition_name, context, request, guards)
def _transition_to_state(self, content, to_state, context=None,
@@ -265,6 +287,11 @@ def transition_to_state(self, content, request, to_state, context=None,
permission_guard = PermissionGuard(request, to_state,
self.permission_checker)
guards.append(permission_guard)
+ if self.roles_checker:
+ guards = list(guards)
+ roles_guard = RolesGuard(request, to_state,
+ self.roles_checker)
+ guards.append(roles_guard)
self._transition_to_state(content, to_state, context, guards=guards,
request=request, skip_same=skip_same)
@@ -291,6 +318,12 @@ def get_transitions(self, content, request, context=None, from_state=None):
if not self.permission_checker(permission, context,
request):
continue
+ roles = transition.get('roles')
+ if roles is not None:
+ if self.roles_checker:
+ if not self.roles_checker(roles, context,
+ request):
+ continue
L.append(transition)
return L
@@ -317,6 +350,21 @@ def __call__(self, context, info):
permission, self.name)
)
+class RolesGuard:
+ def __init__(self, request, name, checker):
+ self.request = request
+ self.name = name
+ self.checker = checker
+
+ def __call__(self, context, info):
+ roles = info.transition.get('roles')
+ if self.request is not None and roles:
+ if not self.checker(roles, context, self.request):
+ raise WorkflowError(
+ 'one role of %s required for transition using %r' % (
+ roles, self.name)
+ )
+
def process_wf_list(wf_list, context):
# Try all workflows that have an elector first in ZCML order; if
# one of those electors returns true, return the workflow
@@ -348,7 +396,11 @@ def get_workflow(content_type, type, context=None,
content_type = providedBy(content_type)
if content_type not in (None, IDefaultWorkflow):
- wf_list = look((content_type,), IWorkflowList, name=type, default=None)
+ try:
+ wf_list = look((content_type,), IWorkflowList, name=type, default=None)
+ except ValueError as ex:
+ print(ex)
+ raise
if wf_list is not None:
wf = process_wf_list(wf_list, context)
if wf is not None:
diff --git a/repoze/workflow/zcml.py b/repoze/workflow/zcml.py
index 7d646bc..bf64ad7 100644
--- a/repoze/workflow/zcml.py
+++ b/repoze/workflow/zcml.py
@@ -27,6 +27,10 @@ class IGuardDirective(Interface):
""" A directive for a guard on a transition. """
function = GlobalObject(title=_u('enter guard function'), required=True)
+class IRoleDirective(Interface):
+ """ The interface for a key/value pair subdirective """
+ name = TextLine(title=_u('name'), required=True)
+
class IKeyValueDirective(Interface):
""" The interface for a key/value pair subdirective """
name = TextLine(title=_u('key'), required=True)
@@ -44,6 +48,7 @@ class ITransitionDirective(Interface):
permission = TextLine(title=_u('permission'), required=False)
title = TextLine(title=_u('title'), required=False)
callback = GlobalObject(title=_u('callback'), required=False)
+ callback_after = GlobalObject(title=_u('callback_after'), required=False)
class IStateDirective(Interface):
""" The interface for a state directive """
@@ -59,14 +64,15 @@ class IWorkflowDirective(Interface):
content_types = Tokens(title=_u('content_types'), required=False,
value_type=GlobalObject())
elector = GlobalObject(title=_u('elector'), required=False)
- permission_checker = GlobalObject(title=_u('checker'), required=False)
+ permission_checker = GlobalObject(title=_u('permission checker'), required=False)
description = TextLine(title=_u('description'), required=False)
+ roles_checker = GlobalObject(title=_u('roles checker'), required=False)
@implementer(IConfigurationContext, IWorkflowDirective)
class WorkflowDirective(GroupingContextDecorator):
def __init__(self, context, type, name, state_attr, initial_state,
content_types=(), elector=None, permission_checker=None,
- description=''):
+ description='', roles_checker=None):
self.context = context
self.type = type
self.name = name
@@ -80,12 +86,13 @@ def __init__(self, context, type, name, state_attr, initial_state,
self.description = description
self.transitions = [] # mutated by subdirectives
self.states = [] # mutated by subdirectives
+ self.roles_checker = roles_checker
def after(self):
def register(content_type):
workflow = Workflow(self.state_attr, self.initial_state,
self.permission_checker, self.name,
- self.description)
+ self.description, roles_checker=self.roles_checker)
for state in self.states:
try:
workflow.add_state(state.name,
@@ -105,6 +112,8 @@ def register(content_type):
transition.permission,
transition.title,
guards=transition.guards,
+ roles=transition.roles,
+ callback_after=transition.callback_after,
**transition.extras)
except WorkflowError as why:
raise ConfigurationError(str(why))
@@ -138,7 +147,7 @@ class TransitionDirective(GroupingContextDecorator):
"""
def __init__(self, context, name, from_state, to_state,
- callback=None, permission=None, title=None):
+ callback=None, permission=None, title=None, callback_after=None):
self.context = context
self.name = name
if not from_state:
@@ -149,7 +158,9 @@ def __init__(self, context, name, from_state, to_state,
self.permission = permission
self.title = title
self.guards = []
+ self.roles = []
self.extras = {} # mutated by subdirectives
+ self.callback_after = callback_after
def after(self):
self.context.transitions.append(self)
@@ -170,6 +181,9 @@ def after(self):
def guard_function(context, function):
context.guards.append(function)
+def role(context, name):
+ context.roles.append(name)
+
def key_value_pair(context, name, value):
ob = context.context
if not hasattr(ob, 'extras'):
diff --git a/setup.py b/setup.py
index 83fd813..37fbbaa 100644
--- a/setup.py
+++ b/setup.py
@@ -70,6 +70,7 @@ def _read_file(name):
url="http://www.repoze.org",
license="BSD-derived (http://www.repoze.org/LICENSE.txt)",
packages=find_packages(),
+ package_data={'': ['*.zcml']},
include_package_data=True,
namespace_packages=['repoze'],
zip_safe=False,