1 from cubicweb.devtools.apptest import EnvBasedTC |
1 from cubicweb.devtools.apptest import EnvBasedTC |
2 from cubicweb import ValidationError |
2 from cubicweb import ValidationError |
|
3 |
|
4 class WorkflowBuildingTC(EnvBasedTC): |
|
5 |
|
6 def test_wf_construction(self): |
|
7 wf = self.execute('INSERT Workflow X: X name "test"').get_entity(0, 0) |
|
8 self.execute('SET WF workflow_of ET WHERE ET name "Company"') |
|
9 foo = wf.add_state(u'foo', initial=True) |
|
10 bar = wf.add_state(u'bar') |
|
11 self.assertEquals(wf.state_by_name('bar').eid, bar.eid) |
|
12 self.assertEquals(wf.state_by_name('barrr'), None) |
|
13 baz = wf.add_transition(u'baz', (foo,), bar, ('managers',)) |
|
14 self.assertEquals(wf.transition_by_name('baz').eid, baz.eid) |
|
15 self.assertEquals(len(baz.require_group), 1) |
|
16 self.assertEquals(baz.require_group[0].name, 'managers') |
|
17 |
|
18 def test_duplicated_state(self): |
|
19 wf = self.execute('INSERT Workflow X: X name "test"').get_entity(0, 0) |
|
20 self.execute('SET WF workflow_of ET WHERE ET name "Company"') |
|
21 wf.add_state(u'foo', initial=True) |
|
22 wf.add_state(u'foo') |
|
23 ex = self.assertRaises(ValidationError, self.commit) |
|
24 # XXX enhance message |
|
25 self.assertEquals(ex.errors, {'state_of': 'unique constraint S name N, Y state_of O, Y name N failed'}) |
|
26 |
|
27 def test_duplicated_transition(self): |
|
28 wf = self.execute('INSERT Workflow X: X name "test"').get_entity(0, 0) |
|
29 self.execute('SET WF workflow_of ET WHERE ET name "Company"') |
|
30 foo = wf.add_state(u'foo', initial=True) |
|
31 bar = wf.add_state(u'bar') |
|
32 wf.add_transition(u'baz', (foo,), bar, ('managers',)) |
|
33 wf.add_transition(u'baz', (bar,), foo) |
|
34 ex = self.assertRaises(ValidationError, self.commit) |
|
35 # XXX enhance message |
|
36 self.assertEquals(ex.errors, {'transition_of': 'unique constraint S name N, Y transition_of O, Y name N failed'}) |
|
37 |
3 |
38 |
4 class WorkflowTC(EnvBasedTC): |
39 class WorkflowTC(EnvBasedTC): |
5 |
40 |
6 def setup_database(self): |
41 def setup_database(self): |
7 rschema = self.schema['in_state'] |
42 rschema = self.schema['in_state'] |
20 self.commit() |
55 self.commit() |
21 e.clear_related_cache('wf_info_for', 'object') |
56 e.clear_related_cache('wf_info_for', 'object') |
22 self.assertEquals([tr.comment for tr in e.reverse_wf_info_for], |
57 self.assertEquals([tr.comment for tr in e.reverse_wf_info_for], |
23 ['deactivate 1', 'activate 1', 'deactivate 2']) |
58 ['deactivate 1', 'activate 1', 'deactivate 2']) |
24 self.assertEquals(e.latest_trinfo().comment, 'deactivate 2') |
59 self.assertEquals(e.latest_trinfo().comment, 'deactivate 2') |
25 |
|
26 # def test_wf_construction(self): # XXX update or kill me |
|
27 # bar = self.mh.cmd_add_state(u'bar', ('Personne', 'Email'), initial=True) |
|
28 # baz = self.mh.cmd_add_transition(u'baz', ('Personne', 'Email'), |
|
29 # (foo,), bar, ('managers',)) |
|
30 # for etype in ('Personne', 'Email'): |
|
31 # t1 = self.mh.rqlexec('Any N WHERE T transition_of ET, ET name "%s", T name N' % |
|
32 # etype)[0][0] |
|
33 # self.assertEquals(t1, "baz") |
|
34 # gn = self.mh.rqlexec('Any GN WHERE T require_group G, G name GN, T eid %s' % baz)[0][0] |
|
35 # self.assertEquals(gn, 'managers') |
|
36 |
60 |
37 def test_possible_transitions(self): |
61 def test_possible_transitions(self): |
38 user = self.entity('CWUser X') |
62 user = self.entity('CWUser X') |
39 trs = list(user.possible_transitions()) |
63 trs = list(user.possible_transitions()) |
40 self.assertEquals(len(trs), 1) |
64 self.assertEquals(len(trs), 1) |