Commit 9a450f17 authored by Sylvain Thénault's avatar Sylvain Thénault
Browse files

start a new workflow engine

--HG--
branch : 3.5
parent 662f35236d1c
......@@ -13,7 +13,7 @@ from logilab.common.decorators import cached
from cubicweb import typed_eid
from cubicweb.selectors import implements
from cubicweb.interfaces import IWorkflowable, IEmailable, ITree
from cubicweb.interfaces import IEmailable, ITree
class TreeMixIn(object):
......@@ -158,97 +158,6 @@ class TreeMixIn(object):
return self.req.entity_from_eid(self.path()[0])
class WorkflowableMixIn(object):
"""base mixin providing workflow helper methods for workflowable entities.
This mixin will be automatically set on class supporting the 'in_state'
relation (which implies supporting 'wf_info_for' as well)
"""
__implements__ = (IWorkflowable,)
@property
def state(self):
try:
return self.in_state[0].name
except IndexError:
self.warning('entity %s has no state', self)
return None
@property
def displayable_state(self):
return self.req._(self.state)
def wf_state(self, statename):
rset = self.req.execute('Any S, SN WHERE S name SN, S name %(n)s, S state_of E, E name %(e)s',
{'n': statename, 'e': str(self.e_schema)})
if rset:
return rset.get_entity(0, 0)
return None
def wf_transition(self, trname):
rset = self.req.execute('Any T, TN WHERE T name TN, T name %(n)s, T transition_of E, E name %(e)s',
{'n': trname, 'e': str(self.e_schema)})
if rset:
return rset.get_entity(0, 0)
return None
def change_state(self, state, trcomment=None, trcommentformat=None):
"""change the entity's state according to a state defined in given
parameters
"""
if isinstance(state, basestring):
state = self.wf_state(state)
assert state is not None, 'not a %s state: %s' % (self.id, state)
if hasattr(state, 'eid'):
stateeid = state.eid
else:
stateeid = state
stateeid = typed_eid(stateeid)
if trcomment:
self.req.set_shared_data('trcomment', trcomment)
if trcommentformat:
self.req.set_shared_data('trcommentformat', trcommentformat)
self.req.execute('SET X in_state S WHERE X eid %(x)s, S eid %(s)s',
{'x': self.eid, 's': stateeid}, 'x')
def can_pass_transition(self, trname):
"""return the Transition instance if the current user can pass the
transition with the given name, else None
"""
stateeid = self.in_state[0].eid
rset = self.req.execute('Any T,N,DS WHERE S allowed_transition T,'
'S eid %(x)s,T name %(trname)s,ET name %(et)s,'
'T name N,T destination_state DS,T transition_of ET',
{'x': stateeid, 'et': str(self.e_schema),
'trname': trname}, 'x')
for tr in rset.entities():
if tr.may_be_passed(self.eid, stateeid):
return tr
def latest_trinfo(self):
"""return the latest transition information for this entity"""
return self.reverse_wf_info_for[-1]
# __method methods ########################################################
def set_state(self, params=None):
"""change the entity's state according to a state defined in given
parameters, used to be called using __method controler facility
"""
params = params or self.req.form
self.change_state(typed_eid(params.pop('state')),
params.get('trcomment'),
params.get('trcomment_format'))
self.req.set_message(self.req._('__msg state changed'))
# specific vocabulary methods #############################################
@deprecated('use EntityFieldsForm.subject_in_state_vocabulary')
def subject_in_state_vocabulary(self, rschema, limit=None):
form = self.vreg.select('forms', 'edition', self.req, entity=self)
return form.subject_in_state_vocabulary(rschema, limit)
class EmailableMixIn(object):
"""base mixin providing the default get_email() method used by
the massmailing view
......@@ -288,7 +197,6 @@ class EmailableMixIn(object):
MI_REL_TRIGGERS = {
('in_state', 'subject'): WorkflowableMixIn,
('primary_email', 'subject'): EmailableMixIn,
('use_email', 'subject'): EmailableMixIn,
}
......
"""
:organization: Logilab
:copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.
:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
"""
from logilab.common.testlib import unittest_main
from cubicweb.devtools.apptest import EnvBasedTC
class WorkfloableMixInTC(EnvBasedTC):
def test_wf_state(self):
s = self.add_entity('State', name=u'activated')
self.execute('SET X state_of ET WHERE ET name "Bookmark", X eid %(x)s',
{'x': s.eid})
es = self.user().wf_state('activated')
self.assertEquals(es.state_of[0].name, 'CWUser')
def test_wf_transition(self):
t = self.add_entity('Transition', name=u'deactivate')
self.execute('SET X transition_of ET WHERE ET name "Bookmark", X eid %(x)s',
{'x': t.eid})
et = self.user().wf_transition('deactivate')
self.assertEquals(et.transition_of[0].name, 'CWUser')
def test_change_state(self):
user = self.user()
user.change_state(user.wf_state('deactivated').eid)
self.assertEquals(user.state, 'deactivated')
if __name__ == '__main__':
unittest_main()
......@@ -26,7 +26,7 @@ SYSTEM_ENTITIES = ('CWGroup', 'CWUser',
'CWAttribute', 'CWRelation',
'CWConstraint', 'CWConstraintType', 'CWProperty',
'CWEType', 'CWRType',
'State', 'Transition', 'TrInfo',
'Workflow', 'State', 'BaseTransition', 'Transition', 'WorkflowTransition', 'TrInfo', 'SubWorkflowExitPoint',
'RQLExpression',
)
SYSTEM_RELATIONS = (
......@@ -35,9 +35,9 @@ SYSTEM_RELATIONS = (
# metadata
'is', 'is_instance_of', 'owned_by', 'created_by', 'specializes',
# workflow related
'state_of', 'transition_of', 'initial_state', 'allowed_transition',
'workflow_of', 'state_of', 'transition_of', 'initial_state', 'allowed_transition',
'destination_state', 'in_state', 'wf_info_for', 'from_state', 'to_state',
'condition',
'condition', 'subworkflow', 'subworkflow_state', 'subworkflow_exit',
# permission
'in_group', 'require_group', 'require_permission',
'read_permission', 'update_permission', 'delete_permission', 'add_permission',
......@@ -121,8 +121,7 @@ class TestEnvironment(object):
def create_user(self, login, groups=('users',), req=None):
req = req or self.create_request()
cursor = self._orig_cnx.cursor(req)
rset = cursor.execute('INSERT CWUser X: X login %(login)s, X upassword %(passwd)s,'
'X in_state S WHERE S name "activated"',
rset = cursor.execute('INSERT CWUser X: X login %(login)s, X upassword %(passwd)s',
{'login': unicode(login), 'passwd': login.encode('utf8')})
user = rset.get_entity(0, 0)
cursor.execute('SET X in_group G WHERE X eid %%(x)s, G name IN(%s)'
......
......@@ -371,8 +371,7 @@ class RepositoryBasedTC(TestCase):
def create_user(self, user, groups=('users',), password=None, commit=True):
if password is None:
password = user
eid = self.execute('INSERT CWUser X: X login %(x)s, X upassword %(p)s,'
'X in_state S WHERE S name "activated"',
eid = self.execute('INSERT CWUser X: X login %(x)s, X upassword %(p)s',
{'x': unicode(user), 'p': password})[0][0]
groups = ','.join(repr(group) for group in groups)
self.execute('SET X in_group Y WHERE X eid %%(x)s, Y name IN (%s)' % groups,
......
wf = add_workflow(u'bmk wf', 'Bookmark')
wf.add_state(u'hop', initial=True)
......@@ -6,6 +6,7 @@
:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
"""
from yams.buildobjs import EntityType, String
from cubicweb.schema import make_workflowable
class Company(EntityType):
name = String()
......@@ -16,3 +17,7 @@ class Division(Company):
class SubDivision(Division):
__specializes_schema__ = True
from cubicweb.schemas import bootstrap, Bookmark
make_workflowable(bootstrap.CWGroup)
make_workflowable(Bookmark.Bookmark)
......@@ -58,149 +58,6 @@ class CWUserTC(BaseEntityTC):
self.assertEquals(e.dc_title(), 'member')
self.assertEquals(e.name(), u'bouah lôt')
class StateAndTransitionsTC(BaseEntityTC):
def test_transitions(self):
user = self.entity('CWUser X')
e = self.entity('State S WHERE S name "activated"')
trs = list(e.transitions(user))
self.assertEquals(len(trs), 1)
self.assertEquals(trs[0].name, u'deactivate')
self.assertEquals(trs[0].destination().name, u'deactivated')
self.assert_(user.can_pass_transition('deactivate'))
self.assert_(not user.can_pass_transition('activate'))
# test a std user get no possible transition
self.login('member')
# fetch the entity using the new session
e = self.entity('State S WHERE S name "activated"')
trs = list(e.transitions(user))
self.assertEquals(len(trs), 0)
user = self.entity('CWUser X')
self.assert_(not user.can_pass_transition('deactivate'))
self.assert_(not user.can_pass_transition('activate'))
def test_transitions_with_dest_specfied(self):
user = self.entity('CWUser X')
e = self.entity('State S WHERE S name "activated"')
e2 = self.entity('State S WHERE S name "deactivated"')
trs = list(e.transitions(user, e2.eid))
self.assertEquals(len(trs), 1)
self.assertEquals(trs[0].name, u'deactivate')
self.assertEquals(trs[0].destination().name, u'deactivated')
trs = list(e.transitions(user, e.eid))
self.assertEquals(len(trs), 0)
def test_transitions_maybe_passed(self):
self.execute('INSERT RQLExpression X: X exprtype "ERQLExpression", '
'X expression "X owned_by U", T condition X '
'WHERE T name "deactivate"')
self._test_deactivated()
def test_transitions_maybe_passed_using_has_update_perm(self):
self.execute('INSERT RQLExpression X: X exprtype "ERQLExpression", '
'X expression "U has_update_permission X", T condition X '
'WHERE T name "deactivate"')
self._test_deactivated()
def _test_deactivated(self):
ueid = self.create_user('toto').eid
self.create_user('tutu')
cnx = self.login('tutu')
cu = cnx.cursor()
self.assertRaises(ValidationError,
cu.execute, 'SET X in_state S WHERE X eid %(x)s, S name "deactivated"',
{'x': ueid}, 'x')
cnx.close()
cnx = self.login('toto')
cu = cnx.cursor()
cu.execute('SET X in_state S WHERE X eid %(x)s, S name "deactivated"',
{'x': ueid}, 'x')
cnx.commit()
self.assertRaises(ValidationError,
cu.execute, 'SET X in_state S WHERE X eid %(x)s, S name "activated"',
{'x': ueid}, 'x')
def test_transitions_selection(self):
"""
------------------------ tr1 -----------------
| state1 (CWGroup, Bookmark) | ------> | state2 (CWGroup) |
------------------------ -----------------
| tr2 ------------------
`------> | state3 (Bookmark) |
------------------
"""
state1 = self.add_entity('State', name=u'state1')
state2 = self.add_entity('State', name=u'state2')
state3 = self.add_entity('State', name=u'state3')
tr1 = self.add_entity('Transition', name=u'tr1')
tr2 = self.add_entity('Transition', name=u'tr2')
self.execute('SET X state_of Y WHERE X eid in (%s, %s), Y is CWEType, Y name "CWGroup"' %
(state1.eid, state2.eid))
self.execute('SET X state_of Y WHERE X eid in (%s, %s), Y is CWEType, Y name "Bookmark"' %
(state1.eid, state3.eid))
self.execute('SET X transition_of Y WHERE X eid %s, Y name "CWGroup"' % tr1.eid)
self.execute('SET X transition_of Y WHERE X eid %s, Y name "Bookmark"' % tr2.eid)
self.execute('SET X allowed_transition Y WHERE X eid %s, Y eid %s' %
(state1.eid, tr1.eid))
self.execute('SET X allowed_transition Y WHERE X eid %s, Y eid %s' %
(state1.eid, tr2.eid))
self.execute('SET X destination_state Y WHERE X eid %s, Y eid %s' %
(tr1.eid, state2.eid))
self.execute('SET X destination_state Y WHERE X eid %s, Y eid %s' %
(tr2.eid, state3.eid))
self.execute('SET X initial_state Y WHERE Y eid %s, X name "CWGroup"' % state1.eid)
self.execute('SET X initial_state Y WHERE Y eid %s, X name "Bookmark"' % state1.eid)
group = self.add_entity('CWGroup', name=u't1')
transitions = list(state1.transitions(group))
self.assertEquals(len(transitions), 1)
self.assertEquals(transitions[0].name, 'tr1')
bookmark = self.add_entity('Bookmark', title=u'111', path=u'/view')
transitions = list(state1.transitions(bookmark))
self.assertEquals(len(transitions), 1)
self.assertEquals(transitions[0].name, 'tr2')
def test_transitions_selection2(self):
"""
------------------------ tr1 (Bookmark) -----------------------
| state1 (CWGroup, Bookmark) | -------------> | state2 (CWGroup,Bookmark) |
------------------------ -----------------------
| tr2 (CWGroup) |
`---------------------------------/
"""
state1 = self.add_entity('State', name=u'state1')
state2 = self.add_entity('State', name=u'state2')
tr1 = self.add_entity('Transition', name=u'tr1')
tr2 = self.add_entity('Transition', name=u'tr2')
self.execute('SET X state_of Y WHERE X eid in (%s, %s), Y is CWEType, Y name "CWGroup"' %
(state1.eid, state2.eid))
self.execute('SET X state_of Y WHERE X eid in (%s, %s), Y is CWEType, Y name "Bookmark"' %
(state1.eid, state2.eid))
self.execute('SET X transition_of Y WHERE X eid %s, Y name "CWGroup"' % tr1.eid)
self.execute('SET X transition_of Y WHERE X eid %s, Y name "Bookmark"' % tr2.eid)
self.execute('SET X allowed_transition Y WHERE X eid %s, Y eid %s' %
(state1.eid, tr1.eid))
self.execute('SET X allowed_transition Y WHERE X eid %s, Y eid %s' %
(state1.eid, tr2.eid))
self.execute('SET X destination_state Y WHERE X eid %s, Y eid %s' %
(tr1.eid, state2.eid))
self.execute('SET X destination_state Y WHERE X eid %s, Y eid %s' %
(tr2.eid, state2.eid))
self.execute('SET X initial_state Y WHERE Y eid %s, X name "CWGroup"' % state1.eid)
self.execute('SET X initial_state Y WHERE Y eid %s, X name "Bookmark"' % state1.eid)
group = self.add_entity('CWGroup', name=u't1')
transitions = list(state1.transitions(group))
self.assertEquals(len(transitions), 1)
self.assertEquals(transitions[0].name, 'tr1')
bookmark = self.add_entity('Bookmark', title=u'111', path=u'/view')
transitions = list(state1.transitions(bookmark))
self.assertEquals(len(transitions), 1)
self.assertEquals(transitions[0].name, 'tr2')
class EmailAddressTC(BaseEntityTC):
def test_canonical_form(self):
eid1 = self.execute('INSERT EmailAddress X: X address "maarten.ter.huurne@philips.com"')[0][0]
......@@ -234,7 +91,6 @@ class CWUserTC(BaseEntityTC):
e = self.entity('CWUser X WHERE X login "admin"')
e.complete()
def test_matching_groups(self):
e = self.entity('CWUser X WHERE X login "admin"')
self.failUnless(e.matching_groups('managers'))
......@@ -242,23 +98,6 @@ class CWUserTC(BaseEntityTC):
self.failUnless(e.matching_groups(('xyz', 'managers')))
self.failIf(e.matching_groups(('xyz', 'abcd')))
def test_workflow_base(self):
e = self.create_user('toto')
self.assertEquals(e.state, 'activated')
activatedeid = self.execute('State X WHERE X name "activated"')[0][0]
deactivatedeid = self.execute('State X WHERE X name "deactivated"')[0][0]
e.change_state(deactivatedeid, u'deactivate 1')
self.commit()
e.change_state(activatedeid, u'activate 1')
self.commit()
e.change_state(deactivatedeid, u'deactivate 2')
self.commit()
# get a fresh user to avoid potential cache issues
e = self.entity('CWUser X WHERE X eid %s' % e.eid)
self.assertEquals([tr.comment for tr in e.reverse_wf_info_for],
[None, 'deactivate 1', 'activate 1', 'deactivate 2'])
self.assertEquals(e.latest_trinfo().comment, 'deactivate 2')
class InterfaceTC(EnvBasedTC):
......
from cubicweb.devtools.apptest import EnvBasedTC
from cubicweb import ValidationError
class WorkflowTC(EnvBasedTC):
def setup_database(self):
rschema = self.schema['in_state']
for x, y in rschema.iter_rdefs():
self.assertEquals(rschema.rproperty(x, y, 'cardinality'), '1*')
self.member = self.create_user('member')
def test_workflow_base(self):
e = self.create_user('toto')
self.assertEquals(e.state, 'activated')
e.change_state('deactivated', u'deactivate 1')
self.commit()
e.change_state('activated', u'activate 1')
self.commit()
e.change_state('deactivated', u'deactivate 2')
self.commit()
e.clear_related_cache('wf_info_for', 'object')
self.assertEquals([tr.comment for tr in e.reverse_wf_info_for],
['deactivate 1', 'activate 1', 'deactivate 2'])
self.assertEquals(e.latest_trinfo().comment, 'deactivate 2')
# def test_wf_construction(self): # XXX update or kill me
# bar = self.mh.cmd_add_state(u'bar', ('Personne', 'Email'), initial=True)
# baz = self.mh.cmd_add_transition(u'baz', ('Personne', 'Email'),
# (foo,), bar, ('managers',))
# for etype in ('Personne', 'Email'):
# t1 = self.mh.rqlexec('Any N WHERE T transition_of ET, ET name "%s", T name N' %
# etype)[0][0]
# self.assertEquals(t1, "baz")
# gn = self.mh.rqlexec('Any GN WHERE T require_group G, G name GN, T eid %s' % baz)[0][0]
# self.assertEquals(gn, 'managers')
def test_possible_transitions(self):
user = self.entity('CWUser X')
trs = list(user.possible_transitions())
self.assertEquals(len(trs), 1)
self.assertEquals(trs[0].name, u'deactivate')
self.assertEquals(trs[0].destination().name, u'deactivated')
# test a std user get no possible transition
cnx = self.login('member')
# fetch the entity using the new session
trs = list(cnx.user().possible_transitions())
self.assertEquals(len(trs), 0)
def _test_manager_deactivate(self, user):
user.clear_related_cache('in_state', 'subject')
self.assertEquals(len(user.in_state), 1)
self.assertEquals(user.state, 'deactivated')
trinfo = user.latest_trinfo()
self.assertEquals(trinfo.previous_state.name, 'activated')
self.assertEquals(trinfo.new_state.name, 'deactivated')
self.assertEquals(trinfo.comment, 'deactivate user')
self.assertEquals(trinfo.comment_format, 'text/plain')
return trinfo
def test_change_state(self):
user = self.user()
user.change_state('deactivated', comment=u'deactivate user')
trinfo = self._test_manager_deactivate(user)
self.assertEquals(trinfo.transition, None)
def test_fire_transition(self):
user = self.user()
user.fire_transition('deactivate', comment=u'deactivate user')
self.assertEquals(user.state, 'deactivated')
self._test_manager_deactivate(user)
trinfo = self._test_manager_deactivate(user)
self.assertEquals(trinfo.transition.name, 'deactivate')
# XXX test managers can change state without matching transition
def _test_stduser_deactivate(self):
ueid = self.member.eid
self.create_user('tutu')
cnx = self.login('tutu')
req = self.request()
member = req.entity_from_eid(self.member.eid)
ex = self.assertRaises(ValidationError,
member.fire_transition, 'deactivate')
self.assertEquals(ex.errors, {'by_transition': "transition may not be fired"})
cnx.close()
cnx = self.login('member')
req = self.request()
member = req.entity_from_eid(self.member.eid)
member.fire_transition('deactivate')
cnx.commit()
ex = self.assertRaises(ValidationError,
member.fire_transition, 'activate')
self.assertEquals(ex.errors, {'by_transition': "transition may not be fired"})
def test_fire_transition_owned_by(self):
self.execute('INSERT RQLExpression X: X exprtype "ERQLExpression", '
'X expression "X owned_by U", T condition X '
'WHERE T name "deactivate"')
self._test_stduser_deactivate()
def test_fire_transition_has_update_perm(self):
self.execute('INSERT RQLExpression X: X exprtype "ERQLExpression", '
'X expression "U has_update_permission X", T condition X '
'WHERE T name "deactivate"')
self._test_stduser_deactivate()
def _init_wf_with_shared_state_or_tr(self):
req = self.request()
etypes = dict(self.execute('Any N, ET WHERE ET is CWEType, ET name N'
', ET name IN ("CWGroup", "Bookmark")'))
self.grpwf = req.create_entity('Workflow', ('workflow_of', 'ET'),
ET=etypes['CWGroup'], name=u'group workflow')
self.bmkwf = req.execute('Any X WHERE X is Workflow, X workflow_of ET, ET name "Bookmark"').get_entity(0, 0)
self.state1 = self.grpwf.add_state(u'state1', initial=True)
self.execute('SET S state_of WF WHERE S eid %(s)s, WF eid %(wf)s',
{'s': self.state1.eid, 'wf': self.bmkwf.eid})
self.execute('SET WF initial_state S WHERE S eid %(s)s, WF eid %(wf)s',
{'s': self.state1.eid, 'wf': self.bmkwf.eid})
self.state2 = self.grpwf.add_state(u'state2')
self.group = self.add_entity('CWGroup', name=u't1')
self.bookmark = self.add_entity('Bookmark', title=u'111', path=u'/view')
# commit to link to the initial state
self.commit()
def test_transitions_selection(self):
"""
------------------------ tr1 -----------------
| state1 (CWGroup, Bookmark) | ------> | state2 (CWGroup) |
------------------------ -----------------
| tr2 ------------------
`------> | state3 (Bookmark) |
------------------
"""
self._init_wf_with_shared_state_or_tr()
state3 = self.bmkwf.add_state(u'state3')
tr1 = self.grpwf.add_transition(u'tr1', (self.state1,), self.state2)
tr2 = self.bmkwf.add_transition(u'tr2', (self.state1,), state3)
transitions = list(self.group.possible_transitions())
self.assertEquals(len(transitions), 1)
self.assertEquals(transitions[0].name, 'tr1')
transitions = list(self.bookmark.possible_transitions())
self.assertEquals(len(transitions), 1)
self.assertEquals(transitions[0].name, 'tr2')
def test_transitions_selection2(self):
"""
------------------------ tr1 (Bookmark) -----------------------
| state1 (CWGroup, Bookmark) | -------------> | state2 (CWGroup,Bookmark) |
------------------------ -----------------------
| tr2 (CWGroup) |
`---------------------------------/
"""
self._init_wf_with_shared_state_or_tr()
self.execute('SET S state_of WF WHERE S eid %(s)s, WF eid %(wf)s',
{'s': self.state2.eid, 'wf': self.bmkwf.eid})
tr1 = self.bmkwf.add_transition(u'tr1', (self.state1,), self.state2)
tr2 = self.grpwf.add_transition(u'tr2', (self.state1,), self.state2)
transitions = list(self.group.possible_transitions())
self.assertEquals(len(transitions), 1)
self.assertEquals(transitions[0].name, 'tr2')
transitions = list(self.bookmark.possible_transitions())
self.assertEquals(len(transitions), 1)
self.assertEquals(transitions[0].name, 'tr1')
from cubicweb.devtools.apptest import RepositoryBasedTC
class WorkflowHooksTC(RepositoryBasedTC):
def setUp(self):
RepositoryBasedTC.setUp(self)
self.wf = self.session.user.current_workflow
self.s_activated = self.wf.state_by_name('activated').eid
self.s_deactivated = self.wf.state_by_name('deactivated').eid
self.s_dummy = self.wf.add_state(u'dummy').eid
self.wf.add_transition(u'dummy', (self.s_deactivated,), self.s_dummy)
ueid = self.create_user('stduser', commit=False)
# test initial state is set
rset = self.execute('Any N WHERE S name N, X in_state S, X eid %(x)s',
{'x' : ueid})
self.failIf(rset, rset.rows)
self.commit()
initialstate = self.execute('Any N WHERE S name N, X in_state S, X eid %(x)s',
{'x' : ueid})[0][0]
self.assertEquals(initialstate, u'activated')
# give access to users group on the user's wf transitions
# so we can test wf enforcing on euser (managers don't have anymore this
# enforcement
self.execute('SET X require_group G '
'WHERE G name "users", X transition_of WF, WF eid %(wf)s',
{'wf': self.wf.eid})
self.commit()
def tearDown(self):
self.execute('DELETE X require_group G '
'WHERE G name "users", X transition_of WF, WF eid %(wf)s',
{'wf': self.wf.eid})
self.commit()