test/unittest_selectors.py
changeset 6931 0af44a38fe41
parent 6919 8fd6921f3e7c
child 6981 d1a279ece4a0
equal deleted inserted replaced
6884:6fa712e9dfa5 6931:0af44a38fe41
    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)
    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 
    30 
    31 
    31 class _1_(Selector):
    32 class _1_(Selector):
    32     def __call__(self, *args, **kwargs):
    33     def __call__(self, *args, **kwargs):
    33         return 1
    34         return 1
    34 
    35 
   134         self.assertEqual(selector(None), 1)
   135         self.assertEqual(selector(None), 1)
   135         selector = _0_()
   136         selector = _0_()
   136         selector |= _0_()
   137         selector |= _0_()
   137         selector |= _0_()
   138         selector |= _0_()
   138         self.assertEqual(selector(None), 0)
   139         self.assertEqual(selector(None), 0)
       
   140 
       
   141 
       
   142 class IsInStateSelectorTC(CubicWebTC):
       
   143     def setup_database(self):
       
   144         wf = self.shell().add_workflow("testwf", 'StateFull', default=True)
       
   145         initial = wf.add_state(u'initial', initial=True)
       
   146         final = wf.add_state(u'final')
       
   147         wf.add_transition(u'forward', (initial,), final)
       
   148 
       
   149     def test_initial_state(self):
       
   150         req = self.request()
       
   151         entity = req.create_entity('StateFull')
       
   152         selector = is_in_state(u'initial')
       
   153         self.commit()
       
   154         score = selector(entity.__class__, None, entity=entity)
       
   155         self.assertEqual(score, 1)
       
   156 
       
   157     def test_final_state(self):
       
   158         req = self.request()
       
   159         entity = req.create_entity('StateFull')
       
   160         selector = is_in_state(u'initial')
       
   161         self.commit()
       
   162         entity.cw_adapt_to('IWorkflowable').fire_transition(u'forward')
       
   163         self.commit()
       
   164         score = selector(entity.__class__, None, entity=entity)
       
   165         self.assertEqual(score, 0)
       
   166         selector = is_in_state(u'final')
       
   167         score = selector(entity.__class__, None, entity=entity)
       
   168         self.assertEqual(score, 1)
   139 
   169 
   140 
   170 
   141 class ImplementsSelectorTC(CubicWebTC):
   171 class ImplementsSelectorTC(CubicWebTC):
   142     def test_etype_priority(self):
   172     def test_etype_priority(self):
   143         req = self.request()
   173         req = self.request()
   155 
   185 
   156     def test_yams_inheritance(self):
   186     def test_yams_inheritance(self):
   157         cls = self.vreg['etypes'].etype_class('Transition')
   187         cls = self.vreg['etypes'].etype_class('Transition')
   158         self.assertEqual(is_instance('BaseTransition').score_class(cls, self.request()),
   188         self.assertEqual(is_instance('BaseTransition').score_class(cls, self.request()),
   159                           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)
   160 
   315 
   161 
   316 
   162 class MatchUserGroupsTC(CubicWebTC):
   317 class MatchUserGroupsTC(CubicWebTC):
   163     def test_owners_group(self):
   318     def test_owners_group(self):
   164         """tests usage of 'owners' group with match_user_group"""
   319         """tests usage of 'owners' group with match_user_group"""
   247 
   402 
   248 class ScoreEntitySelectorTC(CubicWebTC):
   403 class ScoreEntitySelectorTC(CubicWebTC):
   249 
   404 
   250     def test_intscore_entity_selector(self):
   405     def test_intscore_entity_selector(self):
   251         req = self.request()
   406         req = self.request()
       
   407         rset = req.execute('Any E WHERE E eid 1')
   252         selector = score_entity(lambda x: None)
   408         selector = score_entity(lambda x: None)
   253         rset = req.execute('Any E WHERE E eid 0')
       
   254         self.assertEqual(selector(None, req, rset), 0)
   409         self.assertEqual(selector(None, req, rset), 0)
   255         selector = score_entity(lambda x: "something")
   410         selector = score_entity(lambda x: "something")
   256         self.assertEqual(selector(None, req, rset), 1)
   411         self.assertEqual(selector(None, req, rset), 1)
   257         selector = score_entity(lambda x: object)
   412         selector = score_entity(lambda x: object)
   258         self.assertEqual(selector(None, req, rset), 1)
   413         self.assertEqual(selector(None, req, rset), 1)