1 # copyright 2003-2012 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 module cubicweb.server.msplanner"""

    19 

    20 from logilab.common.decorators import clear_cache

    21 from yams.buildobjs import RelationDefinition

    22 from rql import BadRQLQuery

    23 

    24 from cubicweb.devtools import get_test_db_handler, TestServerConfiguration

    25 from cubicweb.devtools.repotest import BasePlannerTC, test_plan

    26 

    27 class _SetGenerator(object):

    28     """singleton to easily create set using "s[0]" or "s[0,1,2]" for instance

    29     """

    30     def __getitem__(self, key):

    31         try:

    32             it = iter(key)

    33         except TypeError:

    34             it = (key,)

    35         return set(it)

    36 s = _SetGenerator()

    37 

    38 from cubicweb.schema import ERQLExpression

    39 from cubicweb.server.sources import AbstractSource

    40 from cubicweb.server.msplanner import MSPlanner, PartPlanInformation

    41 

    42 class FakeUserROSource(AbstractSource):

    43     support_entities = {'CWUser': False}

    44     support_relations = {}

    45     def syntax_tree_search(self, *args, **kwargs):

    46         return []

    47 

    48 

    49 class FakeCardSource(AbstractSource):

    50     support_entities = {'Card': True, 'Note': True, 'State': True}

    51     support_relations = {'in_state': True, 'multisource_rel': True, 'multisource_inlined_rel': True,

    52                          'multisource_crossed_rel': True,}

    53     dont_cross_relations = set(('fiche', 'state_of'))

    54     cross_relations = set(('multisource_crossed_rel',))

    55 

    56     def syntax_tree_search(self, *args, **kwargs):

    57         return []

    58 

    59 

    60 class FakeDataFeedSource(FakeCardSource):

    61     copy_based_source = True

    62 

    63 X_ALL_SOLS = sorted([{'X': 'Affaire'}, {'X': 'BaseTransition'}, {'X': 'Basket'},

    64                      {'X': 'Bookmark'}, {'X': 'CWAttribute'}, {'X': 'CWCache'},

    65                      {'X': 'CWConstraint'}, {'X': 'CWConstraintType'}, {'X': 'CWDataImport'}, {'X': 'CWEType'},

    66                      {'X': 'CWGroup'}, {'X': 'CWPermission'}, {'X': 'CWProperty'},

    67                      {'X': 'CWRType'}, {'X': 'CWRelation'},

    68                      {'X': 'CWSource'}, {'X': 'CWSourceHostConfig'}, {'X': 'CWSourceSchemaConfig'},

    69                      {'X': 'CWUser'}, {'X': 'CWUniqueTogetherConstraint'},

    70                      {'X': 'Card'}, {'X': 'Comment'}, {'X': 'Division'},

    71                      {'X': 'Email'}, {'X': 'EmailAddress'}, {'X': 'EmailPart'},

    72                      {'X': 'EmailThread'}, {'X': 'ExternalUri'}, {'X': 'File'},

    73                      {'X': 'Folder'}, {'X': 'Note'}, {'X': 'Old'},

    74                      {'X': 'Personne'}, {'X': 'RQLExpression'}, {'X': 'Societe'},

    75                      {'X': 'State'}, {'X': 'SubDivision'}, {'X': 'SubWorkflowExitPoint'},

    76                      {'X': 'Tag'}, {'X': 'TrInfo'}, {'X': 'Transition'},

    77                      {'X': 'Workflow'}, {'X': 'WorkflowTransition'}])

    78 

    79 

    80 # keep cnx so it's not garbage collected and the associated session is closed

    81 def setUpModule(*args):

    82     global repo, cnx

    83     handler = get_test_db_handler(TestServerConfiguration(apphome=BaseMSPlannerTC.datadir))

    84     handler.build_db_cache()

    85     repo, cnx = handler.get_repo_and_cnx()

    86 

    87 def tearDownModule(*args):

    88     global repo, cnx

    89     del repo, cnx

    90 

    91 

    92 class BaseMSPlannerTC(BasePlannerTC):

    93     """test planner related feature on a 3-sources repository:

    94 

    95     * system source supporting everything

    96     * ldap source supporting CWUser

    97     * rql source supporting Card

    98     """

    99 

   100     def setUp(self):

   101         self.__class__.repo = repo

   102         #_QuerierTC.setUp(self)

   103         self.setup()

   104         # hijack Affaire security

   105         affreadperms = list(self.schema['Affaire'].permissions['read'])

   106         self.prevrqlexpr_affaire = affreadperms[-1]

   107         # add access to type attribute so S can't be invariant

   108         affreadperms[-1] = ERQLExpression('X concerne S?, S owned_by U, S type "X"')

   109         self.schema['Affaire'].set_action_permissions('read', affreadperms)

   110         # hijack CWUser security

   111         userreadperms = list(self.schema['CWUser'].permissions['read'])

   112         self.prevrqlexpr_user = userreadperms[-1]

   113         userreadperms[-1] = ERQLExpression('X owned_by U')

   114         self.schema['CWUser'].set_action_permissions('read', userreadperms)

   115         self.add_source(FakeUserROSource, 'ldap')

   116         self.add_source(FakeCardSource, 'cards')

   117         self.add_source(FakeDataFeedSource, 'datafeed')

   118 

   119     def tearDown(self):

   120         # restore hijacked security

   121         self.restore_orig_affaire_security()

   122         self.restore_orig_cwuser_security()

   123         super(BaseMSPlannerTC, self).tearDown()

   124 

   125     def restore_orig_affaire_security(self):

   126         affreadperms = list(self.schema['Affaire'].permissions['read'])

   127         affreadperms[-1] = self.prevrqlexpr_affaire

   128         self.schema['Affaire'].set_action_permissions('read', affreadperms)

   129 

   130     def restore_orig_cwuser_security(self):

   131         if hasattr(self, '_orig_cwuser_security_restored'):

   132             return

   133         self._orig_cwuser_security_restored = True

   134         userreadperms = list(self.schema['CWUser'].permissions['read'])

   135         userreadperms[-1] = self.prevrqlexpr_user

   136         self.schema['CWUser'].set_action_permissions('read', userreadperms)

   137 

   138 

   139 class PartPlanInformationTC(BaseMSPlannerTC):

   140 

   141     def _test(self, rql, *args):

   142         if len(args) == 3:

   143             kwargs, sourcesterms, needsplit = args

   144         else:

   145             sourcesterms, needsplit = args

   146             kwargs = None

   147         plan = self._prepare_plan(rql, kwargs)

   148         union = plan.rqlst

   149         plan.preprocess(union)

   150         ppi = PartPlanInformation(plan, union.children[0])

   151         for sourcevars in ppi._sourcesterms.itervalues():

   152             for var in list(sourcevars):

   153                 solindices = sourcevars.pop(var)

   154                 sourcevars[var._ms_table_key()] = solindices

   155         self.assertEqual(ppi._sourcesterms, sourcesterms)

   156         self.assertEqual(ppi.needsplit, needsplit)

   157 

   158 

   159     def test_simple_system_only(self):

   160         """retrieve entities only supported by the system source"""

   161         self._test('CWGroup X',

   162                    {self.system: {'X': s[0]}}, False)

   163 

   164     def test_simple_system_ldap(self):

   165         """retrieve CWUser X from both sources and return concatenation of results

   166         """

   167         self._test('CWUser X',

   168                    {self.system: {'X': s[0]}, self.ldap: {'X': s[0]}}, False)

   169 

   170     def test_simple_system_rql(self):

   171         """retrieve Card X from both sources and return concatenation of results

   172         """

   173         self._test('Any X, XT WHERE X is Card, X title XT',

   174                    {self.system: {'X': s[0]}, self.cards: {'X': s[0]}}, False)

   175 

   176     def test_simple_eid_specified(self):

   177         """retrieve CWUser X from system source (eid is specified, can locate the entity)

   178         """

   179         ueid = self.session.user.eid

   180         self._test('Any X,L WHERE X eid %(x)s, X login L', {'x': ueid},

   181                    {self.system: {'X': s[0]}}, False)

   182 

   183     def test_simple_eid_invariant(self):

   184         """retrieve CWUser X from system source (eid is specified, can locate the entity)

   185         """

   186         ueid = self.session.user.eid

   187         self._test('Any X WHERE X eid %(x)s', {'x': ueid},

   188                    {self.system: {'x': s[0]}}, False)

   189 

   190     def test_simple_invariant(self):

   191         """retrieve CWUser X from system source only (X is invariant and in_group not supported by ldap source)

   192         """

   193         self._test('Any X WHERE X is CWUser, X in_group G, G name "users"',

   194                    {self.system: {'X': s[0], 'G': s[0], 'in_group': s[0]}}, False)

   195 

   196     def test_security_has_text(self):

   197         """retrieve CWUser X from system source only (has_text not supported by ldap source)

   198         """

   199         # specify CWUser instead of any since the way this test is written we aren't well dealing

   200         # with ambigous query (eg only considering the first solution)

   201         self._test('CWUser X WHERE X has_text "bla"',

   202                    {self.system: {'X': s[0]}}, False)

   203 

   204     def test_complex_base(self):

   205         """

   206         1. retrieve Any X, L WHERE X is CWUser, X login L from system and ldap sources, store

   207            concatenation of results into a temporary table

   208         2. return the result of Any X, L WHERE X is TMP, X login L, X in_group G,

   209            G name 'users' on the system source

   210         """

   211         self._test('Any X,L WHERE X is CWUser, X in_group G, X login L, G name "users"',

   212                    {self.system: {'X': s[0], 'G': s[0], 'in_group': s[0]},

   213                     self.ldap : {'X': s[0]}}, True)

   214 

   215     def test_complex_invariant_ordered(self):

   216         """

   217         1. retrieve Any X,AA WHERE X modification_date AA from system and ldap sources, store

   218            concatenation of results into a temporary table

   219         2. return the result of Any X,AA ORDERBY AA WHERE %s owned_by X, X modification_date AA

   220            on the system source

   221         """

   222         ueid = self.session.user.eid

   223         self._test('Any X,AA ORDERBY AA WHERE E eid %(x)s, E owned_by X, X modification_date AA', {'x': ueid},

   224                    {self.system: {'x': s[0], 'X': s[0], 'owned_by': s[0]},

   225                     self.ldap : {'X': s[0]}}, True)

   226 

   227     def test_complex_invariant(self):

   228         """

   229         1. retrieve Any X,L,AA WHERE X login L, X modification_date AA from system and ldap sources, store

   230            concatenation of results into a temporary table

   231         2. return the result of Any X,L,AA WHERE %s owned_by X, X login L, X modification_date AA

   232            on the system source

   233         """

   234         ueid = self.session.user.eid

   235         self._test('Any X,L,AA WHERE E eid %(x)s, E owned_by X, X login L, X modification_date AA', {'x': ueid},

   236                    {self.system: {'x': s[0], 'X': s[0], 'owned_by': s[0]},

   237                     self.ldap : {'X': s[0]}}, True)

   238 

   239     def test_complex_ambigous(self):

   240         """retrieve CWUser X from system and ldap sources, Person X from system source only

   241         """

   242         self._test('Any X,F WHERE X firstname F',

   243                    {self.system: {'X': s[0, 1]},

   244                     self.ldap: {'X': s[0]}}, True)

   245 

   246     def test_complex_multiple(self):

   247         """

   248         1. retrieve Any X,A,Y,B WHERE X login A, Y login B from system and ldap sources, store

   249            cartesian product of results into a temporary table

   250         2. return the result of Any X,Y WHERE X login 'syt', Y login 'adim'

   251            on the system source

   252         """

   253         ueid = self.session.user.eid

   254         self._test('Any X,Y WHERE X login "syt", Y login "adim"', {'x': ueid},

   255                    {self.system: {'Y': s[0], 'X': s[0]},

   256                     self.ldap: {'Y': s[0], 'X': s[0]}}, True)

   257 

   258     def test_complex_aggregat(self):

   259         solindexes = set(range(len([e for e in self.schema.entities() if not e.final])))

   260         self._test('Any MAX(X)',

   261                    {self.system: {'X': solindexes}}, False)

   262 

   263     def test_complex_optional(self):

   264         ueid = self.session.user.eid

   265         self._test('Any U WHERE WF wf_info_for X, X eid %(x)s, WF owned_by U?, WF from_state FS', {'x': ueid},

   266                    {self.system: {'WF': s[0], 'FS': s[0], 'U': s[0],

   267                                   'from_state': s[0], 'owned_by': s[0], 'wf_info_for': s[0],

   268                                   'x': s[0]}},

   269                    False)

   270 

   271     def test_exists4(self):

   272         """

   273         State S could come from both rql source and system source,

   274         but since X cannot come from the rql source, the solution

   275         {self.cards : 'S'} must be removed

   276         """

   277         self._test('Any G,L WHERE X in_group G, X login L, G name "managers", '

   278                    'EXISTS(X copain T, T login L, T login in ("comme", "cochon")) OR '

   279                    'EXISTS(X in_state S, S name "pascontent", NOT X copain T2, T2 login "billy")',

   280                    {self.system: {'X': s[0], 'S': s[0], 'T2': s[0], 'T': s[0], 'G': s[0], 'copain': s[0], 'in_group': s[0]},

   281                     self.ldap: {'X': s[0], 'T2': s[0], 'T': s[0]}},

   282                    True)

   283 

   284     def test_relation_need_split(self):

   285         self._test('Any X, S WHERE X in_state S',

   286                    {self.system: {'X': s[0, 1, 2], 'S': s[0, 1, 2]},

   287                     self.cards: {'X': s[2], 'S': s[2]}},

   288                    True)

   289 

   290     def test_not_relation_need_split(self):

   291         self._test('Any SN WHERE NOT X in_state S, S name SN',

   292                    {self.cards: {'X': s[2], 'S': s[0, 1, 2]},

   293                     self.system: {'X': s[0, 1, 2], 'S': s[0, 1, 2]}},

   294                    True)

   295 

   296     def test_not_relation_no_split_external(self):

   297         repo._type_source_cache[999999] = ('Note', 'cards', 999999, 'cards')

   298         # similar to the above test but with an eid coming from the external source.

   299         # the same plan may be used, since we won't find any record in the system source

   300         # linking 9999999 to a state

   301         self._test('Any SN WHERE NOT X in_state S, X eid %(x)s, S name SN',

   302                    {'x': 999999},

   303                    {self.cards: {'x': s[0], 'S': s[0]},

   304                     self.system: {'x': s[0], 'S': s[0]}},

   305                    False)

   306 

   307     def test_relation_restriction_ambigous_need_split(self):

   308         self._test('Any X,T WHERE X in_state S, S name "pending", T tags X',

   309                    {self.system: {'X': s[0, 1, 2], 'S': s[0, 1, 2], 'T': s[0, 1, 2], 'tags': s[0, 1, 2]},

   310                     self.cards: {'X': s[2], 'S': s[2]}},

   311                    True)

   312 

   313     def test_simplified_var(self):

   314         repo._type_source_cache[999999] = ('Note', 'cards', 999999, 'cards')

   315         # need access to source since X table has to be accessed because of the outer join

   316         self._test('Any U WHERE U in_group G, (G name IN ("managers", "logilab") OR (X require_permission P?, P name "bla", P require_group G)), X eid %(x)s, U eid %(u)s',

   317                    {'x': 999999, 'u': self.session.user.eid},

   318                    {self.system: {'P': s[0], 'G': s[0],

   319                                   'require_permission': s[0], 'in_group': s[0], 'P': s[0], 'require_group': s[0],

   320                                   'u': s[0]},

   321                     self.cards: {'X': s[0]}},

   322                    True)

   323 

   324     def test_delete_relation1(self):

   325         ueid = self.session.user.eid

   326         self._test('Any X, Y WHERE X created_by Y, X eid %(x)s, NOT Y eid %(y)s',

   327                    {'x': ueid, 'y': ueid},

   328                    {self.system: {'Y': s[0], 'created_by': s[0], 'x': s[0]}},

   329                    False)

   330 

   331     def test_crossed_relation_eid_1_needattr(self):

   332         repo._type_source_cache[999999] = ('Note', 'system', 999999, 'system')

   333         ueid = self.session.user.eid

   334         self._test('Any Y,T WHERE X eid %(x)s, X multisource_crossed_rel Y, Y type T',

   335                    {'x': 999999,},

   336                    {self.cards: {'Y': s[0]}, self.system: {'Y': s[0], 'x': s[0]}},

   337                    True)

   338 

   339     def test_crossed_relation_eid_1_invariant(self):

   340         repo._type_source_cache[999999] = ('Note', 'system', 999999, 'system')

   341         self._test('Any Y WHERE X eid %(x)s, X multisource_crossed_rel Y',

   342                    {'x': 999999},

   343                    {self.system: {'Y': s[0], 'x': s[0]}},

   344                    False)

   345 

   346     def test_crossed_relation_eid_2_invariant(self):

   347         repo._type_source_cache[999999] = ('Note', 'cards', 999999, 'cards')

   348         self._test('Any Y WHERE X eid %(x)s, X multisource_crossed_rel Y',

   349                    {'x': 999999,},

   350                    {self.cards: {'Y': s[0], 'multisource_crossed_rel': s[0], 'x': s[0]},

   351                     self.system: {'Y': s[0], 'multisource_crossed_rel': s[0], 'x': s[0]}},

   352                    False)

   353 

   354     def test_version_crossed_depends_on_1(self):

   355         repo._type_source_cache[999999] = ('Note', 'cards', 999999, 'cards')

   356         self._test('Any X,AD,AE WHERE E eid %(x)s, E multisource_crossed_rel X, X in_state AD, AD name AE',

   357                    {'x': 999999},

   358                    {self.cards: {'X': s[0], 'AD': s[0], 'multisource_crossed_rel': s[0], 'x': s[0]},

   359                     self.system: {'X': s[0], 'AD': s[0], 'multisource_crossed_rel': s[0], 'x': s[0]}},

   360                    True)

   361 

   362     def test_version_crossed_depends_on_2(self):

   363         repo._type_source_cache[999999] = ('Note', 'system', 999999, 'system')

   364         self._test('Any X,AD,AE WHERE E eid %(x)s, E multisource_crossed_rel X, X in_state AD, AD name AE',

   365                    {'x': 999999},

   366                    {self.cards: {'X': s[0], 'AD': s[0]},

   367                     self.system: {'X': s[0], 'AD': s[0], 'x': s[0]}},

   368                     True)

   369 

   370     def test_simplified_var_3(self):

   371         repo._type_source_cache[999999] = ('Note', 'cards', 999999, 'cards')

   372         repo._type_source_cache[999998] = ('State', 'cards', 999998, 'cards')

   373         self._test('Any S,T WHERE S eid %(s)s, N eid %(n)s, N type T, N is Note, S is State',

   374                    {'n': 999999, 's': 999998},

   375                    {self.cards: {'s': s[0], 'N': s[0]}}, False)

   376 

   377 

   378 

   379 class MSPlannerTC(BaseMSPlannerTC):

   380 

   381     def setUp(self):

   382         BaseMSPlannerTC.setUp(self)

   383         self.planner = MSPlanner(self.o.schema, self.repo.vreg.rqlhelper)

   384         for cached in ('rel_type_sources', 'can_cross_relation', 'is_multi_sources_relation'):

   385             clear_cache(self.repo, cached)

   386 

   387     _test = test_plan

   388 

   389     def test_simple_system_only(self):

   390         """retrieve entities only supported by the system source

   391         """

   392         self._test('CWGroup X',

   393                    [('OneFetchStep', [('Any X WHERE X is CWGroup', [{'X': 'CWGroup'}])],

   394                      None, None, [self.system], {}, [])])

   395 

   396     def test_simple_system_only_limit(self):

   397         """retrieve entities only supported by the system source

   398         """

   399         self._test('CWGroup X LIMIT 10',

   400                    [('OneFetchStep', [('Any X LIMIT 10 WHERE X is CWGroup', [{'X': 'CWGroup'}])],

   401                      10, None, [self.system], {}, [])])

   402 

   403     def test_simple_system_only_limit_offset(self):

   404         """retrieve entities only supported by the system source

   405         """

   406         self._test('CWGroup X LIMIT 10 OFFSET 10',

   407                    [('OneFetchStep', [('Any X LIMIT 10 OFFSET 10 WHERE X is CWGroup', [{'X': 'CWGroup'}])],

   408                      10, 10, [self.system], {}, [])])

   409 

   410     def test_simple_system_ldap(self):

   411         """retrieve CWUser X from both sources and return concatenation of results

   412         """

   413         self._test('CWUser X',

   414                    [('OneFetchStep', [('Any X WHERE X is CWUser', [{'X': 'CWUser'}])],

   415                      None, None, [self.ldap, self.system], {}, [])])

   416 

   417     def test_simple_system_ldap_limit(self):

   418         """retrieve CWUser X from both sources and return concatenation of results

   419         """

   420         self._test('CWUser X LIMIT 10',

   421                    [('OneFetchStep', [('Any X LIMIT 10 WHERE X is CWUser', [{'X': 'CWUser'}])],

   422                      10, None, [self.ldap, self.system], {}, [])])

   423 

   424     def test_simple_system_ldap_limit_offset(self):

   425         """retrieve CWUser X from both sources and return concatenation of results

   426         """

   427         self._test('CWUser X LIMIT 10 OFFSET 10',

   428                    [('OneFetchStep', [('Any X LIMIT 10 OFFSET 10 WHERE X is CWUser', [{'X': 'CWUser'}])],

   429                      10, 10, [self.ldap, self.system], {}, [])])

   430 

   431     def test_simple_system_ldap_ordered_limit_offset(self):

   432         """retrieve CWUser X from both sources and return concatenation of results

   433         """

   434         self._test('CWUser X ORDERBY X LIMIT 10 OFFSET 10',

   435                    [('AggrStep', 'SELECT table0.C0 FROM table0\nORDER BY table0.C0\nLIMIT 10\nOFFSET 10', None, [

   436                        ('FetchStep', [('Any X WHERE X is CWUser', [{'X': 'CWUser'}])],

   437                         [self.ldap, self.system], {}, {'X': 'table0.C0'}, []),

   438                        ]),

   439                    ])

   440     def test_simple_system_ldap_aggregat(self):

   441         """retrieve CWUser X from both sources and return concatenation of results

   442         """

   443         # COUNT(X) is kept in sub-step and transformed into SUM(X) in the AggrStep

   444         self._test('Any COUNT(X) WHERE X is CWUser',

   445                    [('AggrStep', 'SELECT SUM(table0.C0) FROM table0', None, [

   446                        ('FetchStep', [('Any COUNT(X) WHERE X is CWUser', [{'X': 'CWUser'}])],

   447                         [self.ldap, self.system], {}, {'COUNT(X)': 'table0.C0'}, []),

   448                        ]),

   449                    ])

   450 

   451     def test_simple_system_rql(self):

   452         """retrieve Card X from both sources and return concatenation of results

   453         """

   454         self._test('Any X, XT WHERE X is Card, X title XT',

   455                    [('OneFetchStep', [('Any X,XT WHERE X is Card, X title XT', [{'X': 'Card', 'XT': 'String'}])],

   456                      None, None, [self.cards, self.system], {}, [])])

   457 

   458     def test_simple_eid_specified(self):

   459         """retrieve CWUser X from system source (eid is specified, can locate the entity)

   460         """

   461         ueid = self.session.user.eid

   462         self._test('Any X,L WHERE X eid %(x)s, X login L',

   463                    [('OneFetchStep', [('Any X,L WHERE X eid %s, X login L'%ueid, [{'X': 'CWUser', 'L': 'String'}])],

   464                      None, None, [self.system], {}, [])],

   465                    {'x': ueid})

   466 

   467     def test_simple_eid_invariant(self):

   468         """retrieve CWUser X from system source (eid is specified, can locate the entity)

   469         """

   470         ueid = self.session.user.eid

   471         self._test('Any X WHERE X eid %(x)s',

   472                    [('OneFetchStep', [('Any %s'%ueid, [{}])],

   473                      None, None, [self.system], {}, [])],

   474                    {'x': ueid})

   475 

   476     def test_simple_invariant(self):

   477         """retrieve CWUser X from system source only (X is invariant and in_group not supported by ldap source)

   478         """

   479         self._test('Any X WHERE X is CWUser, X in_group G, G name "users"',

   480                    [('OneFetchStep', [('Any X WHERE X is CWUser, X in_group G, G name "users"',

   481                                        [{'X': 'CWUser', 'G': 'CWGroup'}])],

   482                      None, None, [self.system], {}, [])])

   483 

   484     def test_complex_base(self):

   485         """

   486         1. retrieve Any X, L WHERE X is CWUser, X login L from system and ldap sources, store

   487            concatenation of results into a temporary table

   488         2. return the result of Any X, L WHERE X is TMP, X login LX in_group G,

   489            G name 'users' on the system source

   490         """

   491         self._test('Any X,L WHERE X is CWUser, X in_group G, X login L, G name "users"',

   492                    [('FetchStep', [('Any X,L WHERE X login L, X is CWUser', [{'X': 'CWUser', 'L': 'String'}])],

   493                      [self.ldap, self.system], None,

   494                      {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, []),

   495                     ('OneFetchStep', [('Any X,L WHERE X in_group G, X login L, G name "users", G is CWGroup, X is CWUser',

   496                                        [{'X': 'CWUser', 'L': 'String', 'G': 'CWGroup'}])],

   497                      None, None, [self.system],

   498                      {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, [])

   499                     ])

   500 

   501     def test_complex_base_limit_offset(self):

   502         """

   503         1. retrieve Any X, L WHERE X is CWUser, X login L from system and ldap sources, store

   504            concatenation of results into a temporary table

   505         2. return the result of Any X, L WHERE X is TMP, X login LX in_group G,

   506            G name 'users' on the system source

   507         """

   508         self._test('Any X,L LIMIT 10 OFFSET 10 WHERE X is CWUser, X in_group G, X login L, G name "users"',

   509                    [('FetchStep', [('Any X,L WHERE X login L, X is CWUser', [{'X': 'CWUser', 'L': 'String'}])],

   510                      [self.ldap, self.system], None,

   511                      {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, []),

   512                     ('OneFetchStep', [('Any X,L LIMIT 10 OFFSET 10 WHERE X in_group G, X login L, G name "users", G is CWGroup, X is CWUser',

   513                                        [{'X': 'CWUser', 'L': 'String', 'G': 'CWGroup'}])],

   514                      10, 10,

   515                      [self.system], {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, [])

   516                     ])

   517 

   518     def test_complex_ordered(self):

   519         self._test('Any L ORDERBY L WHERE X login L',

   520                    [('AggrStep', 'SELECT table0.C0 FROM table0\nORDER BY table0.C0', None,

   521                      [('FetchStep', [('Any L WHERE X login L, X is CWUser',

   522                                       [{'X': 'CWUser', 'L': 'String'}])],

   523                        [self.ldap, self.system], {}, {'X.login': 'table0.C0', 'L': 'table0.C0'}, []),

   524                       ])

   525                     ])

   526 

   527     def test_complex_ordered_limit_offset(self):

   528         self._test('Any L ORDERBY L LIMIT 10 OFFSET 10 WHERE X login L',

   529                    [('AggrStep', 'SELECT table0.C0 FROM table0\nORDER BY table0.C0\nLIMIT 10\nOFFSET 10', None,

   530                      [('FetchStep', [('Any L WHERE X login L, X is CWUser',

   531                                       [{'X': 'CWUser', 'L': 'String'}])],

   532                        [self.ldap, self.system], {}, {'X.login': 'table0.C0', 'L': 'table0.C0'}, []),

   533                       ])

   534                     ])

   535 

   536     def test_complex_invariant_ordered(self):

   537         """

   538         1. retrieve Any X,AA WHERE X modification_date AA from system and ldap sources, store

   539            concatenation of results into a temporary table

   540         2. return the result of Any X,AA ORDERBY AA WHERE %s owned_by X, X modification_date AA

   541            on the system source

   542 

   543         herrr, this is what is expected by the XXX :(, not the actual result (which is correct anyway)

   544         """

   545         ueid = self.session.user.eid

   546         self._test('Any X,AA ORDERBY AA WHERE E eid %(x)s, E owned_by X, X modification_date AA',

   547                    [('FetchStep',

   548                      [('Any X,AA WHERE X modification_date AA, X is CWUser',

   549                        [{'AA': 'Datetime', 'X': 'CWUser'}])],

   550                      [self.ldap, self.system], None,

   551                      {'AA': 'table0.C1', 'X': 'table0.C0', 'X.modification_date': 'table0.C1'}, []),

   552                     ('OneFetchStep',

   553                      [('Any X,AA ORDERBY AA WHERE %s owned_by X, X modification_date AA, X is CWUser' % ueid,

   554                        [{'AA': 'Datetime', 'X': 'CWUser'}])],

   555                      None, None, [self.system],

   556                      {'AA': 'table0.C1', 'X': 'table0.C0', 'X.modification_date': 'table0.C1'}, []),

   557                     ],

   558                    {'x': ueid})

   559 

   560     def test_complex_invariant(self):

   561         """

   562         1. retrieve Any X,L,AA WHERE X login L, X modification_date AA from system and ldap sources, store

   563            concatenation of results into a temporary table

   564         2. return the result of Any X,L,AA WHERE %s owned_by X, X login L, X modification_date AA

   565            on the system source

   566         """

   567         ueid = self.session.user.eid

   568         self._test('Any X,L,AA WHERE E eid %(x)s, E owned_by X, X login L, X modification_date AA',

   569                    [('FetchStep', [('Any X,L,AA WHERE X login L, X modification_date AA, X is CWUser',

   570                                     [{'AA': 'Datetime', 'X': 'CWUser', 'L': 'String'}])],

   571                      [self.ldap, self.system], None,

   572                      {'AA': 'table0.C2', 'X': 'table0.C0', 'X.login': 'table0.C1', 'X.modification_date': 'table0.C2', 'L': 'table0.C1'}, []),

   573                     ('OneFetchStep', [('Any X,L,AA WHERE %s owned_by X, X login L, X modification_date AA, X is CWUser'%ueid,

   574                                        [{'AA': 'Datetime', 'X': 'CWUser', 'L': 'String'}])],

   575                      None, None, [self.system],

   576                      {'AA': 'table0.C2', 'X': 'table0.C0', 'X.login': 'table0.C1', 'X.modification_date': 'table0.C2', 'L': 'table0.C1'}, [])],

   577                    {'x': ueid})

   578 

   579     def test_complex_ambigous(self):

   580         """retrieve CWUser X from system and ldap sources, Person X from system source only

   581         """

   582         self._test('Any X,F WHERE X firstname F',

   583                    [('UnionStep', None, None, [

   584                        ('OneFetchStep', [('Any X,F WHERE X firstname F, X is CWUser',

   585                                           [{'X': 'CWUser', 'F': 'String'}])],

   586                         None, None, [self.ldap, self.system], {}, []),

   587                        ('OneFetchStep', [('Any X,F WHERE X firstname F, X is Personne',

   588                                           [{'X': 'Personne', 'F': 'String'}])],

   589                         None, None, [self.system], {}, []),

   590                        ]),

   591                     ])

   592 

   593     def test_complex_ambigous_limit_offset(self):

   594         """retrieve CWUser X from system and ldap sources, Person X from system source only

   595         """

   596         self._test('Any X,F LIMIT 10 OFFSET 10 WHERE X firstname F',

   597                    [('UnionStep', 10, 10, [

   598                        ('OneFetchStep', [('Any X,F WHERE X firstname F, X is CWUser',

   599                                           [{'X': 'CWUser', 'F': 'String'}])],

   600                         None, None,

   601                         [self.ldap, self.system], {}, []),

   602                        ('OneFetchStep', [('Any X,F WHERE X firstname F, X is Personne',

   603                                           [{'X': 'Personne', 'F': 'String'}])],

   604                         None, None, [self.system], {}, []),

   605                        ]),

   606                     ])

   607 

   608     def test_complex_ambigous_ordered(self):

   609         """

   610         1. retrieve CWUser X from system and ldap sources, Person X from system source only, store

   611            each result in the same temp table

   612         2. return content of the table sorted

   613         """

   614         self._test('Any X,F ORDERBY F WHERE X firstname F',

   615                    [('AggrStep', 'SELECT table0.C0, table0.C1 FROM table0\nORDER BY table0.C1', None,

   616                      [('FetchStep', [('Any X,F WHERE X firstname F, X is CWUser',

   617                                       [{'X': 'CWUser', 'F': 'String'}])],

   618                        [self.ldap, self.system], {},

   619                        {'X': 'table0.C0', 'X.firstname': 'table0.C1', 'F': 'table0.C1'}, []),

   620                       ('FetchStep', [('Any X,F WHERE X firstname F, X is Personne',

   621                                       [{'X': 'Personne', 'F': 'String'}])],

   622                        [self.system], {},

   623                        {'X': 'table0.C0', 'X.firstname': 'table0.C1', 'F': 'table0.C1'}, []),

   624                       ]),

   625                     ])

   626 

   627     def test_complex_multiple(self):

   628         """

   629         1. retrieve Any X,A,Y,B WHERE X login A, Y login B from system and ldap sources, store

   630            cartesian product of results into a temporary table

   631         2. return the result of Any X,Y WHERE X login 'syt', Y login 'adim'

   632            on the system source

   633         """

   634         ueid = self.session.user.eid

   635         self._test('Any X,Y WHERE X login "syt", Y login "adim"',

   636                    [('FetchStep',

   637                      [('Any X WHERE X login "syt", X is CWUser', [{'X': 'CWUser'}])],

   638                      [self.ldap, self.system], None,

   639                      {'X': 'table0.C0'}, []),

   640                     ('FetchStep',

   641                      [('Any Y WHERE Y login "adim", Y is CWUser', [{'Y': 'CWUser'}])],

   642                      [self.ldap, self.system], None,

   643                      {'Y': 'table1.C0'}, []),

   644                     ('OneFetchStep',

   645                      [('Any X,Y WHERE X is CWUser, Y is CWUser', [{'X': 'CWUser', 'Y': 'CWUser'}])],

   646                      None, None, [self.system],

   647                      {'X': 'table0.C0', 'Y': 'table1.C0'}, [])

   648                     ], {'x': ueid})

   649 

   650     def test_complex_multiple_limit_offset(self):

   651         """

   652         1. retrieve Any X,A,Y,B WHERE X login A, Y login B from system and ldap sources, store

   653            cartesian product of results into a temporary table

   654         2. return the result of Any X,Y WHERE X login 'syt', Y login 'adim'

   655            on the system source

   656         """

   657         self._test('Any X,Y LIMIT 10 OFFSET 10 WHERE X login "syt", Y login "adim"',

   658                    [('FetchStep',

   659                      [('Any X WHERE X login "syt", X is CWUser', [{'X': 'CWUser'}])],

   660                      [self.ldap, self.system], None, {'X': 'table0.C0'}, []),

   661                     ('FetchStep',

   662                      [('Any Y WHERE Y login "adim", Y is CWUser', [{'Y': 'CWUser'}])],

   663                      [self.ldap, self.system], None, {'Y': 'table1.C0'}, []),

   664                     ('OneFetchStep',

   665                      [('Any X,Y LIMIT 10 OFFSET 10 WHERE X is CWUser, Y is CWUser', [{'X': 'CWUser', 'Y': 'CWUser'}])],

   666                      10, 10, [self.system],

   667                      {'X': 'table0.C0', 'Y': 'table1.C0'}, [])

   668                     ])

   669 

   670     def test_complex_aggregat(self):

   671         self._test('Any MAX(X)',

   672                    [('OneFetchStep',

   673                      [('Any MAX(X)', X_ALL_SOLS)],

   674                      None, None, [self.system], {}, [])

   675                     ])

   676 

   677     def test_complex_typed_aggregat(self):

   678         self._test('Any MAX(X) WHERE X is Card',

   679                    [('AggrStep', 'SELECT MAX(table0.C0) FROM table0',  None,

   680                      [('FetchStep',

   681                        [('Any MAX(X) WHERE X is Card', [{'X': 'Card'}])],

   682                        [self.cards, self.system], {}, {'MAX(X)': 'table0.C0'}, [])

   683                       ])

   684                     ])

   685 

   686     def test_complex_greater_eid(self):

   687         self._test('Any X WHERE X eid > 12',

   688                    [('OneFetchStep',

   689                      [('Any X WHERE X eid > 12', X_ALL_SOLS)],

   690                      None, None, [self.system], {}, [])

   691                     ])

   692 

   693     def test_complex_greater_typed_eid(self):

   694         self._test('Any X WHERE X eid > 12, X is Card',

   695                    [('OneFetchStep',

   696                      [('Any X WHERE X eid > 12, X is Card', [{'X': 'Card'}])],

   697                      None, None, [self.system], {}, [])

   698                     ])

   699 

   700     def test_complex_optional(self):

   701         ueid = self.session.user.eid

   702         self._test('Any U WHERE WF wf_info_for X, X eid %(x)s, WF owned_by U?, WF from_state FS',

   703                    [('OneFetchStep', [('Any U WHERE WF wf_info_for %s, WF owned_by U?, WF from_state FS' % ueid,

   704                                        [{'WF': 'TrInfo', 'FS': 'State', 'U': 'CWUser'}])],

   705                      None, None, [self.system], {}, [])],

   706                    {'x': ueid})

   707 

   708     def test_complex_optional(self):

   709         ueid = self.session.user.eid

   710         self._test('Any U WHERE WF wf_info_for X, X eid %(x)s, WF owned_by U?, WF from_state FS',

   711                    [('OneFetchStep', [('Any U WHERE WF wf_info_for %s, WF owned_by U?, WF from_state FS' % ueid,

   712                                        [{'WF': 'TrInfo', 'FS': 'State', 'U': 'CWUser'}])],

   713                      None, None, [self.system], {}, [])],

   714                    {'x': ueid})

   715 

   716 

   717     def test_3sources_ambigous(self):

   718         self._test('Any X,T WHERE X owned_by U, U login "syt", X title T, X is IN(Bookmark, Card, EmailThread)',

   719                    [('FetchStep', [('Any X,T WHERE X title T, X is Card', [{'X': 'Card', 'T': 'String'}])],

   720                      [self.cards, self.system], None,

   721                      {'T': 'table0.C1', 'X': 'table0.C0', 'X.title': 'table0.C1'}, []),

   722                     ('FetchStep', [('Any U WHERE U login "syt", U is CWUser', [{'U': 'CWUser'}])],

   723                      [self.ldap, self.system], None,

   724                      {'U': 'table1.C0'}, []),

   725                     ('UnionStep', None, None, [

   726                         ('OneFetchStep', [('Any X,T WHERE X owned_by U, X title T, U is CWUser, X is IN(Bookmark, EmailThread)',

   727                                            [{'T': 'String', 'U': 'CWUser', 'X': 'Bookmark'},

   728                                             {'T': 'String', 'U': 'CWUser', 'X': 'EmailThread'}])],

   729                          None, None, [self.system], {'U': 'table1.C0'}, []),

   730                         ('OneFetchStep', [('Any X,T WHERE X owned_by U, X title T, U is CWUser, X is Card',

   731                                            [{'X': 'Card', 'U': 'CWUser', 'T': 'String'}])],

   732                          None, None, [self.system],

   733                          {'X': 'table0.C0', 'X.title': 'table0.C1', 'T': 'table0.C1', 'U': 'table1.C0'}, []),

   734                         ]),

   735                     ])

   736 

   737     def test_restricted_max(self):

   738         # dumb query to emulate the one generated by svnfile.entities.rql_revision_content

   739         self._test('Any V, MAX(VR) WHERE V is Card, V creation_date VR, '

   740                    '(V creation_date TODAY OR (V creation_date < TODAY AND NOT EXISTS('

   741                    'X is Card, X creation_date < TODAY, X creation_date >= VR)))',

   742                    [('FetchStep', [('Any VR WHERE X creation_date < TODAY, X creation_date VR, X is Card',

   743                                     [{'X': 'Card', 'VR': 'Datetime'}])],

   744                      [self.cards, self.system], None,

   745                      {'VR': 'table0.C0', 'X.creation_date': 'table0.C0'}, []),

   746                     ('FetchStep', [('Any V,VR WHERE V creation_date VR, V is Card',

   747                                     [{'VR': 'Datetime', 'V': 'Card'}])],

   748                      [self.cards, self.system], None,

   749                      {'VR': 'table1.C1', 'V': 'table1.C0', 'V.creation_date': 'table1.C1'}, []),

   750                     ('OneFetchStep', [('Any V,MAX(VR) WHERE V creation_date VR, (V creation_date TODAY) OR (V creation_date < TODAY, NOT EXISTS(X creation_date >= VR, X is Card)), V is Card',

   751                                        [{'X': 'Card', 'VR': 'Datetime', 'V': 'Card'}])],

   752                      None, None, [self.system],

   753                      {'VR': 'table1.C1', 'V': 'table1.C0', 'V.creation_date': 'table1.C1', 'X.creation_date': 'table0.C0'}, [])

   754                     ])

   755 

   756     def test_outer_supported_rel1(self):

   757         # both system and rql support all variables, can be

   758         self._test('Any X, R WHERE X is Note, X in_state S, X type R, '

   759                    'NOT EXISTS(Y is Note, Y in_state S, Y type R, X identity Y)',

   760                    [('OneFetchStep', [('Any X,R WHERE X is Note, X in_state S, X type R, NOT EXISTS(Y is Note, Y in_state S, Y type R, X identity Y), S is State',

   761                                        [{'Y': 'Note', 'X': 'Note', 'S': 'State', 'R': 'String'}])],

   762                      None, None,

   763                      [self.cards, self.system], {}, [])

   764                     ])

   765 

   766     def test_not_identity(self):

   767         ueid = self.session.user.eid

   768         self._test('Any X WHERE NOT X identity U, U eid %s, X is CWUser' % ueid,

   769                    [('OneFetchStep',

   770                      [('Any X WHERE NOT X identity %s, X is CWUser' % ueid, [{'X': 'CWUser'}])],

   771                      None, None,

   772                      [self.ldap, self.system], {}, [])

   773                     ])

   774 

   775     def test_outer_supported_rel2(self):

   776         self._test('Any X, MAX(R) GROUPBY X WHERE X in_state S, X login R, '

   777                    'NOT EXISTS(Y is Note, Y in_state S, Y type R)',

   778                    [('FetchStep', [('Any A,R WHERE Y in_state A, Y type R, A is State, Y is Note',

   779                                     [{'Y': 'Note', 'A': 'State', 'R': 'String'}])],

   780                      [self.cards, self.system], None,

   781                      {'A': 'table0.C0', 'R': 'table0.C1', 'Y.type': 'table0.C1'}, []),

   782                     ('FetchStep', [('Any X,R WHERE X login R, X is CWUser', [{'X': 'CWUser', 'R': 'String'}])],

   783                      [self.ldap, self.system], None,

   784                      {'X': 'table1.C0', 'X.login': 'table1.C1', 'R': 'table1.C1'}, []),

   785                     ('OneFetchStep', [('Any X,MAX(R) GROUPBY X WHERE X in_state S, X login R, NOT EXISTS(Y type R, S identity A, A is State, Y is Note), S is State, X is CWUser',

   786                                        [{'Y': 'Note', 'X': 'CWUser', 'S': 'State', 'R': 'String', 'A': 'State'}])],

   787                      None, None, [self.system],

   788                      {'A': 'table0.C0', 'X': 'table1.C0', 'X.login': 'table1.C1', 'R': 'table1.C1', 'Y.type': 'table0.C1'}, [])

   789                     ])

   790 

   791     def test_security_has_text(self):

   792         # use a guest user

   793         self.session = self.user_groups_session('guests')

   794         ueid = self.session.user.eid

   795         self._test('Any X WHERE X has_text "bla"',

   796                    [('FetchStep', [('Any E WHERE E type "X", E is Note', [{'E': 'Note'}])],

   797                      [self.cards, self.system], None, {'E': 'table0.C0'}, []),

   798                     ('UnionStep', None, None,

   799                      [('OneFetchStep',

   800                        [(u'Any X WHERE X has_text "bla", (EXISTS(X owned_by %(ueid)s)) OR ((((EXISTS(D concerne C?, C owned_by %(ueid)s, C type "X", X identity D, C is Division, D is Affaire)) OR (EXISTS(H concerne G?, G owned_by %(ueid)s, G type "X", X identity H, G is SubDivision, H is Affaire))) OR (EXISTS(I concerne F?, F owned_by %(ueid)s, F type "X", X identity I, F is Societe, I is Affaire))) OR (EXISTS(J concerne E?, E owned_by %(ueid)s, X identity J, E is Note, J is Affaire))), X is Affaire' % {'ueid': ueid},

   801                          [{'C': 'Division', 'E': 'Note', 'D': 'Affaire', 'G': 'SubDivision', 'F': 'Societe', 'I': 'Affaire', 'H': 'Affaire', 'J': 'Affaire', 'X': 'Affaire'}])],

   802                        None, None, [self.system], {'E': 'table0.C0'}, []),

   803                       ('OneFetchStep',

   804                        [('Any X WHERE X has_text "bla", EXISTS(X owned_by %s), X is Basket' % ueid,

   805                          [{'X': 'Basket'}]),

   806                         ('Any X WHERE X has_text "bla", EXISTS(X owned_by %s), X is CWUser' % ueid,

   807                          [{'X': 'CWUser'}]),

   808                         ('Any X WHERE X has_text "bla", X is IN(Card, Comment, Division, Email, EmailThread, File, Folder, Note, Personne, Societe, SubDivision, Tag)',

   809                          [{'X': 'Card'}, {'X': 'Comment'},

   810                           {'X': 'Division'}, {'X': 'Email'}, {'X': 'EmailThread'},

   811                           {'X': 'File'}, {'X': 'Folder'},

   812                           {'X': 'Note'}, {'X': 'Personne'}, {'X': 'Societe'},

   813                           {'X': 'SubDivision'}, {'X': 'Tag'}]),],

   814                        None, None, [self.system], {}, []),

   815                       ])

   816                      ])

   817 

   818     def test_security_has_text_limit_offset(self):

   819         # use a guest user

   820         self.session = self.user_groups_session('guests')

   821         ueid = self.session.user.eid

   822         # note: same as the above query but because of the subquery usage, the

   823         # display differs (not printing solutions for each union)

   824         self._test('Any X LIMIT 10 OFFSET 10 WHERE X has_text "bla"',

   825                    [('FetchStep', [('Any E WHERE E type "X", E is Note', [{'E': 'Note'}])],

   826                       [self.cards, self.system], None, {'E': 'table1.C0'}, []),

   827                      ('UnionFetchStep', [

   828                         ('FetchStep', [('Any X WHERE X has_text "bla", (EXISTS(X owned_by %(ueid)s)) OR ((((EXISTS(D concerne C?, C owned_by %(ueid)s, C type "X", X identity D, C is Division, D is Affaire)) OR (EXISTS(H concerne G?, G owned_by %(ueid)s, G type "X", X identity H, G is SubDivision, H is Affaire))) OR (EXISTS(I concerne F?, F owned_by %(ueid)s, F type "X", X identity I, F is Societe, I is Affaire))) OR (EXISTS(J concerne E?, E owned_by %(ueid)s, X identity J, E is Note, J is Affaire))), X is Affaire' % {'ueid': ueid},

   829                                             [{'C': 'Division', 'E': 'Note', 'D': 'Affaire', 'G': 'SubDivision', 'F': 'Societe', 'I': 'Affaire', 'H': 'Affaire', 'J': 'Affaire', 'X': 'Affaire'}])],

   830                           [self.system], {'E': 'table1.C0'}, {'X': 'table0.C0'}, []),

   831                          ('FetchStep',

   832                           [('Any X WHERE X has_text "bla", EXISTS(X owned_by %s), X is Basket' % ueid,

   833                             [{'X': 'Basket'}]),

   834                            ('Any X WHERE X has_text "bla", EXISTS(X owned_by %s), X is CWUser' % ueid,

   835                             [{'X': 'CWUser'}]),

   836                            ('Any X WHERE X has_text "bla", X is IN(Card, Comment, Division, Email, EmailThread, File, Folder, Note, Personne, Societe, SubDivision, Tag)',

   837                             [{'X': 'Card'}, {'X': 'Comment'},

   838                              {'X': 'Division'}, {'X': 'Email'}, {'X': 'EmailThread'},

   839                              {'X': 'File'}, {'X': 'Folder'},

   840                              {'X': 'Note'}, {'X': 'Personne'}, {'X': 'Societe'},

   841                              {'X': 'SubDivision'}, {'X': 'Tag'}])],

   842                           [self.system], {}, {'X': 'table0.C0'}, []),

   843                          ]),

   844                     ('OneFetchStep',

   845                      [('Any X LIMIT 10 OFFSET 10',

   846                        [{'X': 'Affaire'}, {'X': 'Basket'},

   847                         {'X': 'CWUser'}, {'X': 'Card'}, {'X': 'Comment'},

   848                         {'X': 'Division'}, {'X': 'Email'}, {'X': 'EmailThread'},

   849                         {'X': 'File'}, {'X': 'Folder'},

   850                         {'X': 'Note'}, {'X': 'Personne'}, {'X': 'Societe'},

   851                         {'X': 'SubDivision'}, {'X': 'Tag'}])],

   852                      10, 10, [self.system], {'X': 'table0.C0'}, [])

   853                      ])

   854 

   855     def test_security_user(self):

   856         """a guest user trying to see another user: EXISTS(X owned_by U) is automatically inserted"""

   857         # use a guest user

   858         self.session = self.user_groups_session('guests')

   859         ueid = self.session.user.eid

   860         self._test('Any X WHERE X login "bla"',

   861                    [('FetchStep',

   862                      [('Any X WHERE X login "bla", X is CWUser', [{'X': 'CWUser'}])],

   863                      [self.ldap, self.system], None, {'X': 'table0.C0'}, []),

   864                     ('OneFetchStep',

   865                      [('Any X WHERE EXISTS(X owned_by %s), X is CWUser' % ueid, [{'X': 'CWUser'}])],

   866                      None, None, [self.system], {'X': 'table0.C0'}, [])])

   867 

   868     def test_security_complex_has_text(self):

   869         # use a guest user

   870         self.session = self.user_groups_session('guests')

   871         ueid = self.session.user.eid

   872         self._test('Any X WHERE X has_text "bla", X firstname "bla"',

   873                    [('FetchStep', [('Any X WHERE X firstname "bla", X is CWUser', [{'X': 'CWUser'}])],

   874                      [self.ldap, self.system], None, {'X': 'table0.C0'}, []),

   875                     ('UnionStep', None, None, [

   876                         ('OneFetchStep', [('Any X WHERE X has_text "bla", EXISTS(X owned_by %s), X is CWUser' % ueid, [{'X': 'CWUser'}])],

   877                          None, None, [self.system], {'X': 'table0.C0'}, []),

   878                         ('OneFetchStep', [('Any X WHERE X has_text "bla", X firstname "bla", X is Personne', [{'X': 'Personne'}])],

   879                          None, None, [self.system], {}, []),

   880                         ]),

   881                     ])

   882 

   883     def test_security_complex_has_text_limit_offset(self):

   884         # use a guest user

   885         self.session = self.user_groups_session('guests')

   886         ueid = self.session.user.eid

   887         self._test('Any X LIMIT 10 OFFSET 10 WHERE X has_text "bla", X firstname "bla"',

   888                    [('FetchStep', [('Any X WHERE X firstname "bla", X is CWUser', [{'X': 'CWUser'}])],

   889                      [self.ldap, self.system], None, {'X': 'table1.C0'}, []),

   890                     ('UnionFetchStep', [

   891                         ('FetchStep', [('Any X WHERE X has_text "bla", EXISTS(X owned_by %s), X is CWUser' % ueid, [{'X': 'CWUser'}])],

   892                          [self.system], {'X': 'table1.C0'}, {'X': 'table0.C0'}, []),

   893                         ('FetchStep', [('Any X WHERE X has_text "bla", X firstname "bla", X is Personne', [{'X': 'Personne'}])],

   894                          [self.system], {}, {'X': 'table0.C0'}, []),

   895                         ]),

   896                      ('OneFetchStep',

   897                       [('Any X LIMIT 10 OFFSET 10', [{'X': 'CWUser'}, {'X': 'Personne'}])],

   898                       10, 10, [self.system], {'X': 'table0.C0'}, [])

   899                     ])

   900 

   901     def test_security_complex_aggregat(self):

   902         # use a guest user

   903         self.session = self.user_groups_session('guests')

   904         ueid = self.session.user.eid

   905         ALL_SOLS = X_ALL_SOLS[:]

   906         ALL_SOLS.remove({'X': 'CWSourceHostConfig'}) # not authorized

   907         ALL_SOLS.remove({'X': 'CWSourceSchemaConfig'}) # not authorized

   908         ALL_SOLS.remove({'X': 'CWDataImport'}) # not authorized

   909         self._test('Any MAX(X)',

   910                    [('FetchStep', [('Any E WHERE E type "X", E is Note', [{'E': 'Note'}])],

   911                      [self.cards, self.system],  None, {'E': 'table1.C0'}, []),

   912                     ('FetchStep', [('Any X WHERE X is CWUser', [{'X': 'CWUser'}])],

   913                      [self.ldap, self.system], None, {'X': 'table2.C0'}, []),

   914                     ('UnionFetchStep', [

   915                         ('FetchStep', [('Any X WHERE EXISTS(%s use_email X), X is EmailAddress' % ueid,

   916       [{'X': 'EmailAddress'}]),

   917                                        ('Any X WHERE EXISTS(X owned_by %s), X is Basket' % ueid, [{'X': 'Basket'}])],

   918                           [self.system], {}, {'X': 'table0.C0'}, []),

   919                         ('UnionFetchStep',

   920                          [('FetchStep', [('Any X WHERE X is IN(Card, Note, State)',

   921                                           [{'X': 'Card'}, {'X': 'Note'}, {'X': 'State'}])],

   922                            [self.cards, self.system], {}, {'X': 'table0.C0'}, []),

   923                           ('FetchStep',

   924                            [('Any X WHERE X is IN(BaseTransition, Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWSource, CWUniqueTogetherConstraint, Comment, Division, Email, EmailPart, EmailThread, ExternalUri, File, Folder, Old, Personne, RQLExpression, Societe, SubDivision, SubWorkflowExitPoint, Tag, TrInfo, Transition, Workflow, WorkflowTransition)',

   925                              [{'X': 'BaseTransition'}, {'X': 'Bookmark'},

   926                               {'X': 'CWAttribute'}, {'X': 'CWCache'},

   927                               {'X': 'CWConstraint'}, {'X': 'CWConstraintType'},

   928                               {'X': 'CWEType'}, {'X': 'CWGroup'},

   929                               {'X': 'CWPermission'}, {'X': 'CWProperty'},

   930                               {'X': 'CWRType'}, {'X': 'CWRelation'},

   931                               {'X': 'CWSource'},

   932                               {'X': 'CWUniqueTogetherConstraint'},

   933                               {'X': 'Comment'}, {'X': 'Division'},

   934                               {'X': 'Email'},

   935                               {'X': 'EmailPart'}, {'X': 'EmailThread'},

   936                               {'X': 'ExternalUri'}, {'X': 'File'},

   937                               {'X': 'Folder'}, {'X': 'Old'},

   938                               {'X': 'Personne'}, {'X': 'RQLExpression'},

   939                               {'X': 'Societe'}, {'X': 'SubDivision'},

   940                               {'X': 'SubWorkflowExitPoint'}, {'X': 'Tag'},

   941                               {'X': 'TrInfo'}, {'X': 'Transition'},

   942                               {'X': 'Workflow'}, {'X': 'WorkflowTransition'}])],

   943                            [self.system], {}, {'X': 'table0.C0'}, []),

   944                           ]),

   945                         ('FetchStep', [('Any X WHERE EXISTS(X owned_by %s), X is CWUser' % ueid, [{'X': 'CWUser'}])],

   946                          [self.system], {'X': 'table2.C0'}, {'X': 'table0.C0'}, []),

   947                         ('FetchStep', [('Any X WHERE (EXISTS(X owned_by %(ueid)s)) OR ((((EXISTS(D concerne C?, C owned_by %(ueid)s, C type "X", X identity D, C is Division, D is Affaire)) OR (EXISTS(H concerne G?, G owned_by %(ueid)s, G type "X", X identity H, G is SubDivision, H is Affaire))) OR (EXISTS(I concerne F?, F owned_by %(ueid)s, F type "X", X identity I, F is Societe, I is Affaire))) OR (EXISTS(J concerne E?, E owned_by %(ueid)s, X identity J, E is Note, J is Affaire))), X is Affaire' % {'ueid': ueid},

   948                                         [{'C': 'Division', 'E': 'Note', 'D': 'Affaire', 'G': 'SubDivision', 'F': 'Societe', 'I': 'Affaire', 'H': 'Affaire', 'J': 'Affaire', 'X': 'Affaire'}])],

   949                          [self.system], {'E': 'table1.C0'}, {'X': 'table0.C0'}, []),

   950                         ]),

   951                     ('OneFetchStep', [('Any MAX(X)', ALL_SOLS)],

   952                      None, None, [self.system], {'X': 'table0.C0'}, [])

   953                     ])

   954 

   955     def test_security_complex_aggregat2(self):

   956         # use a guest user

   957         self.session = self.user_groups_session('guests')

   958         ueid = self.session.user.eid

   959         X_ET_ALL_SOLS = []

   960         for s in X_ALL_SOLS:

   961             if s in ({'X': 'CWSourceHostConfig'}, {'X': 'CWSourceSchemaConfig'}, {'X': 'CWDataImport'}):

   962                 continue # not authorized

   963             ets = {'ET': 'CWEType'}

   964             ets.update(s)

   965             X_ET_ALL_SOLS.append(ets)

   966         self._test('Any ET, COUNT(X) GROUPBY ET ORDERBY ET WHERE X is ET',

   967                    [('FetchStep', [('Any X WHERE X is IN(Card, Note, State)',

   968                                     [{'X': 'Card'}, {'X': 'Note'}, {'X': 'State'}])],

   969                      [self.cards, self.system], None, {'X': 'table1.C0'}, []),

   970                     ('FetchStep', [('Any E WHERE E type "X", E is Note', [{'E': 'Note'}])],

   971                      [self.cards, self.system],  None, {'E': 'table2.C0'}, []),

   972                     ('FetchStep', [('Any X WHERE X is CWUser', [{'X': 'CWUser'}])],

   973                      [self.ldap, self.system], None, {'X': 'table3.C0'}, []),

   974                     ('UnionFetchStep',

   975                      [('FetchStep', [('Any ET,X WHERE X is ET, EXISTS(%s use_email X), ET is CWEType, X is EmailAddress' % ueid,

   976       [{'ET': 'CWEType', 'X': 'EmailAddress'}]), ('Any ET,X WHERE X is ET, EXISTS(X owned_by %s), ET is CWEType, X is Basket' % ueid,

   977                                       [{'ET': 'CWEType', 'X': 'Basket'}])],

   978                        [self.system], {}, {'ET': 'table0.C0', 'X': 'table0.C1'}, []),

   979                       ('FetchStep', [('Any ET,X WHERE X is ET, (EXISTS(X owned_by %(ueid)s)) OR ((((EXISTS(D concerne C?, C owned_by %(ueid)s, C type "X", X identity D, C is Division, D is Affaire)) OR (EXISTS(H concerne G?, G owned_by %(ueid)s, G type "X", X identity H, G is SubDivision, H is Affaire))) OR (EXISTS(I concerne F?, F owned_by %(ueid)s, F type "X", X identity I, F is Societe, I is Affaire))) OR (EXISTS(J concerne E?, E owned_by %(ueid)s, X identity J, E is Note, J is Affaire))), ET is CWEType, X is Affaire' % {'ueid': ueid},

   980                                       [{'C': 'Division', 'E': 'Note', 'D': 'Affaire',

   981                                         'G': 'SubDivision', 'F': 'Societe', 'I': 'Affaire',

   982                                         'H': 'Affaire', 'J': 'Affaire', 'X': 'Affaire',

   983                                         'ET': 'CWEType'}])],

   984                        [self.system], {'E': 'table2.C0'}, {'ET': 'table0.C0', 'X': 'table0.C1'},

   985                        []),

   986                       ('FetchStep', [('Any ET,X WHERE X is ET, EXISTS(X owned_by %s), ET is CWEType, X is CWUser' % ueid,

   987                                       [{'ET': 'CWEType', 'X': 'CWUser'}])],

   988                        [self.system], {'X': 'table3.C0'}, {'ET': 'table0.C0', 'X': 'table0.C1'}, []),

   989                       # extra UnionFetchStep could be avoided but has no cost, so don't care

   990                       ('UnionFetchStep',

   991                        [('FetchStep', [('Any ET,X WHERE X is ET, ET is CWEType, X is IN(BaseTransition, Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWSource, CWUniqueTogetherConstraint, Comment, Division, Email, EmailPart, EmailThread, ExternalUri, File, Folder, Old, Personne, RQLExpression, Societe, SubDivision, SubWorkflowExitPoint, Tag, TrInfo, Transition, Workflow, WorkflowTransition)',

   992                                         [{'X': 'BaseTransition', 'ET': 'CWEType'},

   993                                          {'X': 'Bookmark', 'ET': 'CWEType'}, {'X': 'CWAttribute', 'ET': 'CWEType'},

   994                                          {'X': 'CWCache', 'ET': 'CWEType'}, {'X': 'CWConstraint', 'ET': 'CWEType'},

   995                                          {'X': 'CWConstraintType', 'ET': 'CWEType'},

   996                                          {'X': 'CWEType', 'ET': 'CWEType'},

   997                                          {'X': 'CWGroup', 'ET': 'CWEType'}, {'X': 'CWPermission', 'ET': 'CWEType'},

   998                                          {'X': 'CWProperty', 'ET': 'CWEType'}, {'X': 'CWRType', 'ET': 'CWEType'},

   999                                          {'X': 'CWSource', 'ET': 'CWEType'},

  1000                                          {'X': 'CWRelation', 'ET': 'CWEType'},

  1001                                          {'X': 'CWUniqueTogetherConstraint', 'ET': 'CWEType'},

  1002                                          {'X': 'Comment', 'ET': 'CWEType'},

  1003                                          {'X': 'Division', 'ET': 'CWEType'}, {'X': 'Email', 'ET': 'CWEType'},

  1004                                          {'X': 'EmailPart', 'ET': 'CWEType'},

  1005                                          {'X': 'EmailThread', 'ET': 'CWEType'}, {'X': 'ExternalUri', 'ET': 'CWEType'},

  1006                                          {'X': 'File', 'ET': 'CWEType'}, {'X': 'Folder', 'ET': 'CWEType'},

  1007                                          {'X': 'Old', 'ET': 'CWEType'}, {'X': 'Personne', 'ET': 'CWEType'},

  1008                                          {'X': 'RQLExpression', 'ET': 'CWEType'}, {'X': 'Societe', 'ET': 'CWEType'},

  1009                                          {'X': 'SubDivision', 'ET': 'CWEType'}, {'X': 'SubWorkflowExitPoint', 'ET': 'CWEType'},

  1010                                          {'X': 'Tag', 'ET': 'CWEType'}, {'X': 'TrInfo', 'ET': 'CWEType'},

  1011                                          {'X': 'Transition', 'ET': 'CWEType'}, {'X': 'Workflow', 'ET': 'CWEType'},

  1012                                          {'X': 'WorkflowTransition', 'ET': 'CWEType'}])],

  1013                          [self.system], {}, {'ET': 'table0.C0', 'X': 'table0.C1'}, []),

  1014                         ('FetchStep',

  1015                          [('Any ET,X WHERE X is ET, ET is CWEType, X is IN(Card, Note, State)',

  1016                            [{'ET': 'CWEType', 'X': 'Card'},

  1017                             {'ET': 'CWEType', 'X': 'Note'},

  1018                             {'ET': 'CWEType', 'X': 'State'}])],

  1019                          [self.system], {'X': 'table1.C0'}, {'ET': 'table0.C0', 'X': 'table0.C1'}, []),

  1020                         ]),

  1021                     ]),

  1022                     ('OneFetchStep',

  1023                      [('Any ET,COUNT(X) GROUPBY ET ORDERBY ET', X_ET_ALL_SOLS)],

  1024                      None, None, [self.system], {'ET': 'table0.C0', 'X': 'table0.C1'}, [])

  1025                     ])

  1026 

  1027     def test_security_3sources(self):

  1028         # use a guest user

  1029         self.session = self.user_groups_session('guests')

  1030         ueid = self.session.user.eid

  1031         self._test('Any X, XT WHERE X is Card, X owned_by U, X title XT, U login "syt"',

  1032                    [('FetchStep',

  1033                      [('Any X,XT WHERE X title XT, X is Card', [{'X': 'Card', 'XT': 'String'}])],

  1034                      [self.cards, self.system], None, {'X': 'table0.C0', 'X.title': 'table0.C1', 'XT': 'table0.C1'}, []),

  1035                     ('FetchStep',

  1036                      [('Any U WHERE U login "syt", U is CWUser', [{'U': 'CWUser'}])],

  1037                      [self.ldap, self.system], None, {'U': 'table1.C0'}, []),

  1038                     ('OneFetchStep',

  1039                      [('Any X,XT WHERE X owned_by U, X title XT, EXISTS(U owned_by %s), U is CWUser, X is Card' % ueid,

  1040                        [{'X': 'Card', 'U': 'CWUser', 'XT': 'String'}])],

  1041                      None, None, [self.system],

  1042                      {'X': 'table0.C0', 'X.title': 'table0.C1', 'XT': 'table0.C1', 'U': 'table1.C0'}, [])

  1043                     ])

  1044 

  1045     def test_security_3sources_identity(self):

  1046         self.restore_orig_cwuser_security()

  1047         # use a guest user

  1048         self.session = self.user_groups_session('guests')

  1049         ueid = self.session.user.eid

  1050         self._test('Any X, XT WHERE X is Card, X owned_by U, X title XT, U login "syt"',

  1051                    [('FetchStep',

  1052                      [('Any X,XT WHERE X title XT, X is Card', [{'X': 'Card', 'XT': 'String'}])],

  1053                      [self.cards, self.system], None, {'X': 'table0.C0', 'X.title': 'table0.C1', 'XT': 'table0.C1'}, []),

  1054                     ('OneFetchStep',

  1055                      [('Any X,XT WHERE X owned_by U, X title XT, U login "syt", EXISTS(U identity %s), U is CWUser, X is Card' % ueid,

  1056                        [{'U': 'CWUser', 'X': 'Card', 'XT': 'String'}])],

  1057                      None, None, [self.system], {'X': 'table0.C0', 'X.title': 'table0.C1', 'XT': 'table0.C1'}, [])

  1058                     ])

  1059 

  1060     def test_security_3sources_identity_optional_var(self):

  1061         self.restore_orig_cwuser_security()

  1062         # use a guest user

  1063         self.session = self.user_groups_session('guests')

  1064         ueid = self.session.user.eid

  1065         self._test('Any X,XT,U WHERE X is Card, X owned_by U?, X title XT, U login L',

  1066                    [('FetchStep',

  1067                      [('Any U,L WHERE U login L, EXISTS(U identity %s), U is CWUser' % ueid,

  1068                        [{'L': 'String', u'U': 'CWUser'}])],

  1069                      [self.system], {}, {'L': 'table0.C1', 'U': 'table0.C0', 'U.login': 'table0.C1'}, []),

  1070                     ('FetchStep',

  1071                      [('Any X,XT WHERE X title XT, X is Card', [{'X': 'Card', 'XT': 'String'}])],

  1072                      [self.cards, self.system], None, {'X': 'table1.C0', 'X.title': 'table1.C1', 'XT': 'table1.C1'}, []),

  1073                     ('OneFetchStep',

  1074                      [('Any X,XT,U WHERE X owned_by U?, X title XT, X is Card',

  1075                        [{'X': 'Card', 'U': 'CWUser', 'XT': 'String'}])],

  1076                      None, None, [self.system], {'L': 'table0.C1',

  1077                                                  'U': 'table0.C0',

  1078                                                  'X': 'table1.C0',

  1079                                                  'X.title': 'table1.C1',

  1080                                                  'XT': 'table1.C1'}, [])

  1081                     ])

  1082 

  1083     def test_security_3sources_limit_offset(self):

  1084         # use a guest user

  1085         self.session = self.user_groups_session('guests')

  1086         ueid = self.session.user.eid

  1087         self._test('Any X, XT LIMIT 10 OFFSET 10 WHERE X is Card, X owned_by U, X title XT, U login "syt"',

  1088                    [('FetchStep',

  1089                      [('Any X,XT WHERE X title XT, X is Card', [{'X': 'Card', 'XT': 'String'}])],

  1090                      [self.cards, self.system], None, {'X': 'table0.C0', 'X.title': 'table0.C1', 'XT': 'table0.C1'}, []),

  1091                     ('FetchStep',

  1092                      [('Any U WHERE U login "syt", U is CWUser', [{'U': 'CWUser'}])],

  1093                      [self.ldap, self.system], None, {'U': 'table1.C0'}, []),

  1094                     ('OneFetchStep',

  1095                      [('Any X,XT LIMIT 10 OFFSET 10 WHERE X owned_by U, X title XT, EXISTS(U owned_by %s), U is CWUser, X is Card' % ueid,

  1096                        [{'X': 'Card', 'U': 'CWUser', 'XT': 'String'}])],

  1097                      10, 10, [self.system],

  1098                      {'X': 'table0.C0', 'X.title': 'table0.C1', 'XT': 'table0.C1', 'U': 'table1.C0'}, [])

  1099                     ])

  1100 

  1101     def test_exists_base(self):

  1102         self._test('Any X,L,S WHERE X in_state S, X login L, EXISTS(X in_group G, G name "bougloup")',

  1103                    [('FetchStep', [('Any X,L WHERE X login L, X is CWUser', [{'X': 'CWUser', 'L': 'String'}])],

  1104                      [self.ldap, self.system], None, {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, []),

  1105                     ('OneFetchStep', [("Any X,L,S WHERE X in_state S, X login L, "

  1106                                       'EXISTS(X in_group G, G name "bougloup", G is CWGroup), S is State, X is CWUser',

  1107                                        [{'X': 'CWUser', 'L': 'String', 'S': 'State', 'G': 'CWGroup'}])],

  1108                      None, None, [self.system],

  1109                      {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, [])])

  1110 

  1111     def test_exists_complex(self):

  1112         self._test('Any G WHERE X in_group G, G name "managers", EXISTS(X copain T, T login in ("comme", "cochon"))',

  1113                    [('FetchStep', [('Any T WHERE T login IN("comme", "cochon"), T is CWUser', [{'T': 'CWUser'}])],

  1114                      [self.ldap, self.system], None, {'T': 'table0.C0'}, []),

  1115                     ('OneFetchStep',

  1116                      [('Any G WHERE X in_group G, G name "managers", EXISTS(X copain T, T is CWUser), G is CWGroup, X is CWUser',

  1117                        [{'X': 'CWUser', 'T': 'CWUser', 'G': 'CWGroup'}])],

  1118                      None, None, [self.system], {'T': 'table0.C0'}, [])])

  1119 

  1120     def test_exists3(self):

  1121         self._test('Any G,L WHERE X in_group G, X login L, G name "managers", EXISTS(X copain T, T login in ("comme", "cochon"))',

  1122                    [('FetchStep',

  1123                      [('Any T WHERE T login IN("comme", "cochon"), T is CWUser',

  1124                        [{'T': 'CWUser'}])],

  1125                      [self.ldap, self.system], None, {'T': 'table0.C0'}, []),

  1126                     ('FetchStep',

  1127                      [('Any L,X WHERE X login L, X is CWUser', [{'X': 'CWUser', 'L': 'String'}])],

  1128                      [self.ldap, self.system], None,

  1129                      {'X': 'table1.C1', 'X.login': 'table1.C0', 'L': 'table1.C0'}, []),

  1130                     ('OneFetchStep',

  1131                      [('Any G,L WHERE X in_group G, X login L, G name "managers", EXISTS(X copain T, T is CWUser), G is CWGroup, X is CWUser',

  1132                        [{'G': 'CWGroup', 'L': 'String', 'T': 'CWUser', 'X': 'CWUser'}])],

  1133                      None, None,

  1134                      [self.system], {'T': 'table0.C0', 'X': 'table1.C1', 'X.login': 'table1.C0', 'L': 'table1.C0'}, [])])

  1135