1 from logilab.common.decorators import clear_cache

     2 from cubicweb.devtools import init_test_database

     3 from cubicweb.devtools.repotest import BasePlannerTC, do_monkey_patch, undo_monkey_patch, test_plan

     4 

     5 class _SetGenerator(object):

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

     7     """

     8     def __getitem__(self, key):

     9         try:

    10             it = iter(key)

    11         except TypeError:

    12             it = (key,)

    13         return set(it)

    14 s = _SetGenerator()

    15 

    16 from cubicweb.schema import ERQLExpression

    17 from cubicweb.server.sources import AbstractSource

    18 from cubicweb.server.msplanner import MSPlanner, PartPlanInformation

    19 

    20 class FakeUserROSource(AbstractSource):

    21     uri = 'zzz'

    22     support_entities = {'EUser': False}

    23     support_relations = {}

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

    25         return []

    26 

    27         

    28 class FakeCardSource(AbstractSource):

    29     uri = 'ccc'

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

    31     support_relations = {'in_state': True, 'multisource_rel': True}

    32     dont_cross_relations = set(('fiche',))

    33     

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

    35         return []

    36 

    37 X_ALL_SOLS = sorted([{'X': 'Affaire'}, {'X': 'Basket'}, {'X': 'Bookmark'},

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

    39                      {'X': 'ECache'}, {'X': 'EConstraint'}, {'X': 'EConstraintType'},

    40                      {'X': 'EEType'}, {'X': 'EFRDef'}, {'X': 'EGroup'},

    41                      {'X': 'ENFRDef'}, {'X': 'EPermission'}, {'X': 'EProperty'},

    42                      {'X': 'ERType'}, {'X': 'EUser'}, {'X': 'Email'},

    43                      {'X': 'EmailAddress'}, {'X': 'EmailPart'}, {'X': 'EmailThread'},

    44                      {'X': 'File'}, {'X': 'Folder'}, {'X': 'Image'},

    45                      {'X': 'Note'}, {'X': 'Personne'}, {'X': 'RQLExpression'},

    46                      {'X': 'Societe'}, {'X': 'State'}, {'X': 'SubDivision'},

    47                      {'X': 'Tag'}, {'X': 'TrInfo'}, {'X': 'Transition'}])

    48 

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

    50 repo, cnx = init_test_database('sqlite')

    51 

    52 class BaseMSPlannerTC(BasePlannerTC):

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

    54     

    55     * system source supporting everything

    56     * ldap source supporting EUser

    57     * rql source supporting Card

    58     """

    59     repo = repo

    60     

    61     def setUp(self):

    62         #_QuerierTC.setUp(self)

    63         self.o = repo.querier

    64         self.session = repo._sessions.values()[0]

    65         self.pool = self.session.set_pool()

    66         self.schema = self.o.schema

    67         # hijack Affaire security

    68         affreadperms = list(self.schema['Affaire']._groups['read'])

    69         self.prevrqlexpr_affaire = affreadperms[-1]

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

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

    72         self.schema['Affaire']._groups['read'] = tuple(affreadperms)

    73         # hijack EUser security

    74         userreadperms = list(self.schema['EUser']._groups['read'])

    75         self.prevrqlexpr_user = userreadperms[-1]

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

    77         self.schema['EUser']._groups['read'] = tuple(userreadperms)

    78         

    79         self.sources = self.o._repo.sources

    80         self.system = self.sources[-1]

    81         self.sources.append(FakeUserROSource(self.o._repo, self.o.schema,

    82                                              {'uri': 'ldapuser'}))

    83         repo.sources_by_uri['ldapuser'] = self.sources[-1]

    84         self.ldap = self.sources[-1]

    85         self.sources.append(FakeCardSource(self.o._repo, self.o.schema,

    86                                            {'uri': 'cards'}))

    87         repo.sources_by_uri['cards'] = self.sources[-1]

    88         self.rql = self.sources[-1]

    89         do_monkey_patch()

    90         

    91     def tearDown(self):

    92         undo_monkey_patch()

    93         del self.sources[-1]

    94         del self.sources[-1]

    95         del repo.sources_by_uri['ldapuser']

    96         del repo.sources_by_uri['cards']

    97         # restore hijacked security

    98         self.restore_orig_affaire_security()

    99         self.restore_orig_euser_security()

   100         

   101     def restore_orig_affaire_security(self):

   102         affreadperms = list(self.schema['Affaire']._groups['read'])

   103         affreadperms[-1] = self.prevrqlexpr_affaire

   104         self.schema['Affaire']._groups['read'] = tuple(affreadperms)

   105         clear_cache(self.schema['Affaire'], 'ERSchema_get_rqlexprs')

   106         

   107     def restore_orig_euser_security(self):

   108         userreadperms = list(self.schema['EUser']._groups['read'])

   109         userreadperms[-1] = self.prevrqlexpr_user

   110         self.schema['EUser']._groups['read'] = tuple(userreadperms)

   111         clear_cache(self.schema['EUser'], 'ERSchema_get_rqlexprs')

   112 

   113                   

   114 class PartPlanInformationTC(BaseMSPlannerTC):

   115 

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

   117         if len(args) == 3:

   118             kwargs, sourcesvars, needsplit = args

   119         else:

   120             sourcesvars, needsplit = args

   121             kwargs = None

   122         plan = self._prepare_plan(rql, kwargs)

   123         union = plan.rqlst

   124         plan.preprocess(union)

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

   126         for sourcevars in ppi._sourcesvars.itervalues():

   127             for var in sourcevars.keys():

   128                 solindices = sourcevars.pop(var)

   129                 sourcevars[var._ms_table_key()] = solindices

   130         self.assertEquals(ppi._sourcesvars, sourcesvars)

   131         self.assertEquals(ppi.needsplit, needsplit)

   132 

   133         

   134     def test_simple_system_only(self):

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

   136         self._test('EGroup X',

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

   138         

   139     def test_simple_system_ldap(self):

   140         """retrieve EUser X from both sources and return concatenation of results

   141         """

   142         self._test('EUser X',

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

   144         

   145     def test_simple_system_rql(self):

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

   147         """

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

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

   150         

   151     def test_simple_eid_specified(self):

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

   153         """

   154         ueid = self.session.user.eid

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

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

   157         

   158     def test_simple_eid_invariant(self):

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

   160         """

   161         ueid = self.session.user.eid

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

   163                    {}, False)

   164         

   165     def test_simple_invariant(self):

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

   167         """

   168         self._test('Any X WHERE X is EUser, X in_group G, G name "users"',

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

   170         

   171     def test_security_has_text(self):

   172         """retrieve EUser X from system source only (has_text not supported by ldap source)

   173         """

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

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

   176         self._test('EUser X WHERE X has_text "bla"',

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

   178         

   179     def test_complex_base(self):

   180         """

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

   182            concatenation of results into a temporary table

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

   184            G name 'users' on the system source

   185         """

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

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

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

   189 

   190     def test_complex_invariant_ordered(self):

   191         """

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

   193            concatenation of results into a temporary table

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

   195            on the system source   

   196         """

   197         ueid = self.session.user.eid

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

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

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

   201 

   202     def test_complex_invariant(self):

   203         """

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

   205            concatenation of results into a temporary table

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

   207            on the system source   

   208         """

   209         ueid = self.session.user.eid

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

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

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

   213 

   214     def test_complex_ambigous(self):

   215         """retrieve EUser X from system and ldap sources, Person X from system source only

   216         """

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

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

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

   220 

   221     def test_complex_multiple(self):

   222         """

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

   224            cartesian product of results into a temporary table

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

   226            on the system source   

   227         """

   228         ueid = self.session.user.eid

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

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

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

   232         

   233     def test_complex_aggregat(self):

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

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

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

   237                    

   238     def test_complex_optional(self):

   239         ueid = self.session.user.eid

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

   241                    {self.system: {'WF': s[0], 'FS': s[0], 'U': s[0], 'from_state': s[0], 'owned_by': s[0], 'wf_info_for': s[0]}}, False)

   242 

   243     def test_exists4(self):

   244         """

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

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

   247         {self.rql : 'S'} must be removed

   248         """

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

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

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

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

   253                     self.ldap: {'X': s[0], 'T2': s[0], 'T': s[0]}}, True)

   254 

   255     def test_relation_need_split(self):

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

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

   258                      self.rql: {'X': s[2], 'S': s[2]}}, True)

   259 

   260     def test_relation_restriction_ambigous_need_split(self):

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

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

   263                     self.rql: {'X': s[2], 'S': s[2]}}, True)

   264 

   265     def test_simplified_var(self):

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

   267         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',

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

   269                    {self.system: {'P': s[0], 'G': s[0], 'X': s[0], 'require_permission': s[0], 'in_group': s[0], 'P': s[0], 'require_group': s[0]}}, False)

   270         

   271     def test_delete_relation1(self):

   272         ueid = self.session.user.eid

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

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

   275                    {self.system: {'Y': s[0], 'created_by': s[0]}}, False)

   276                    

   277 

   278         

   279 class MSPlannerTC(BaseMSPlannerTC):

   280     

   281     def setUp(self):

   282         BaseMSPlannerTC.setUp(self)

   283         self.planner = MSPlanner(self.o.schema, self.o._rqlhelper)

   284 

   285     _test = test_plan

   286 

   287     def test_simple_system_only(self):

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

   289         """

   290         self._test('EGroup X',

   291                    [('OneFetchStep', [('Any X WHERE X is EGroup', [{'X': 'EGroup'}])],

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

   293 

   294     def test_simple_system_only_limit(self):

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

   296         """

   297         self._test('EGroup X LIMIT 10',

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

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

   300 

   301     def test_simple_system_only_limit_offset(self):

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

   303         """

   304         self._test('EGroup X LIMIT 10 OFFSET 10',

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

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

   307         

   308     def test_simple_system_ldap(self):

   309         """retrieve EUser X from both sources and return concatenation of results

   310         """

   311         self._test('EUser X',

   312                    [('OneFetchStep', [('Any X WHERE X is EUser', [{'X': 'EUser'}])],

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

   314         

   315     def test_simple_system_ldap_limit(self):

   316         """retrieve EUser X from both sources and return concatenation of results

   317         """

   318         self._test('EUser X LIMIT 10',

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

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

   321 

   322     def test_simple_system_ldap_limit_offset(self):

   323         """retrieve EUser X from both sources and return concatenation of results

   324         """

   325         self._test('EUser X LIMIT 10 OFFSET 10',

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

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

   328 

   329     def test_simple_system_ldap_ordered_limit_offset(self):

   330         """retrieve EUser X from both sources and return concatenation of results

   331         """

   332         self._test('EUser X ORDERBY X LIMIT 10 OFFSET 10',

   333                    [('AggrStep', 'Any X ORDERBY X', 10, 10, 'table0', None, [

   334                        ('FetchStep', [('Any X WHERE X is EUser', [{'X': 'EUser'}])],

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

   336                        ]),

   337                    ])

   338     def test_simple_system_ldap_aggregat(self):

   339         """retrieve EUser X from both sources and return concatenation of results

   340         """

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

   342         self._test('Any COUNT(X) WHERE X is EUser',

   343                    [('AggrStep', 'Any COUNT(X)', None, None, 'table0', None, [

   344                        ('FetchStep', [('Any COUNT(X) WHERE X is EUser', [{'X': 'EUser'}])],

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

   346                        ]),

   347                    ])

   348         

   349     def test_simple_system_rql(self):

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

   351         """

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

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

   354                      None, None, [self.rql, self.system], {}, [])])

   355         

   356     def test_simple_eid_specified(self):

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

   358         """

   359         ueid = self.session.user.eid

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

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

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

   363                    {'x': ueid})

   364         

   365     def test_simple_eid_invariant(self):

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

   367         """

   368         ueid = self.session.user.eid

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

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

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

   372                    {'x': ueid})

   373         

   374     def test_simple_invariant(self):

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

   376         """

   377         self._test('Any X WHERE X is EUser, X in_group G, G name "users"',

   378                    [('OneFetchStep', [('Any X WHERE X is EUser, X in_group G, G name "users"',

   379                                        [{'X': 'EUser', 'G': 'EGroup'}])],

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

   381         

   382     def test_complex_base(self):

   383         """

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

   385            concatenation of results into a temporary table

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

   387            G name 'users' on the system source

   388         """

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

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

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

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

   393                     ('OneFetchStep', [('Any X,L WHERE X in_group G, X login L, G name "users", G is EGroup, X is EUser',

   394                                        [{'X': 'EUser', 'L': 'String', 'G': 'EGroup'}])],

   395                      None, None, [self.system],

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

   397                     ])

   398 

   399     def test_complex_base_limit_offset(self):

   400         """

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

   402            concatenation of results into a temporary table

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

   404            G name 'users' on the system source

   405         """

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

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

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

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

   410                     ('OneFetchStep', [('Any X,L LIMIT 10 OFFSET 10 WHERE X in_group G, X login L, G name "users", G is EGroup, X is EUser',

   411                                        [{'X': 'EUser', 'L': 'String', 'G': 'EGroup'}])],

   412                      10, 10,

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

   414                     ])

   415 

   416     def test_complex_ordered(self):

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

   418                    [('AggrStep', 'Any L ORDERBY L', None, None, 'table0', None, 

   419                      [('FetchStep', [('Any L WHERE X login L, X is EUser',

   420                                       [{'X': 'EUser', 'L': 'String'}])],

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

   422                       ])

   423                     ])

   424 

   425     def test_complex_ordered_limit_offset(self):

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

   427                    [('AggrStep', 'Any L ORDERBY L', 10, 10, 'table0', None, 

   428                      [('FetchStep', [('Any L WHERE X login L, X is EUser',

   429                                       [{'X': 'EUser', 'L': 'String'}])],

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

   431                       ])

   432                     ])

   433         

   434     def test_complex_invariant_ordered(self):

   435         """

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

   437            concatenation of results into a temporary table

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

   439            on the system source

   440 

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

   442         """

   443         ueid = self.session.user.eid

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

   445                    [('FetchStep',

   446                      [('Any X,AA WHERE X modification_date AA, X is EUser',

   447                        [{'AA': 'Datetime', 'X': 'EUser'}])],

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

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

   450                     ('OneFetchStep',

   451                      [('Any X,AA ORDERBY AA WHERE 5 owned_by X, X modification_date AA, X is EUser',

   452                        [{'AA': 'Datetime', 'X': 'EUser'}])],

   453                      None, None, [self.system],

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

   455                     ],

   456                    {'x': ueid})

   457 

   458     def test_complex_invariant(self):

   459         """

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

   461            concatenation of results into a temporary table

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

   463            on the system source   

   464         """

   465         ueid = self.session.user.eid

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

   467                    [('FetchStep', [('Any X,L,AA WHERE X login L, X modification_date AA, X is EUser',

   468                                     [{'AA': 'Datetime', 'X': 'EUser', 'L': 'String'}])],

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

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

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

   472                                        [{'AA': 'Datetime', 'X': 'EUser', 'L': 'String'}])],

   473                      None, None, [self.system],

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

   475                    {'x': ueid})

   476 

   477     def test_complex_ambigous(self):

   478         """retrieve EUser X from system and ldap sources, Person X from system source only

   479         """

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

   481                    [('UnionStep', None, None, [

   482                        ('OneFetchStep', [('Any X,F WHERE X firstname F, X is EUser',

   483                                           [{'X': 'EUser', 'F': 'String'}])],

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

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

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

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

   488                        ]),

   489                     ])

   490 

   491     def test_complex_ambigous_limit_offset(self):

   492         """retrieve EUser X from system and ldap sources, Person X from system source only

   493         """

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

   495                    [('UnionStep', 10, 10, [

   496                        ('OneFetchStep', [('Any X,F WHERE X firstname F, X is EUser',

   497                                           [{'X': 'EUser', 'F': 'String'}])],

   498                         None, None,

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

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

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

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

   503                        ]),

   504                     ])

   505 

   506     def test_complex_ambigous_ordered(self):

   507         """

   508         1. retrieve EUser X from system and ldap sources, Person X from system source only, store

   509            each result in the same temp table

   510         2. return content of the table sorted

   511         """

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

   513                    [('AggrStep', 'Any X,F ORDERBY F', None, None, 'table0', None, 

   514                      [('FetchStep', [('Any X,F WHERE X firstname F, X is EUser',

   515                                       [{'X': 'EUser', 'F': 'String'}])],

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

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

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

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

   520                        [self.system], {},

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

   522                       ]),

   523                     ])

   524         

   525     def test_complex_multiple(self):

   526         """

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

   528            cartesian product of results into a temporary table

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

   530            on the system source   

   531         """

   532         ueid = self.session.user.eid

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

   534                    [('FetchStep',

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

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

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

   538                     ('FetchStep',

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

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

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

   542                     ('OneFetchStep',

   543                      [('Any X,Y WHERE X is EUser, Y is EUser', [{'X': 'EUser', 'Y': 'EUser'}])],

   544                      None, None, [self.system],

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

   546                     ], {'x': ueid})

   547         

   548     def test_complex_multiple_limit_offset(self):

   549         """

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

   551            cartesian product of results into a temporary table

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

   553            on the system source   

   554         """

   555         ueid = self.session.user.eid

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

   557                    [('FetchStep',

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

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

   560                     ('FetchStep',

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

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

   563                     ('OneFetchStep',

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

   565                      10, 10, [self.system],

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

   567                     ], {'x': ueid})

   568         

   569     def test_complex_aggregat(self):

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

   571                    [('OneFetchStep',

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

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

   574                     ])

   575         

   576     def test_complex_typed_aggregat(self):

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

   578                    [('AggrStep', 'Any MAX(X)', None, None, 'table0',  None,

   579                      [('FetchStep',

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

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

   582                       ])

   583                     ])

   584         

   585     def test_complex_greater_eid(self):

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

   587                    [('OneFetchStep',

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

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

   590                     ])

   591         

   592     def test_complex_greater_typed_eid(self):

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

   594                    [('OneFetchStep',

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

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

   597                     ])

   598         

   599     def test_complex_optional(self):

   600         ueid = self.session.user.eid

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

   602                    [('OneFetchStep', [('Any U WHERE WF wf_info_for 5, WF owned_by U?, WF from_state FS',

   603                                        [{'WF': 'TrInfo', 'FS': 'State', 'U': 'EUser'}])],

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

   605                    {'x': ueid})

   606 

   607     def test_complex_optional(self):

   608         ueid = self.session.user.eid

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

   610                    [('OneFetchStep', [('Any U WHERE WF wf_info_for 5, WF owned_by U?, WF from_state FS',

   611                                        [{'WF': 'TrInfo', 'FS': 'State', 'U': 'EUser'}])],

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

   613                    {'x': ueid})

   614 

   615     

   616     def test_3sources_ambigous(self):

   617         self._test('Any X,T WHERE X owned_by U, U login "syt", X title T',

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

   619                      [self.rql, self.system], None,

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

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

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

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

   624                     ('UnionStep', None, None, [

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

   626                                            [{'T': 'String', 'U': 'EUser', 'X': 'Bookmark'},

   627                                             {'T': 'String', 'U': 'EUser', 'X': 'EmailThread'}])],

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

   629                         ('OneFetchStep', [('Any X,T WHERE X owned_by U, X title T, U is EUser, X is Card',

   630                                            [{'X': 'Card', 'U': 'EUser', 'T': 'String'}])],

   631                          None, None, [self.system],

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

   633                         ]),

   634                     ])

   635 

   636     def test_restricted_max(self):

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

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

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

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

   641                    [('FetchStep', [('Any VR WHERE X creation_date < TODAY, X creation_date >= VR, X is Card',

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

   643                      [self.rql, self.system], None,

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

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

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

   647                      [self.rql, self.system], None,

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

   649                     ('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',

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

   651                      None, None, [self.system],

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

   653                     ])

   654 

   655     def test_outer_supported_rel1(self):

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

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

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

   659                    [('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',

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

   661                      None, None,

   662                      [self.rql, self.system], {}, [])

   663                     ])

   664 

   665     def test_not_identity(self):

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

   667         self._test('Any X WHERE NOT X identity U, U eid %s' % self.session.user.eid,

   668                    [('OneFetchStep',

   669                      [('Any X WHERE NOT X identity 5, X is EUser', [{'X': 'EUser'}])],

   670                      None, None,

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

   672                     ])

   673 

   674     def test_outer_supported_rel2(self):

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

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

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

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

   679                      [self.rql, self.system], None,

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

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

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

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

   684                     ('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 EUser',

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

   686                      None, None, [self.system],

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

   688                     ])

   689             

   690     def test_security_has_text(self):

   691         # use a guest user

   692         self.session = self._user_session()[1]

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

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

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

   696                     ('UnionStep', None, None,

   697                      [('OneFetchStep',

   698                        [(u'Any X WHERE X has_text "bla", (EXISTS(X owned_by 5)) OR ((((EXISTS(D concerne C?, C owned_by 5, C type "X", X identity D, C is Division, D is Affaire)) OR (EXISTS(H concerne G?, G owned_by 5, G type "X", X identity H, G is SubDivision, H is Affaire))) OR (EXISTS(I concerne F?, F owned_by 5, F type "X", X identity I, F is Societe, I is Affaire))) OR (EXISTS(J concerne E?, E owned_by 5, X identity J, E is Note, J is Affaire))), X is Affaire',

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

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

   701                       ('OneFetchStep',

   702                        [('Any X WHERE X has_text "bla", EXISTS(X owned_by 5), X is Basket',

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

   704                         ('Any X WHERE X has_text "bla", EXISTS(X owned_by 5), X is EUser',

   705                          [{'X': 'EUser'}]),

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

   707                          [{'X': 'Card'}, {'X': 'Comment'}, {'X': 'Division'},

   708                           {'X': 'Email'}, {'X': 'EmailThread'}, {'X': 'File'},

   709                           {'X': 'Folder'}, {'X': 'Image'}, {'X': 'Note'},

   710                           {'X': 'Personne'}, {'X': 'Societe'}, {'X': 'State'},

   711                           {'X': 'SubDivision'}, {'X': 'Tag'}, {'X': 'Transition'}]),],

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

   713                       ])

   714                      ])

   715         

   716     def test_security_has_text_limit_offset(self):

   717         # use a guest user

   718         self.session = self._user_session()[1]

   719         # note: same as the above query but because of the subquery usage, the display differs (not printing solutions for each union)

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

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

   722                       [self.rql, self.system], None, {'E': 'table1.C0'}, []),

   723                      ('UnionFetchStep', [

   724                          ('FetchStep', [('Any X WHERE X has_text "bla", (EXISTS(X owned_by 5)) OR ((((EXISTS(D concerne C?, C owned_by 5, C type "X", X identity D, C is Division, D is Affaire)) OR (EXISTS(H concerne G?, G owned_by 5, G type "X", X identity H, G is SubDivision, H is Affaire))) OR (EXISTS(I concerne F?, F owned_by 5, F type "X", X identity I, F is Societe, I is Affaire))) OR (EXISTS(J concerne E?, E owned_by 5, X identity J, E is Note, J is Affaire))), X is Affaire',

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

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

   727                          ('FetchStep',

   728                           [('Any X WHERE X has_text "bla", EXISTS(X owned_by 5), X is Basket',

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

   730                         ('Any X WHERE X has_text "bla", EXISTS(X owned_by 5), X is EUser',

   731                          [{'X': 'EUser'}]),

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

   733                          [{'X': 'Card'}, {'X': 'Comment'}, {'X': 'Division'},

   734                           {'X': 'Email'}, {'X': 'EmailThread'}, {'X': 'File'},

   735                           {'X': 'Folder'}, {'X': 'Image'}, {'X': 'Note'},

   736                           {'X': 'Personne'}, {'X': 'Societe'}, {'X': 'State'},

   737                           {'X': 'SubDivision'}, {'X': 'Tag'}, {'X': 'Transition'}]),],

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

   739                          ]),

   740                     ('OneFetchStep',

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

   742                        [{'X': 'Affaire'}, {'X': 'Basket'}, {'X': 'Card'},

   743                         {'X': 'Comment'}, {'X': 'Division'}, {'X': 'EUser'},

   744                         {'X': 'Email'}, {'X': 'EmailThread'}, {'X': 'File'},

   745                         {'X': 'Folder'}, {'X': 'Image'}, {'X': 'Note'},

   746                         {'X': 'Personne'}, {'X': 'Societe'}, {'X': 'State'},

   747                         {'X': 'SubDivision'}, {'X': 'Tag'}, {'X': 'Transition'}])],

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

   749                      ])

   750         

   751     def test_security_user(self):

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

   753         # use a guest user

   754         self.session = self._user_session()[1]

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

   756                    [('FetchStep',

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

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

   759                     ('OneFetchStep',

   760                      [('Any X WHERE EXISTS(X owned_by 5), X is EUser', [{'X': 'EUser'}])],

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

   762                 

   763     def test_security_complex_has_text(self):

   764         # use a guest user

   765         self.session = self._user_session()[1]

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

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

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

   769                     ('UnionStep', None, None, [

   770                         ('OneFetchStep', [('Any X WHERE X has_text "bla", EXISTS(X owned_by 5), X is EUser', [{'X': 'EUser'}])],

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

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

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

   774                         ]),

   775                     ])

   776 

   777     def test_security_complex_has_text_limit_offset(self):

   778         # use a guest user

   779         self.session = self._user_session()[1]

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

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

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

   783                     ('UnionFetchStep', [

   784                         ('FetchStep', [('Any X WHERE X has_text "bla", EXISTS(X owned_by 5), X is EUser', [{'X': 'EUser'}])],

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

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

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

   788                         ]),

   789                      ('OneFetchStep',

   790                       [('Any X LIMIT 10 OFFSET 10', [{'X': 'EUser'}, {'X': 'Personne'}])],

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

   792                     ])

   793 

   794     def test_security_complex_aggregat(self):

   795         # use a guest user

   796         self.session = self._user_session()[1]

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

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

   799                      [self.rql, self.system],  None, {'E': 'table1.C0'}, []), 

   800                     ('FetchStep', [('Any X WHERE X is EUser', [{'X': 'EUser'}])],

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

   802                     ('UnionFetchStep', [

   803                         ('FetchStep', [('Any X WHERE EXISTS(X owned_by 5), X is Basket', [{'X': 'Basket'}])],

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

   805                         ('UnionFetchStep',

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

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

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

   809                           ('FetchStep',

   810                            [('Any X WHERE X is IN(Bookmark, Comment, Division, ECache, EConstraint, EConstraintType, EEType, EFRDef, EGroup, ENFRDef, EPermission, EProperty, ERType, Email, EmailAddress, EmailPart, EmailThread, File, Folder, Image, Personne, RQLExpression, Societe, SubDivision, Tag, TrInfo, Transition)',

   811                              sorted([{'X': 'Bookmark'}, {'X': 'Comment'}, {'X': 'Division'},

   812                                       {'X': 'ECache'}, {'X': 'EConstraint'}, {'X': 'EConstraintType'},

   813                                       {'X': 'EEType'}, {'X': 'EFRDef'}, {'X': 'EGroup'},

   814                                       {'X': 'ENFRDef'}, {'X': 'EPermission'}, {'X': 'EProperty'},

   815                                       {'X': 'ERType'}, {'X': 'Email'}, {'X': 'EmailAddress'},

   816                                       {'X': 'EmailPart'}, {'X': 'EmailThread'}, {'X': 'File'},

   817                                       {'X': 'Folder'}, {'X': 'Image'}, {'X': 'Personne'},

   818                                       {'X': 'RQLExpression'}, {'X': 'Societe'}, {'X': 'SubDivision'},

   819                                       {'X': 'Tag'}, {'X': 'TrInfo'}, {'X': 'Transition'}]))],

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

   821                           ]),

   822                         ('FetchStep', [('Any X WHERE EXISTS(X owned_by 5), X is EUser', [{'X': 'EUser'}])],

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

   824                         ('FetchStep', [('Any X WHERE (EXISTS(X owned_by 5)) OR ((((EXISTS(D concerne C?, C owned_by 5, C type "X", X identity D, C is Division, D is Affaire)) OR (EXISTS(H concerne G?, G owned_by 5, G type "X", X identity H, G is SubDivision, H is Affaire))) OR (EXISTS(I concerne F?, F owned_by 5, F type "X", X identity I, F is Societe, I is Affaire))) OR (EXISTS(J concerne E?, E owned_by 5, X identity J, E is Note, J is Affaire))), X is Affaire',

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

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

   827                         ]),

   828                     ('OneFetchStep', [('Any MAX(X)', X_ALL_SOLS)],

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

   830                     ])

   831             

   832     def test_security_complex_aggregat2(self):

   833         # use a guest user

   834         self.session = self._user_session()[1]

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

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

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

   838                      [self.rql, self.system], None, {'X': 'table1.C0'}, []),

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

   840                      [self.rql, self.system],  None, {'E': 'table2.C0'}, []),

   841                     ('FetchStep', [('Any X WHERE X is EUser', [{'X': 'EUser'}])],

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

   843                     ('UnionFetchStep',

   844                      [('FetchStep', [('Any ET,X WHERE X is ET, EXISTS(X owned_by 5), ET is EEType, X is Basket',

   845                                       [{'ET': 'EEType', 'X': 'Basket'}])],

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

   847                       ('FetchStep', [('Any ET,X WHERE X is ET, (EXISTS(X owned_by 5)) OR ((((EXISTS(D concerne C?, C owned_by 5, C type "X", X identity D, C is Division, D is Affaire)) OR (EXISTS(H concerne G?, G owned_by 5, G type "X", X identity H, G is SubDivision, H is Affaire))) OR (EXISTS(I concerne F?, F owned_by 5, F type "X", X identity I, F is Societe, I is Affaire))) OR (EXISTS(J concerne E?, E owned_by 5, X identity J, E is Note, J is Affaire))), ET is EEType, X is Affaire',

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

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

   850                                         'H': 'Affaire', 'J': 'Affaire', 'X': 'Affaire',

   851                                         'ET': 'EEType'}])],

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

   853                        []),

   854                       ('FetchStep', [('Any ET,X WHERE X is ET, EXISTS(X owned_by 5), ET is EEType, X is EUser',

   855                                       [{'ET': 'EEType', 'X': 'EUser'}])],

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

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

   858                       ('UnionFetchStep',

   859                        [('FetchStep', [('Any ET,X WHERE X is ET, ET is EEType, X is IN(Bookmark, Comment, Division, ECache, EConstraint, EConstraintType, EEType, EFRDef, EGroup, ENFRDef, EPermission, EProperty, ERType, Email, EmailAddress, EmailPart, EmailThread, File, Folder, Image, Personne, RQLExpression, Societe, SubDivision, Tag, TrInfo, Transition)',

   860                                         [{'X': 'Bookmark', 'ET': 'EEType'}, {'X': 'Comment', 'ET': 'EEType'},

   861                                          {'X': 'Division', 'ET': 'EEType'}, {'X': 'ECache', 'ET': 'EEType'},

   862                                          {'X': 'EConstraint', 'ET': 'EEType'}, {'X': 'EConstraintType', 'ET': 'EEType'},

   863                                          {'X': 'EEType', 'ET': 'EEType'}, {'X': 'EFRDef', 'ET': 'EEType'},

   864                                          {'X': 'EGroup', 'ET': 'EEType'}, {'X': 'ENFRDef', 'ET': 'EEType'},

   865                                          {'X': 'EPermission', 'ET': 'EEType'}, {'X': 'EProperty', 'ET': 'EEType'},

   866                                          {'X': 'ERType', 'ET': 'EEType'}, {'X': 'Email', 'ET': 'EEType'},

   867                                          {'X': 'EmailAddress', 'ET': 'EEType'}, {'X': 'EmailPart', 'ET': 'EEType'},

   868                                          {'X': 'EmailThread', 'ET': 'EEType'}, {'X': 'File', 'ET': 'EEType'},

   869                                          {'X': 'Folder', 'ET': 'EEType'}, {'X': 'Image', 'ET': 'EEType'},

   870                                          {'X': 'Personne', 'ET': 'EEType'}, {'X': 'RQLExpression', 'ET': 'EEType'},

   871                                          {'X': 'Societe', 'ET': 'EEType'}, {'X': 'SubDivision', 'ET': 'EEType'},

   872                                          {'X': 'Tag', 'ET': 'EEType'}, {'X': 'TrInfo', 'ET': 'EEType'},

   873                                          {'X': 'Transition', 'ET': 'EEType'}])],

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

   875                         ('FetchStep',

   876                          [('Any ET,X WHERE X is ET, ET is EEType, X is IN(Card, Note, State)',

   877                            [{'ET': 'EEType', 'X': 'Card'},

   878                             {'ET': 'EEType', 'X': 'Note'},

   879                             {'ET': 'EEType', 'X': 'State'}])],

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

   881                         ]),

   882                     ]),

   883                     ('OneFetchStep',

   884                      [('Any ET,COUNT(X) GROUPBY ET ORDERBY ET',

   885                        sorted([{'ET': 'EEType', 'X': 'Affaire'}, {'ET': 'EEType', 'X': 'Basket'},

   886                                {'ET': 'EEType', 'X': 'Bookmark'}, {'ET': 'EEType', 'X': 'Card'},

   887                                {'ET': 'EEType', 'X': 'Comment'}, {'ET': 'EEType', 'X': 'Division'},

   888                                {'ET': 'EEType', 'X': 'ECache'}, {'ET': 'EEType', 'X': 'EConstraint'},

   889                                {'ET': 'EEType', 'X': 'EConstraintType'}, {'ET': 'EEType', 'X': 'EEType'},

   890                                {'ET': 'EEType', 'X': 'EFRDef'}, {'ET': 'EEType', 'X': 'EGroup'},

   891                                {'ET': 'EEType', 'X': 'ENFRDef'}, {'ET': 'EEType', 'X': 'EPermission'},

   892                                {'ET': 'EEType', 'X': 'EProperty'}, {'ET': 'EEType', 'X': 'ERType'},

   893                                {'ET': 'EEType', 'X': 'EUser'}, {'ET': 'EEType', 'X': 'Email'},

   894                                {'ET': 'EEType', 'X': 'EmailAddress'}, {'ET': 'EEType', 'X': 'EmailPart'},

   895                                {'ET': 'EEType', 'X': 'EmailThread'}, {'ET': 'EEType', 'X': 'File'},

   896                                {'ET': 'EEType', 'X': 'Folder'}, {'ET': 'EEType', 'X': 'Image'},

   897                                {'ET': 'EEType', 'X': 'Note'}, {'ET': 'EEType', 'X': 'Personne'},

   898                                {'ET': 'EEType', 'X': 'RQLExpression'}, {'ET': 'EEType', 'X': 'Societe'},

   899                                {'ET': 'EEType', 'X': 'State'}, {'ET': 'EEType', 'X': 'SubDivision'},

   900                                {'ET': 'EEType', 'X': 'Tag'}, {'ET': 'EEType', 'X': 'TrInfo'},

   901                                {'ET': 'EEType', 'X': 'Transition'}]))],

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

   903                     ])

   904 

   905     def test_security_3sources(self):

   906         # use a guest user

   907         self.session = self._user_session()[1]

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

   909                    [('FetchStep',

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

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

   912                     ('FetchStep',

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

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

   915                     ('OneFetchStep',

   916                      [('Any X,XT WHERE X owned_by U, X title XT, EXISTS(U owned_by 5), U is EUser, X is Card',

   917                        [{'X': 'Card', 'U': 'EUser', 'XT': 'String'}])],

   918                      None, None, [self.system],

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

   920                     ])

   921 

   922     def test_security_3sources_identity(self):

   923         self.restore_orig_euser_security()

   924         # use a guest user

   925         self.session = self._user_session()[1]

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

   927                    [('FetchStep',

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

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

   930                     ('OneFetchStep',

   931                      [('Any X,XT WHERE X owned_by U, X title XT, U login "syt", EXISTS(U identity 5), U is EUser, X is Card',

   932                        [{'U': 'EUser', 'X': 'Card', 'XT': 'String'}])],

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

   934                     ])

   935 

   936     def test_security_3sources_identity_optional_var(self):

   937         self.restore_orig_euser_security()

   938         # use a guest user

   939         self.session = self._user_session()[1]

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

   941                    [('FetchStep',

   942                      [('Any U,L WHERE U identity 5, U login L, U is EUser',

   943                        [{'L': 'String', u'U': 'EUser'}])],

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

   945                     ('FetchStep',

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

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

   948                     ('OneFetchStep',

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

   950                        [{'X': 'Card', 'XT': 'String'}])],

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

   952                                                  'U': 'table0.C0',

   953                                                  'X': 'table1.C0',

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

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

   956                     ])

   957 

   958     def test_security_3sources_limit_offset(self):

   959         # use a guest user

   960         self.session = self._user_session()[1]

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

   962                    [('FetchStep',

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

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

   965                     ('FetchStep',

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

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

   968                     ('OneFetchStep',

   969                      [('Any X,XT LIMIT 10 OFFSET 10 WHERE X owned_by U, X title XT, EXISTS(U owned_by 5), U is EUser, X is Card',

   970                        [{'X': 'Card', 'U': 'EUser', 'XT': 'String'}])],

   971                      10, 10, [self.system],

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

   973                     ])

   974     

   975     def test_exists_base(self):

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

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

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

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

   980                                       'EXISTS(X in_group G, G name "bougloup", G is EGroup), S is State, X is EUser',

   981                                        [{'X': 'EUser', 'L': 'String', 'S': 'State', 'G': 'EGroup'}])],

   982                      None, None, [self.system],

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

   984 

   985     def test_exists_complex(self):

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

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

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

   989                     ('OneFetchStep',

   990                      [('Any G WHERE X in_group G, G name "managers", EXISTS(X copain T, T is EUser), G is EGroup, X is EUser',

   991                        [{'X': 'EUser', 'T': 'EUser', 'G': 'EGroup'}])],

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

   993 

   994     def test_exists3(self):

   995         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"))',

   996                    [('FetchStep',

   997                      [('Any T WHERE T login IN("comme", "cochon"), T is EUser',

   998                        [{'T': 'EUser'}])],

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

  1000                     ('FetchStep',

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

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

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

  1004                     ('OneFetchStep',

  1005                      [('Any G,L WHERE X in_group G, X login L, G name "managers", EXISTS(X copain T, T is EUser), G is EGroup, X is EUser',

  1006                        [{'G': 'EGroup', 'L': 'String', 'T': 'EUser', 'X': 'EUser'}])],

  1007                      None, None,

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

  1009 

  1010     def test_exists4(self):

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

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

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

  1014                    [('FetchStep',

  1015                      [('Any T,L WHERE T login L, T login IN("comme", "cochon"), T is EUser', [{'T': 'EUser', 'L': 'String'}])],

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

  1017                      {'T': 'table0.C0', 'T.login': 'table0.C1', 'L': 'table0.C1'}, []),

  1018                     ('FetchStep',

  1019                      [('Any T2 WHERE T2 login "billy", T2 is EUser', [{'T2': 'EUser'}])],

  1020                      [self.ldap, self.system], None, {'T2': 'table1.C0'}, []),

  1021                     ('FetchStep',

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

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

  1024                     ('OneFetchStep',

  1025                      [('Any G,L WHERE X in_group G, X login L, G name "managers", (EXISTS(X copain T, T login L, T is EUser)) OR (EXISTS(X in_state S, S name "pascontent", NOT X copain T2, S is State, T2 is EUser)), G is EGroup, X is EUser',

  1026                        [{'G': 'EGroup', 'L': 'String', 'S': 'State', 'T': 'EUser', 'T2': 'EUser', 'X': 'EUser'}])],

  1027                      None, None, [self.system],

  1028                      {'T2': 'table1.C0', 'L': 'table2.C0',

  1029                       'T': 'table0.C0', 'T.login': 'table0.C1', 'X': 'table2.C1', 'X.login': 'table2.C0'}, [])])

  1030 

  1031     def test_exists5(self):

  1032         self._test('Any GN,L WHERE X in_group G, X login L, G name GN, '

  1033                    'EXISTS(X copain T, T login in ("comme", "cochon")) AND '

  1034                    'NOT EXISTS(X copain T2, T2 login "billy")',

  1035                    [('FetchStep', [('Any T WHERE T login IN("comme", "cochon"), T is EUser',

  1036                                     [{'T': 'EUser'}])],

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

  1038                     ('FetchStep', [('Any T2 WHERE T2 login "billy", T2 is EUser', [{'T2': 'EUser'}])],

  1039                      [self.ldap, self.system], None, {'T2': 'table1.C0'}, []),

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

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

  1042                      {'X': 'table2.C1', 'X.login': 'table2.C0', 'L': 'table2.C0'}, []),

  1043                     ('OneFetchStep', [('Any GN,L WHERE X in_group G, X login L, G name GN, EXISTS(X copain T, T is EUser), NOT EXISTS(X copain T2, T2 is EUser), G is EGroup, X is EUser',

  1044                        [{'G': 'EGroup', 'GN': 'String', 'L': 'String', 'T': 'EUser', 'T2': 'EUser', 'X': 'EUser'}])],

  1045                      None, None, [self.system],

  1046                      {'T': 'table0.C0', 'T2': 'table1.C0',

  1047                       'X': 'table2.C1', 'X.login': 'table2.C0', 'L': 'table2.C0'}, [])])

  1048 

  1049     def test_relation_need_split(self):

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

  1051                    [('UnionStep', None, None, [

  1052                        ('OneFetchStep', [('Any X,S WHERE X in_state S, S is State, X is IN(Affaire, EUser)',

  1053                                           [{'X': 'Affaire', 'S': 'State'}, {'X': 'EUser', 'S': 'State'}])], 

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

  1055                        ('OneFetchStep', [('Any X,S WHERE X in_state S, S is State, X is Note',

  1056                                           [{'X': 'Note', 'S': 'State'}])], 

  1057                         None, None, [self.rql, self.system], {}, []),

  1058                     ])])

  1059 

  1060     def test_relation_selection_need_split(self):

  1061         self._test('Any X,S,U WHERE X in_state S, X todo_by U',

  1062                    [('FetchStep', [('Any X,S WHERE X in_state S, S is State, X is Note',

  1063                                     [{'X': 'Note', 'S': 'State'}])],

  1064                      [self.rql, self.system], None, {'X': 'table0.C0', 'S': 'table0.C1'}, []),

  1065                      ('UnionStep', None, None,

  1066                       [('OneFetchStep', [('Any X,S,U WHERE X in_state S, X todo_by U, S is State, U is EUser, X is Note',

  1067                                           [{'X': 'Note', 'S': 'State', 'U': 'EUser'}])],

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

  1069                        ('OneFetchStep', [('Any X,S,U WHERE X in_state S, X todo_by U, S is State, U is Personne, X is Affaire',

  1070                                           [{'X': 'Affaire', 'S': 'State', 'U': 'Personne'}])],

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

  1072                        ])

  1073                     ])

  1074 

  1075     def test_relation_restriction_need_split(self):

  1076         self._test('Any X,U WHERE X in_state S, S name "pending", X todo_by U',

  1077                    [('FetchStep', [('Any X WHERE X in_state S, S name "pending", S is State, X is Note',

  1078                                     [{'X': 'Note', 'S': 'State'}])],

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

  1080                      ('UnionStep', None, None,

  1081                       [('OneFetchStep', [('Any X,U WHERE X todo_by U, U is EUser, X is Note',

  1082                                           [{'X': 'Note', 'U': 'EUser'}])],

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

  1084                        ('OneFetchStep', [('Any X,U WHERE X in_state S, S name "pending", X todo_by U, S is State, U is Personne, X is Affaire',

  1085                                           [{'S': 'State', 'U': 'Personne', 'X': 'Affaire'}])],

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

  1087                        ])

  1088                     ])

  1089 

  1090     def test_relation_restriction_ambigous_need_split(self):

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

  1092                    [('FetchStep', [('Any X WHERE X in_state S, S name "pending", S is State, X is Note',

  1093                                     [{'X': 'Note', 'S': 'State'}])],

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

  1095                     ('UnionStep', None, None, [

  1096                         ('OneFetchStep', [('Any X,T WHERE T tags X, T is Tag, X is Note',

  1097                                            [{'X': 'Note', 'T': 'Tag'}])],

  1098                          None, None,

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

  1100                         ('OneFetchStep', [('Any X,T WHERE X in_state S, S name "pending", T tags X, S is State, T is Tag, X is IN(Affaire, EUser)',

  1101                                            [{'X': 'Affaire', 'S': 'State', 'T': 'Tag'},

  1102                                             {'X': 'EUser', 'S': 'State', 'T': 'Tag'}])],

  1103                          None, None,

  1104                          [self.system], {}, []),

  1105                         ])

  1106                     ])

  1107 

  1108     def test_not_relation_no_split_internal(self):

  1109         ueid = self.session.user.eid

  1110         # NOT on a relation supported by rql and system source: we want to get

  1111         # all states (eg from both sources) which are not related to entity with the

  1112         # given eid. The "NOT X in_state S, X eid %(x)s" expression is necessarily true

  1113         # in the source where %(x)s is not coming from and will be removed during rql

  1114         # generation for the external source

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

  1116                    [('OneFetchStep', [('Any SN WHERE NOT 5 in_state S, S name SN, S is State', [{'S': 'State', 'SN': 'String'}])], 

  1117                      None, None, [self.rql, self.system], {}, [])],

  1118                    {'x': ueid})

  1119 

  1120     def test_not_relation_no_split_external(self):

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

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

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

  1124                    [('UnionStep', None, None,

  1125                      [('OneFetchStep',

  1126                        [('Any SN WHERE NOT 999999 in_state S, S name SN, S is State',

  1127                          [{'S': 'State', 'SN': 'String'}])],

  1128                        None, None, [self.rql], {},

  1129                        []),

  1130                       ('OneFetchStep',

  1131                        [('Any SN WHERE S name SN, S is State',

  1132                          [{'S': 'State', 'SN': 'String'}])],

  1133                        None, None, [self.system], {},

  1134                        [])]

  1135                      )],