test/unittest_selectors.py
branchstable
changeset 6919 8fd6921f3e7c
parent 6908 f07dc0e8948d
child 6981 d1a279ece4a0
equal deleted inserted replaced
6918:9e607157d4cf 6919:8fd6921f3e7c
    22 
    22 
    23 from cubicweb import Binary
    23 from cubicweb import Binary
    24 from cubicweb.devtools.testlib import CubicWebTC
    24 from cubicweb.devtools.testlib import CubicWebTC
    25 from cubicweb.appobject import Selector, AndSelector, OrSelector
    25 from cubicweb.appobject import Selector, AndSelector, OrSelector
    26 from cubicweb.selectors import (is_instance, adaptable, match_user_groups,
    26 from cubicweb.selectors import (is_instance, adaptable, match_user_groups,
    27                                 multi_lines_rset, score_entity, is_in_state)
    27                                 multi_lines_rset, score_entity, is_in_state,
    28 from cubicweb.interfaces import IDownloadable
    28                                 on_transition)
    29 from cubicweb.web import action
    29 from cubicweb.web import action
    30 from cubicweb.server.migractions import ServerMigrationHelper
    30 
    31 
    31 
    32 class _1_(Selector):
    32 class _1_(Selector):
    33     def __call__(self, *args, **kwargs):
    33     def __call__(self, *args, **kwargs):
    34         return 1
    34         return 1
    35 
    35 
   136         selector = _0_()
   136         selector = _0_()
   137         selector |= _0_()
   137         selector |= _0_()
   138         selector |= _0_()
   138         selector |= _0_()
   139         self.assertEqual(selector(None), 0)
   139         self.assertEqual(selector(None), 0)
   140 
   140 
       
   141 
   141 class IsInStateSelectorTC(CubicWebTC):
   142 class IsInStateSelectorTC(CubicWebTC):
   142     def setup_database(self):
   143     def setup_database(self):
   143         mh = ServerMigrationHelper(self.repo.config, None,
   144         wf = self.shell().add_workflow("testwf", 'StateFull', default=True)
   144                                    repo=self.repo, cnx=self.cnx,
       
   145                                    interactive=False)
       
   146         wf = mh.cmd_add_workflow('testwf', 'StateFull', default=True)
       
   147         initial = wf.add_state(u'initial', initial=True)
   145         initial = wf.add_state(u'initial', initial=True)
   148         final = wf.add_state(u'final')
   146         final = wf.add_state(u'final')
   149         wf.add_transition(u'forward', (initial,), final)
   147         wf.add_transition(u'forward', (initial,), final)
   150 
   148 
   151     def test_initial_state(self):
   149     def test_initial_state(self):
   187 
   185 
   188     def test_yams_inheritance(self):
   186     def test_yams_inheritance(self):
   189         cls = self.vreg['etypes'].etype_class('Transition')
   187         cls = self.vreg['etypes'].etype_class('Transition')
   190         self.assertEqual(is_instance('BaseTransition').score_class(cls, self.request()),
   188         self.assertEqual(is_instance('BaseTransition').score_class(cls, self.request()),
   191                           3)
   189                           3)
       
   190 
       
   191 
       
   192 class WorkflowSelectorTC(CubicWebTC):
       
   193     def _commit(self):
       
   194         self.commit()
       
   195         self.wf_entity.clear_all_caches()
       
   196 
       
   197     def setup_database(self):
       
   198         wf = self.shell().add_workflow("wf_test", 'StateFull', default=True)
       
   199         created   = wf.add_state('created', initial=True)
       
   200         validated = wf.add_state('validated')
       
   201         abandoned = wf.add_state('abandoned')
       
   202         wf.add_transition('validate', created, validated, ('managers',))
       
   203         wf.add_transition('forsake', (created, validated,), abandoned, ('managers',))
       
   204 
       
   205     def setUp(self):
       
   206         super(WorkflowSelectorTC, self).setUp()
       
   207         self.req = self.request()
       
   208         self.wf_entity = self.req.create_entity('StateFull', name=u'')
       
   209         self.rset = self.wf_entity.as_rset()
       
   210         self.adapter = self.wf_entity.cw_adapt_to('IWorkflowable')
       
   211         self._commit()
       
   212         self.assertEqual(self.adapter.state, 'created')
       
   213         # enable debug mode to state/transition validation on the fly
       
   214         self.vreg.config.debugmode = True
       
   215 
       
   216     def tearDown(self):
       
   217         self.vreg.config.debugmode = False
       
   218         super(WorkflowSelectorTC, self).tearDown()
       
   219 
       
   220     def test_is_in_state(self):
       
   221         for state in ('created', 'validated', 'abandoned'):
       
   222             selector = is_in_state(state)
       
   223             self.assertEqual(selector(None, self.req, self.rset),
       
   224                              state=="created")
       
   225 
       
   226         self.adapter.fire_transition('validate')
       
   227         self._commit()
       
   228         self.assertEqual(self.adapter.state, 'validated')
       
   229 
       
   230         selector = is_in_state('created')
       
   231         self.assertEqual(selector(None, self.req, self.rset), 0)
       
   232         selector = is_in_state('validated')
       
   233         self.assertEqual(selector(None, self.req, self.rset), 1)
       
   234         selector = is_in_state('validated', 'abandoned')
       
   235         self.assertEqual(selector(None, self.req, self.rset), 1)
       
   236         selector = is_in_state('abandoned')
       
   237         self.assertEqual(selector(None, self.req, self.rset), 0)
       
   238 
       
   239         self.adapter.fire_transition('forsake')
       
   240         self._commit()
       
   241         self.assertEqual(self.adapter.state, 'abandoned')
       
   242 
       
   243         selector = is_in_state('created')
       
   244         self.assertEqual(selector(None, self.req, self.rset), 0)
       
   245         selector = is_in_state('validated')
       
   246         self.assertEqual(selector(None, self.req, self.rset), 0)
       
   247         selector = is_in_state('validated', 'abandoned')
       
   248         self.assertEqual(selector(None, self.req, self.rset), 1)
       
   249         self.assertEqual(self.adapter.state, 'abandoned')
       
   250         self.assertEqual(selector(None, self.req, self.rset), 1)
       
   251 
       
   252     def test_is_in_state_unvalid_names(self):
       
   253         selector = is_in_state("unknown")
       
   254         with self.assertRaises(ValueError) as cm:
       
   255             selector(None, self.req, self.rset)
       
   256         self.assertEqual(str(cm.exception),
       
   257                          "wf_test: unknown state(s): unknown")
       
   258         selector = is_in_state("weird", "unknown", "created", "weird")
       
   259         with self.assertRaises(ValueError) as cm:
       
   260             selector(None, self.req, self.rset)
       
   261         self.assertEqual(str(cm.exception),
       
   262                          "wf_test: unknown state(s): unknown,weird")
       
   263 
       
   264     def test_on_transition(self):
       
   265         for transition in ('validate', 'forsake'):
       
   266             selector = on_transition(transition)
       
   267             self.assertEqual(selector(None, self.req, self.rset), 0)
       
   268 
       
   269         self.adapter.fire_transition('validate')
       
   270         self._commit()
       
   271         self.assertEqual(self.adapter.state, 'validated')
       
   272 
       
   273         selector = on_transition("validate")
       
   274         self.assertEqual(selector(None, self.req, self.rset), 1)
       
   275         selector = on_transition("validate", "forsake")
       
   276         self.assertEqual(selector(None, self.req, self.rset), 1)
       
   277         selector = on_transition("forsake")
       
   278         self.assertEqual(selector(None, self.req, self.rset), 0)
       
   279 
       
   280         self.adapter.fire_transition('forsake')
       
   281         self._commit()
       
   282         self.assertEqual(self.adapter.state, 'abandoned')
       
   283 
       
   284         selector = on_transition("validate")
       
   285         self.assertEqual(selector(None, self.req, self.rset), 0)
       
   286         selector = on_transition("validate", "forsake")
       
   287         self.assertEqual(selector(None, self.req, self.rset), 1)
       
   288         selector = on_transition("forsake")
       
   289         self.assertEqual(selector(None, self.req, self.rset), 1)
       
   290 
       
   291     def test_on_transition_unvalid_names(self):
       
   292         selector = on_transition("unknown")
       
   293         with self.assertRaises(ValueError) as cm:
       
   294             selector(None, self.req, self.rset)
       
   295         self.assertEqual(str(cm.exception),
       
   296                          "wf_test: unknown transition(s): unknown")
       
   297         selector = on_transition("weird", "unknown", "validate", "weird")
       
   298         with self.assertRaises(ValueError) as cm:
       
   299             selector(None, self.req, self.rset)
       
   300         self.assertEqual(str(cm.exception),
       
   301                          "wf_test: unknown transition(s): unknown,weird")
       
   302 
       
   303     def test_on_transition_with_no_effect(self):
       
   304         """selector will not be triggered with `change_state()`"""
       
   305         self.adapter.change_state('validated')
       
   306         self._commit()
       
   307         self.assertEqual(self.adapter.state, 'validated')
       
   308 
       
   309         selector = on_transition("validate")
       
   310         self.assertEqual(selector(None, self.req, self.rset), 0)
       
   311         selector = on_transition("validate", "forsake")
       
   312         self.assertEqual(selector(None, self.req, self.rset), 0)
       
   313         selector = on_transition("forsake")
       
   314         self.assertEqual(selector(None, self.req, self.rset), 0)
   192 
   315 
   193 
   316 
   194 class MatchUserGroupsTC(CubicWebTC):
   317 class MatchUserGroupsTC(CubicWebTC):
   195     def test_owners_group(self):
   318     def test_owners_group(self):
   196         """tests usage of 'owners' group with match_user_group"""
   319         """tests usage of 'owners' group with match_user_group"""