1 # copyright 2003-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

     2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr

     3 #

     4 # This file is part of CubicWeb.

     5 #

     6 # CubicWeb is free software: you can redistribute it and/or modify it under the

     7 # terms of the GNU Lesser General Public License as published by the Free

     8 # Software Foundation, either version 2.1 of the License, or (at your option)

     9 # any later version.

    10 #

    11 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT

    12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

    13 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

    14 # details.

    15 #

    16 # You should have received a copy of the GNU Lesser General Public License along

    17 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

    18 """unit tests for selectors mechanism"""

    19 

    20 from operator import eq, lt, le, gt

    21 from contextlib import contextmanager

    22 

    23 from six.moves import range

    24 

    25 from logilab.common.testlib import TestCase, unittest_main

    26 from logilab.common.decorators import clear_cache

    27 

    28 from cubicweb import Binary

    29 from cubicweb.devtools.testlib import CubicWebTC

    30 from cubicweb.predicates import (is_instance, adaptable, match_kwargs, match_user_groups,

    31                                  multi_lines_rset, score_entity, is_in_state,

    32                                  rql_condition, relation_possible, match_form_params,

    33                                  paginated_rset)

    34 from cubicweb.selectors import on_transition # XXX on_transition is deprecated

    35 from cubicweb.view import EntityAdapter

    36 from cubicweb.web import action

    37 

    38 

    39 

    40 class ImplementsTC(CubicWebTC):

    41     def test_etype_priority(self):

    42         with self.admin_access.web_request() as req:

    43             f = req.create_entity('FakeFile', data_name=u'hop.txt', data=Binary(b'hop'),

    44                                   data_format=u'text/plain')

    45             rset = f.as_rset()

    46             anyscore = is_instance('Any')(f.__class__, req, rset=rset)

    47             idownscore = adaptable('IDownloadable')(f.__class__, req, rset=rset)

    48             self.assertTrue(idownscore > anyscore, (idownscore, anyscore))

    49             filescore = is_instance('FakeFile')(f.__class__, req, rset=rset)

    50             self.assertTrue(filescore > idownscore, (filescore, idownscore))

    51 

    52     def test_etype_inheritance_no_yams_inheritance(self):

    53         cls = self.vreg['etypes'].etype_class('Personne')

    54         with self.admin_access.web_request() as req:

    55             self.assertFalse(is_instance('Societe').score_class(cls, req))

    56 

    57     def test_yams_inheritance(self):

    58         cls = self.vreg['etypes'].etype_class('Transition')

    59         with self.admin_access.web_request() as req:

    60             self.assertEqual(is_instance('BaseTransition').score_class(cls, req),

    61                              3)

    62 

    63     def test_outer_join(self):

    64         with self.admin_access.web_request() as req:

    65             rset = req.execute('Any U,B WHERE B? bookmarked_by U, U login "anon"')

    66             self.assertEqual(is_instance('Bookmark')(None, req, rset=rset, row=0, col=1),

    67                              0)

    68 

    69 

    70 class WorkflowSelectorTC(CubicWebTC):

    71 

    72     def setUp(self):

    73         super(WorkflowSelectorTC, self).setUp()

    74         # enable debug mode to state/transition validation on the fly

    75         self.vreg.config.debugmode = True

    76 

    77     def tearDown(self):

    78         self.vreg.config.debugmode = False

    79         super(WorkflowSelectorTC, self).tearDown()

    80 

    81     def setup_database(self):

    82         with self.admin_access.shell() as shell:

    83             wf = shell.add_workflow("wf_test", 'StateFull', default=True)

    84             created   = wf.add_state('created', initial=True)

    85             validated = wf.add_state('validated')

    86             abandoned = wf.add_state('abandoned')

    87             wf.add_transition('validate', created, validated, ('managers',))

    88             wf.add_transition('forsake', (created, validated,), abandoned, ('managers',))

    89 

    90     @contextmanager

    91     def statefull_stuff(self):

    92         with self.admin_access.web_request() as req:

    93             wf_entity = req.create_entity('StateFull', name=u'')

    94             rset = wf_entity.as_rset()

    95             adapter = wf_entity.cw_adapt_to('IWorkflowable')

    96             req.cnx.commit()

    97             self.assertEqual(adapter.state, 'created')

    98             yield req, wf_entity, rset, adapter

    99 

   100     def test_is_in_state(self):

   101         with self.statefull_stuff() as (req, wf_entity, rset, adapter):

   102             for state in ('created', 'validated', 'abandoned'):

   103                 selector = is_in_state(state)

   104                 self.assertEqual(selector(None, req, rset=rset),

   105                                  state=="created")

   106 

   107             adapter.fire_transition('validate')

   108             req.cnx.commit(); wf_entity.cw_clear_all_caches()

   109             self.assertEqual(adapter.state, 'validated')

   110 

   111             clear_cache(rset, 'get_entity')

   112 

   113             selector = is_in_state('created')

   114             self.assertEqual(selector(None, req, rset=rset), 0)

   115             selector = is_in_state('validated')

   116             self.assertEqual(selector(None, req, rset=rset), 1)

   117             selector = is_in_state('validated', 'abandoned')

   118             self.assertEqual(selector(None, req, rset=rset), 1)

   119             selector = is_in_state('abandoned')

   120             self.assertEqual(selector(None, req, rset=rset), 0)

   121 

   122             adapter.fire_transition('forsake')

   123             req.cnx.commit(); wf_entity.cw_clear_all_caches()

   124             self.assertEqual(adapter.state, 'abandoned')

   125 

   126             clear_cache(rset, 'get_entity')

   127 

   128             selector = is_in_state('created')

   129             self.assertEqual(selector(None, req, rset=rset), 0)

   130             selector = is_in_state('validated')

   131             self.assertEqual(selector(None, req, rset=rset), 0)

   132             selector = is_in_state('validated', 'abandoned')

   133             self.assertEqual(selector(None, req, rset=rset), 1)

   134             self.assertEqual(adapter.state, 'abandoned')

   135             self.assertEqual(selector(None, req, rset=rset), 1)

   136 

   137     def test_is_in_state_unvalid_names(self):

   138         with self.statefull_stuff() as (req, wf_entity, rset, adapter):

   139             selector = is_in_state("unknown")

   140             with self.assertRaises(ValueError) as cm:

   141                 selector(None, req, rset=rset)

   142             self.assertEqual(str(cm.exception),

   143                              "wf_test: unknown state(s): unknown")

   144             selector = is_in_state("weird", "unknown", "created", "weird")

   145             with self.assertRaises(ValueError) as cm:

   146                 selector(None, req, rset=rset)

   147             self.assertEqual(str(cm.exception),

   148                              "wf_test: unknown state(s): unknown,weird")

   149 

   150     def test_on_transition(self):

   151         with self.statefull_stuff() as (req, wf_entity, rset, adapter):

   152             for transition in ('validate', 'forsake'):

   153                 selector = on_transition(transition)

   154                 self.assertEqual(selector(None, req, rset=rset), 0)

   155 

   156             adapter.fire_transition('validate')

   157             req.cnx.commit(); wf_entity.cw_clear_all_caches()

   158             self.assertEqual(adapter.state, 'validated')

   159 

   160             clear_cache(rset, 'get_entity')

   161 

   162             selector = on_transition("validate")

   163             self.assertEqual(selector(None, req, rset=rset), 1)

   164             selector = on_transition("validate", "forsake")

   165             self.assertEqual(selector(None, req, rset=rset), 1)

   166             selector = on_transition("forsake")

   167             self.assertEqual(selector(None, req, rset=rset), 0)

   168 

   169             adapter.fire_transition('forsake')

   170             req.cnx.commit(); wf_entity.cw_clear_all_caches()

   171             self.assertEqual(adapter.state, 'abandoned')

   172 

   173             clear_cache(rset, 'get_entity')

   174 

   175             selector = on_transition("validate")

   176             self.assertEqual(selector(None, req, rset=rset), 0)

   177             selector = on_transition("validate", "forsake")

   178             self.assertEqual(selector(None, req, rset=rset), 1)

   179             selector = on_transition("forsake")

   180             self.assertEqual(selector(None, req, rset=rset), 1)

   181 

   182     def test_on_transition_unvalid_names(self):

   183         with self.statefull_stuff() as (req, wf_entity, rset, adapter):

   184             selector = on_transition("unknown")

   185             with self.assertRaises(ValueError) as cm:

   186                 selector(None, req, rset=rset)

   187             self.assertEqual(str(cm.exception),

   188                              "wf_test: unknown transition(s): unknown")

   189             selector = on_transition("weird", "unknown", "validate", "weird")

   190             with self.assertRaises(ValueError) as cm:

   191                 selector(None, req, rset=rset)

   192             self.assertEqual(str(cm.exception),

   193                              "wf_test: unknown transition(s): unknown,weird")

   194 

   195     def test_on_transition_with_no_effect(self):

   196         """selector will not be triggered with `change_state()`"""

   197         with self.statefull_stuff() as (req, wf_entity, rset, adapter):

   198             adapter.change_state('validated')

   199             req.cnx.commit(); wf_entity.cw_clear_all_caches()

   200             self.assertEqual(adapter.state, 'validated')

   201 

   202             selector = on_transition("validate")

   203             self.assertEqual(selector(None, req, rset=rset), 0)

   204             selector = on_transition("validate", "forsake")

   205             self.assertEqual(selector(None, req, rset=rset), 0)

   206             selector = on_transition("forsake")

   207             self.assertEqual(selector(None, req, rset=rset), 0)

   208 

   209 

   210 class RelationPossibleTC(CubicWebTC):

   211 

   212     def test_rqlst_1(self):

   213         with self.admin_access.web_request() as req:

   214             selector = relation_possible('in_group')

   215             select = self.vreg.parse(req, 'Any X WHERE X is CWUser').children[0]

   216             score = selector(None, req, rset=1,

   217                              select=select, filtered_variable=select.defined_vars['X'])

   218             self.assertEqual(score, 1)

   219 

   220     def test_rqlst_2(self):

   221         with self.admin_access.web_request() as req:

   222             selector = relation_possible('in_group')

   223             select = self.vreg.parse(req, 'Any 1, COUNT(X) WHERE X is CWUser, X creation_date XD, '

   224                                      'Y creation_date YD, Y is CWGroup '

   225                                      'HAVING DAY(XD)=DAY(YD)').children[0]

   226             score = selector(None, req, rset=1,

   227                              select=select, filtered_variable=select.defined_vars['X'])

   228             self.assertEqual(score, 1)

   229 

   230     def test_ambiguous(self):

   231         # Ambiguous relations are :

   232         # (Service, fabrique_par, Personne) and (Produit, fabrique_par, Usine)

   233         # There used to be a crash here with a bad rdef choice in the strict

   234         # checking case.

   235         selector = relation_possible('fabrique_par', role='object',

   236                                      target_etype='Personne', strict=True)

   237         with self.admin_access.web_request() as req:

   238             usine = req.create_entity('Usine', lieu=u'here')

   239             score = selector(None, req, rset=usine.as_rset())

   240             self.assertEqual(0, score)

   241 

   242 

   243 class MatchUserGroupsTC(CubicWebTC):

   244     def test_owners_group(self):

   245         """tests usage of 'owners' group with match_user_group"""

   246         class SomeAction(action.Action):

   247             __regid__ = 'yo'

   248             category = 'foo'

   249             __select__ = match_user_groups('owners')

   250         self.vreg._loadedmods[__name__] = {}

   251         self.vreg.register(SomeAction)

   252         SomeAction.__registered__(self.vreg['actions'])

   253         self.assertTrue(SomeAction in self.vreg['actions']['yo'], self.vreg['actions'])

   254         try:

   255             with self.admin_access.web_request() as req:

   256                 self.create_user(req, 'john')

   257             # login as a simple user

   258             john_access = self.new_access('john')

   259             with john_access.web_request() as req:

   260                 # it should not be possible to use SomeAction not owned objects

   261                 rset = req.execute('Any G WHERE G is CWGroup, G name "managers"')

   262                 self.assertFalse('yo' in dict(self.pactions(req, rset)))

   263                 # insert a new card, and check that we can use SomeAction on our object

   264                 req.execute('INSERT Card C: C title "zoubidou"')

   265                 req.cnx.commit()

   266             with john_access.web_request() as req:

   267                 rset = req.execute('Card C WHERE C title "zoubidou"')

   268                 self.assertTrue('yo' in dict(self.pactions(req, rset)), self.pactions(req, rset))

   269             # make sure even managers can't use the action

   270             with self.admin_access.web_request() as req:

   271                 rset = req.execute('Card C WHERE C title "zoubidou"')

   272                 self.assertFalse('yo' in dict(self.pactions(req, rset)))

   273         finally:

   274             del self.vreg[SomeAction.__registry__][SomeAction.__regid__]

   275 

   276 

   277 class MultiLinesRsetTC(CubicWebTC):

   278     def setup_database(self):

   279         with self.admin_access.web_request() as req:

   280             req.execute('INSERT CWGroup G: G name "group1"')

   281             req.execute('INSERT CWGroup G: G name "group2"')

   282             req.cnx.commit()

   283 

   284     def test_default_op_in_selector(self):

   285         with self.admin_access.web_request() as req:

   286             rset = req.execute('Any G WHERE G is CWGroup')

   287             expected = len(rset)

   288             selector = multi_lines_rset(expected)

   289             self.assertEqual(selector(None, req, rset=rset), 1)

   290             self.assertEqual(selector(None, req, None), 0)

   291             selector = multi_lines_rset(expected + 1)

   292             self.assertEqual(selector(None, req, rset=rset), 0)

   293             self.assertEqual(selector(None, req, None), 0)

   294             selector = multi_lines_rset(expected - 1)

   295             self.assertEqual(selector(None, req, rset=rset), 0)

   296             self.assertEqual(selector(None, req, None), 0)

   297 

   298     def test_without_rset(self):

   299         with self.admin_access.web_request() as req:

   300             rset = req.execute('Any G WHERE G is CWGroup')

   301             expected = len(rset)

   302             selector = multi_lines_rset(expected)

   303             self.assertEqual(selector(None, req, None), 0)

   304             selector = multi_lines_rset(expected + 1)

   305             self.assertEqual(selector(None, req, None), 0)

   306             selector = multi_lines_rset(expected - 1)

   307             self.assertEqual(selector(None, req, None), 0)

   308 

   309     def test_with_operators(self):

   310         with self.admin_access.web_request() as req:

   311             rset = req.execute('Any G WHERE G is CWGroup')

   312             expected = len(rset)

   313 

   314             # Format     'expected', 'operator', 'assert'

   315             testdata = (( expected,         eq,        1),

   316                         ( expected+1,       eq,        0),

   317                         ( expected-1,       eq,        0),

   318                         ( expected,         le,        1),

   319                         ( expected+1,       le,        1),

   320                         ( expected-1,       le,        0),

   321                         ( expected-1,       gt,        1),

   322                         ( expected,         gt,        0),

   323                         ( expected+1,       gt,        0),

   324                         ( expected+1,       lt,        1),

   325                         ( expected,         lt,        0),

   326                         ( expected-1,       lt,        0))

   327 

   328             for (expected, operator, assertion) in testdata:

   329                 selector = multi_lines_rset(expected, operator)

   330                 yield self.assertEqual, selector(None, req, rset=rset), assertion

   331 

   332 

   333 class MatchKwargsTC(TestCase):

   334 

   335     def test_match_kwargs_default(self):

   336         selector = match_kwargs( set( ('a', 'b') ) )

   337         self.assertEqual(selector(None, None, a=1, b=2), 2)

   338         self.assertEqual(selector(None, None, a=1), 0)

   339         self.assertEqual(selector(None, None, c=1), 0)

   340         self.assertEqual(selector(None, None, a=1, c=1), 0)

   341 

   342     def test_match_kwargs_any(self):

   343         selector = match_kwargs( set( ('a', 'b') ), mode='any')

   344         self.assertEqual(selector(None, None, a=1, b=2), 2)

   345         self.assertEqual(selector(None, None, a=1), 1)

   346         self.assertEqual(selector(None, None, c=1), 0)

   347         self.assertEqual(selector(None, None, a=1, c=1), 1)

   348 

   349 

   350 class ScoreEntityTC(CubicWebTC):

   351 

   352     def test_intscore_entity_selector(self):

   353         with self.admin_access.web_request() as req:

   354             rset = req.execute('Any E WHERE E eid 1')

   355             selector = score_entity(lambda x: None)

   356             self.assertEqual(selector(None, req, rset=rset), 0)

   357             selector = score_entity(lambda x: "something")

   358             self.assertEqual(selector(None, req, rset=rset), 1)

   359             selector = score_entity(lambda x: object)

   360             self.assertEqual(selector(None, req, rset=rset), 1)

   361             rset = req.execute('Any G LIMIT 2 WHERE G is CWGroup')

   362             selector = score_entity(lambda x: 10)

   363             self.assertEqual(selector(None, req, rset=rset), 20)

   364             selector = score_entity(lambda x: 10, mode='any')

   365             self.assertEqual(selector(None, req, rset=rset), 10)

   366 

   367     def test_rql_condition_entity(self):

   368         with self.admin_access.web_request() as req:

   369             selector = rql_condition('X identity U')

   370             rset = req.user.as_rset()

   371             self.assertEqual(selector(None, req, rset=rset), 1)

   372             self.assertEqual(selector(None, req, entity=req.user), 1)

   373             self.assertEqual(selector(None, req), 0)

   374 

   375     def test_rql_condition_user(self):

   376         with self.admin_access.web_request() as req:

   377             selector = rql_condition('U login "admin"', user_condition=True)

   378             self.assertEqual(selector(None, req), 1)

   379             selector = rql_condition('U login "toto"', user_condition=True)

   380             self.assertEqual(selector(None, req), 0)

   381 

   382 

   383 class AdaptablePredicateTC(CubicWebTC):

   384 

   385     def test_multiple_entity_types_rset(self):

   386         class CWUserIWhatever(EntityAdapter):

   387             __regid__ = 'IWhatever'

   388             __select__ = is_instance('CWUser')

   389         class CWGroupIWhatever(EntityAdapter):

   390             __regid__ = 'IWhatever'

   391             __select__ = is_instance('CWGroup')

   392         with self.temporary_appobjects(CWUserIWhatever, CWGroupIWhatever):

   393             with self.admin_access.web_request() as req:

   394                 selector = adaptable('IWhatever')

   395                 rset = req.execute('Any X WHERE X is IN(CWGroup, CWUser)')

   396                 self.assertTrue(selector(None, req, rset=rset))

   397 

   398 

   399 class MatchFormParamsTC(CubicWebTC):

   400     """tests for match_form_params predicate"""

   401 

   402     def test_keyonly_match(self):

   403         """test standard usage: ``match_form_params('param1', 'param2')``

   404 

   405         ``param1`` and ``param2`` must be specified in request's form.

   406         """

   407         web_request = self.admin_access.web_request

   408         vid_selector = match_form_params('vid')

   409         vid_subvid_selector = match_form_params('vid', 'subvid')

   410         # no parameter => KO,KO

   411         with web_request() as req:

   412             self.assertEqual(vid_selector(None, req), 0)

   413             self.assertEqual(vid_subvid_selector(None, req), 0)

   414         # one expected parameter found => OK,KO

   415         with web_request(vid='foo') as req:

   416             self.assertEqual(vid_selector(None, req), 1)

   417             self.assertEqual(vid_subvid_selector(None, req), 0)

   418         # all expected parameters found => OK,OK

   419         with web_request(vid='foo', subvid='bar') as req:

   420             self.assertEqual(vid_selector(None, req), 1)

   421             self.assertEqual(vid_subvid_selector(None, req), 2)

   422 

   423     def test_keyvalue_match_one_parameter(self):

   424         """test dict usage: ``match_form_params(param1=value1)``

   425 

   426         ``param1`` must be specified in the request's form and its value

   427         must be ``value1``.

   428         """

   429         web_request = self.admin_access.web_request

   430         # test both positional and named parameters

   431         vid_selector = match_form_params(vid='foo')

   432         # no parameter => should fail

   433         with web_request() as req:

   434             self.assertEqual(vid_selector(None, req), 0)

   435         # expected parameter found with expected value => OK

   436         with web_request(vid='foo', subvid='bar') as req:

   437             self.assertEqual(vid_selector(None, req), 1)

   438         # expected parameter found but value is incorrect => KO

   439         with web_request(vid='bar') as req:

   440             self.assertEqual(vid_selector(None, req), 0)

   441 

   442     def test_keyvalue_match_two_parameters(self):

   443         """test dict usage: ``match_form_params(param1=value1, param2=value2)``

   444 

   445         ``param1`` and ``param2`` must be specified in the request's form and

   446         their respective value must be ``value1`` and ``value2``.

   447         """

   448         web_request = self.admin_access.web_request

   449         vid_subvid_selector = match_form_params(vid='list', subvid='tsearch')

   450         # missing one expected parameter => KO

   451         with web_request(vid='list') as req:

   452             self.assertEqual(vid_subvid_selector(None, req), 0)

   453         # expected parameters found but values are incorrect => KO

   454         with web_request(vid='list', subvid='foo') as req:

   455             self.assertEqual(vid_subvid_selector(None, req), 0)

   456         # expected parameters found and values are correct => OK

   457         with web_request(vid='list', subvid='tsearch') as req:

   458             self.assertEqual(vid_subvid_selector(None, req), 2)

   459 

   460     def test_keyvalue_multiple_match(self):

   461         """test dict usage with multiple values

   462 

   463         i.e. as in ``match_form_params(param1=('value1', 'value2'))``

   464 

   465         ``param1`` must be specified in the request's form and its value

   466         must be either ``value1`` or ``value2``.

   467         """

   468         web_request = self.admin_access.web_request

   469         vid_subvid_selector = match_form_params(vid='list', subvid=('tsearch', 'listitem'))

   470         # expected parameters found and values correct => OK

   471         with web_request(vid='list', subvid='tsearch') as req:

   472             self.assertEqual(vid_subvid_selector(None, req), 2)

   473         with web_request(vid='list', subvid='listitem') as req:

   474             self.assertEqual(vid_subvid_selector(None, req), 2)

   475         # expected parameters found but values are incorrect => OK

   476         with web_request(vid='list', subvid='foo') as req:

   477             self.assertEqual(vid_subvid_selector(None, req), 0)

   478 

   479     def test_invalid_calls(self):

   480         """checks invalid calls raise a ValueError"""

   481         # mixing named and positional arguments should fail

   482         with self.assertRaises(ValueError) as cm:

   483             match_form_params('list', x='1', y='2')

   484         self.assertEqual(str(cm.exception),

   485                          "match_form_params() can't be called with both "

   486                          "positional and named arguments")

   487         # using a dict as first and unique argument should fail

   488         with self.assertRaises(ValueError) as cm:

   489             match_form_params({'x': 1})

   490         self.assertEqual(str(cm.exception),

   491                          "match_form_params() positional arguments must be strings")

   492 

   493 

   494 class PaginatedTC(CubicWebTC):

   495     """tests for paginated_rset predicate"""

   496 

   497     def setup_database(self):

   498         with self.admin_access.repo_cnx() as cnx:

   499             for i in range(30):

   500                 cnx.create_entity('CWGroup', name=u"group%d" % i)

   501             cnx.commit()

   502 

   503     def test_paginated_rset(self):

   504         default_nb_pages = 1

   505         web_request = self.admin_access.web_request

   506         with web_request() as req:

   507             rset = req.execute('Any G WHERE G is CWGroup')

   508         self.assertEqual(len(rset), 34)

   509         with web_request(vid='list', page_size='10') as req:

   510             self.assertEqual(paginated_rset()(None, req, rset), default_nb_pages)

   511         with web_request(vid='list', page_size='20') as req:

   512             self.assertEqual(paginated_rset()(None, req, rset), default_nb_pages)

   513         with web_request(vid='list', page_size='50') as req:

   514             self.assertEqual(paginated_rset()(None, req, rset), 0)

   515         with web_request(vid='list', page_size='10/') as req:

   516             self.assertEqual(paginated_rset()(None, req, rset), 0)

   517         with web_request(vid='list', page_size='.1') as req:

   518             self.assertEqual(paginated_rset()(None, req, rset), 0)

   519         with web_request(vid='list', page_size='not_an_int') as req:

   520             self.assertEqual(paginated_rset()(None, req, rset), 0)

   521 

   522 

   523 if __name__ == '__main__':

   524     unittest_main()