server/test/unittest_msplanner.py
changeset 5768 1e73a466aa69
parent 5590 a56eb02f9ce7
child 6056 37a37e5ba330
equal deleted inserted replaced
5766:c397819f2482 5768:1e73a466aa69
   411 
   411 
   412     def test_simple_system_ldap_ordered_limit_offset(self):
   412     def test_simple_system_ldap_ordered_limit_offset(self):
   413         """retrieve CWUser X from both sources and return concatenation of results
   413         """retrieve CWUser X from both sources and return concatenation of results
   414         """
   414         """
   415         self._test('CWUser X ORDERBY X LIMIT 10 OFFSET 10',
   415         self._test('CWUser X ORDERBY X LIMIT 10 OFFSET 10',
   416                    [('AggrStep', 'Any X ORDERBY X', 10, 10, 'table0', None, [
   416                    [('AggrStep', 'SELECT table0.C0 FROM table0 ORDER BY table0.C0 LIMIT 10 OFFSET 10', None, [
   417                        ('FetchStep', [('Any X WHERE X is CWUser', [{'X': 'CWUser'}])],
   417                        ('FetchStep', [('Any X WHERE X is CWUser', [{'X': 'CWUser'}])],
   418                         [self.ldap, self.system], {}, {'X': 'table0.C0'}, []),
   418                         [self.ldap, self.system], {}, {'X': 'table0.C0'}, []),
   419                        ]),
   419                        ]),
   420                    ])
   420                    ])
   421     def test_simple_system_ldap_aggregat(self):
   421     def test_simple_system_ldap_aggregat(self):
   422         """retrieve CWUser X from both sources and return concatenation of results
   422         """retrieve CWUser X from both sources and return concatenation of results
   423         """
   423         """
   424         # COUNT(X) is kept in sub-step and transformed into SUM(X) in the AggrStep
   424         # COUNT(X) is kept in sub-step and transformed into SUM(X) in the AggrStep
   425         self._test('Any COUNT(X) WHERE X is CWUser',
   425         self._test('Any COUNT(X) WHERE X is CWUser',
   426                    [('AggrStep', 'Any COUNT(X)', None, None, 'table0', None, [
   426                    [('AggrStep', 'SELECT SUM(table0.C0) FROM table0', None, [
   427                        ('FetchStep', [('Any COUNT(X) WHERE X is CWUser', [{'X': 'CWUser'}])],
   427                        ('FetchStep', [('Any COUNT(X) WHERE X is CWUser', [{'X': 'CWUser'}])],
   428                         [self.ldap, self.system], {}, {'COUNT(X)': 'table0.C0'}, []),
   428                         [self.ldap, self.system], {}, {'COUNT(X)': 'table0.C0'}, []),
   429                        ]),
   429                        ]),
   430                    ])
   430                    ])
   431 
   431 
   496                      [self.system], {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, [])
   496                      [self.system], {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, [])
   497                     ])
   497                     ])
   498 
   498 
   499     def test_complex_ordered(self):
   499     def test_complex_ordered(self):
   500         self._test('Any L ORDERBY L WHERE X login L',
   500         self._test('Any L ORDERBY L WHERE X login L',
   501                    [('AggrStep', 'Any L ORDERBY L', None, None, 'table0', None,
   501                    [('AggrStep', 'SELECT table0.C0 FROM table0 ORDER BY table0.C0', None,
   502                      [('FetchStep', [('Any L WHERE X login L, X is CWUser',
   502                      [('FetchStep', [('Any L WHERE X login L, X is CWUser',
   503                                       [{'X': 'CWUser', 'L': 'String'}])],
   503                                       [{'X': 'CWUser', 'L': 'String'}])],
   504                        [self.ldap, self.system], {}, {'X.login': 'table0.C0', 'L': 'table0.C0'}, []),
   504                        [self.ldap, self.system], {}, {'X.login': 'table0.C0', 'L': 'table0.C0'}, []),
   505                       ])
   505                       ])
   506                     ])
   506                     ])
   507 
   507 
   508     def test_complex_ordered_limit_offset(self):
   508     def test_complex_ordered_limit_offset(self):
   509         self._test('Any L ORDERBY L LIMIT 10 OFFSET 10 WHERE X login L',
   509         self._test('Any L ORDERBY L LIMIT 10 OFFSET 10 WHERE X login L',
   510                    [('AggrStep', 'Any L ORDERBY L', 10, 10, 'table0', None,
   510                    [('AggrStep', 'SELECT table0.C0 FROM table0 ORDER BY table0.C0 LIMIT 10 OFFSET 10', None,
   511                      [('FetchStep', [('Any L WHERE X login L, X is CWUser',
   511                      [('FetchStep', [('Any L WHERE X login L, X is CWUser',
   512                                       [{'X': 'CWUser', 'L': 'String'}])],
   512                                       [{'X': 'CWUser', 'L': 'String'}])],
   513                        [self.ldap, self.system], {}, {'X.login': 'table0.C0', 'L': 'table0.C0'}, []),
   513                        [self.ldap, self.system], {}, {'X.login': 'table0.C0', 'L': 'table0.C0'}, []),
   514                       ])
   514                       ])
   515                     ])
   515                     ])
   591         1. retrieve CWUser X from system and ldap sources, Person X from system source only, store
   591         1. retrieve CWUser X from system and ldap sources, Person X from system source only, store
   592            each result in the same temp table
   592            each result in the same temp table
   593         2. return content of the table sorted
   593         2. return content of the table sorted
   594         """
   594         """
   595         self._test('Any X,F ORDERBY F WHERE X firstname F',
   595         self._test('Any X,F ORDERBY F WHERE X firstname F',
   596                    [('AggrStep', 'Any X,F ORDERBY F', None, None, 'table0', None,
   596                    [('AggrStep', 'SELECT table0.C0, table0.C1 FROM table0 ORDER BY table0.C1', None,
   597                      [('FetchStep', [('Any X,F WHERE X firstname F, X is CWUser',
   597                      [('FetchStep', [('Any X,F WHERE X firstname F, X is CWUser',
   598                                       [{'X': 'CWUser', 'F': 'String'}])],
   598                                       [{'X': 'CWUser', 'F': 'String'}])],
   599                        [self.ldap, self.system], {},
   599                        [self.ldap, self.system], {},
   600                        {'X': 'table0.C0', 'X.firstname': 'table0.C1', 'F': 'table0.C1'}, []),
   600                        {'X': 'table0.C0', 'X.firstname': 'table0.C1', 'F': 'table0.C1'}, []),
   601                       ('FetchStep', [('Any X,F WHERE X firstname F, X is Personne',
   601                       ('FetchStep', [('Any X,F WHERE X firstname F, X is Personne',
   655                      None, None, [self.system], {}, [])
   655                      None, None, [self.system], {}, [])
   656                     ])
   656                     ])
   657 
   657 
   658     def test_complex_typed_aggregat(self):
   658     def test_complex_typed_aggregat(self):
   659         self._test('Any MAX(X) WHERE X is Card',
   659         self._test('Any MAX(X) WHERE X is Card',
   660                    [('AggrStep', 'Any MAX(X)', None, None, 'table0',  None,
   660                    [('AggrStep', 'SELECT MAX(table0.C0) FROM table0',  None,
   661                      [('FetchStep',
   661                      [('FetchStep',
   662                        [('Any MAX(X) WHERE X is Card', [{'X': 'Card'}])],
   662                        [('Any MAX(X) WHERE X is Card', [{'X': 'Card'}])],
   663                        [self.cards, self.system], {}, {'MAX(X)': 'table0.C0'}, [])
   663                        [self.cards, self.system], {}, {'MAX(X)': 'table0.C0'}, [])
   664                       ])
   664                       ])
   665                     ])
   665                     ])
  1297                                            [{'X': 'EmailThread'}])],
  1297                                            [{'X': 'EmailThread'}])],
  1298                          None, None, [self.system], {}, []),
  1298                          None, None, [self.system], {}, []),
  1299                         ]),
  1299                         ]),
  1300                     ])
  1300                     ])
  1301 
  1301 
       
  1302     def test_has_text_orderby_rank(self):
       
  1303         self._test('Any X ORDERBY FTIRANK(X) WHERE X has_text "bla", X firstname "bla"',
       
  1304                    [('FetchStep', [('Any X WHERE X firstname "bla", X is CWUser', [{'X': 'CWUser'}])],
       
  1305                      [self.ldap, self.system], None, {'X': 'table0.C0'}, []),
       
  1306                     ('AggrStep', 'SELECT table1.C1 FROM table1 ORDER BY table1.C0', None, [
       
  1307                         ('FetchStep', [('Any FTIRANK(X),X WHERE X has_text "bla", X is CWUser',
       
  1308                                         [{'X': 'CWUser'}])],
       
  1309                          [self.system], {'X': 'table0.C0'}, {'FTIRANK(X)': 'table1.C0', 'X': 'table1.C1'}, []),
       
  1310                         ('FetchStep', [('Any FTIRANK(X),X WHERE X has_text "bla", X firstname "bla", X is Personne',
       
  1311                                         [{'X': 'Personne'}])],
       
  1312                          [self.system], {}, {'FTIRANK(X)': 'table1.C0', 'X': 'table1.C1'}, []),
       
  1313                         ]),
       
  1314                     ])
       
  1315 
       
  1316     def test_security_has_text_orderby_rank(self):
       
  1317         # use a guest user
       
  1318         self.session = self.user_groups_session('guests')
       
  1319         self._test('Any X ORDERBY FTIRANK(X) WHERE X has_text "bla", X firstname "bla"',
       
  1320                    [('FetchStep', [('Any X WHERE X firstname "bla", X is CWUser', [{'X': 'CWUser'}])],
       
  1321                      [self.ldap, self.system], None, {'X': 'table1.C0'}, []),
       
  1322                     ('UnionFetchStep',
       
  1323                      [('FetchStep', [('Any X WHERE X firstname "bla", X is Personne', [{'X': 'Personne'}])],
       
  1324                        [self.system], {}, {'X': 'table0.C0'}, []),
       
  1325                       ('FetchStep', [('Any X WHERE EXISTS(X owned_by 5), X is CWUser', [{'X': 'CWUser'}])],
       
  1326                        [self.system], {'X': 'table1.C0'}, {'X': 'table0.C0'}, [])]),
       
  1327                     ('OneFetchStep', [('Any X ORDERBY FTIRANK(X) WHERE X has_text "bla"',
       
  1328                                        [{'X': 'CWUser'}, {'X': 'Personne'}])],
       
  1329                      None, None, [self.system], {'X': 'table0.C0'}, []),
       
  1330                     ])
       
  1331 
       
  1332     def test_has_text_select_rank(self):
       
  1333         self._test('Any X, FTIRANK(X) WHERE X has_text "bla", X firstname "bla"',
       
  1334                    # XXX unecessary duplicate selection
       
  1335                    [('FetchStep', [('Any X,X WHERE X firstname "bla", X is CWUser', [{'X': 'CWUser'}])],
       
  1336                      [self.ldap, self.system], None, {'X': 'table0.C1'}, []),
       
  1337                     ('UnionStep', None, None, [
       
  1338                         ('OneFetchStep', [('Any X,FTIRANK(X) WHERE X has_text "bla", X is CWUser', [{'X': 'CWUser'}])],
       
  1339                          None, None, [self.system], {'X': 'table0.C1'}, []),
       
  1340                         ('OneFetchStep', [('Any X,FTIRANK(X) WHERE X has_text "bla", X firstname "bla", X is Personne', [{'X': 'Personne'}])],
       
  1341                          None, None, [self.system], {}, []),
       
  1342                         ]),
       
  1343                     ])
       
  1344 
       
  1345     def test_security_has_text_select_rank(self):
       
  1346         # use a guest user
       
  1347         self.session = self.user_groups_session('guests')
       
  1348         self._test('Any X, FTIRANK(X) WHERE X has_text "bla", X firstname "bla"',
       
  1349                    [('FetchStep', [('Any X,X WHERE X firstname "bla", X is CWUser', [{'X': 'CWUser'}])],
       
  1350                      [self.ldap, self.system], None, {'X': 'table0.C1'}, []),
       
  1351                     ('UnionStep', None, None, [
       
  1352                         ('OneFetchStep', [('Any X,FTIRANK(X) WHERE X has_text "bla", EXISTS(X owned_by 5), X is CWUser', [{'X': 'CWUser'}])],
       
  1353                          None, None, [self.system], {'X': 'table0.C1'}, []),
       
  1354                         ('OneFetchStep', [('Any X,FTIRANK(X) WHERE X has_text "bla", X firstname "bla", X is Personne', [{'X': 'Personne'}])],
       
  1355                          None, None, [self.system], {}, []),
       
  1356                         ]),
       
  1357                     ])
       
  1358 
  1302     def test_sort_func(self):
  1359     def test_sort_func(self):
  1303         self._test('Note X ORDERBY DUMB_SORT(RF) WHERE X type RF',
  1360         self._test('Note X ORDERBY DUMB_SORT(RF) WHERE X type RF',
  1304                    [('AggrStep', 'Any X ORDERBY DUMB_SORT(RF)', None, None, 'table0', None, [
  1361                    [('AggrStep', 'SELECT table0.C0 FROM table0 ORDER BY DUMB_SORT(table0.C1)', None, [
  1305                        ('FetchStep', [('Any X,RF WHERE X type RF, X is Note',
  1362                        ('FetchStep', [('Any X,RF WHERE X type RF, X is Note',
  1306                                        [{'X': 'Note', 'RF': 'String'}])],
  1363                                        [{'X': 'Note', 'RF': 'String'}])],
  1307                         [self.cards, self.system], {}, {'X': 'table0.C0', 'X.type': 'table0.C1', 'RF': 'table0.C1'}, []),
  1364                         [self.cards, self.system], {}, {'X': 'table0.C0', 'X.type': 'table0.C1', 'RF': 'table0.C1'}, []),
  1308                        ])
  1365                        ])
  1309                     ])
  1366                     ])
  1310 
  1367 
  1311     def test_ambigous_sort_func(self):
  1368     def test_ambigous_sort_func(self):
  1312         self._test('Any X ORDERBY DUMB_SORT(RF) WHERE X title RF, X is IN (Bookmark, Card, EmailThread)',
  1369         self._test('Any X ORDERBY DUMB_SORT(RF) WHERE X title RF, X is IN (Bookmark, Card, EmailThread)',
  1313                    [('AggrStep', 'Any X ORDERBY DUMB_SORT(RF)',
  1370                    [('AggrStep', 'SELECT table0.C0 FROM table0 ORDER BY DUMB_SORT(table0.C1)', None,
  1314                      None, None, 'table0', None,
       
  1315                      [('FetchStep', [('Any X,RF WHERE X title RF, X is Card',
  1371                      [('FetchStep', [('Any X,RF WHERE X title RF, X is Card',
  1316                                       [{'X': 'Card', 'RF': 'String'}])],
  1372                                       [{'X': 'Card', 'RF': 'String'}])],
  1317                        [self.cards, self.system], {},
  1373                        [self.cards, self.system], {},
  1318                        {'X': 'table0.C0', 'X.title': 'table0.C1', 'RF': 'table0.C1'}, []),
  1374                        {'X': 'table0.C0', 'X.title': 'table0.C1', 'RF': 'table0.C1'}, []),
  1319                       ('FetchStep', [('Any X,RF WHERE X title RF, X is IN(Bookmark, EmailThread)',
  1375                       ('FetchStep', [('Any X,RF WHERE X title RF, X is IN(Bookmark, EmailThread)',
  1726                                     [{'X': 'Note', 'D': 'Datetime'}])],
  1782                                     [{'X': 'Note', 'D': 'Datetime'}])],
  1727                      [self.cards, self.system], None, {'X': 'table0.C0', 'X.modification_date': 'table0.C1', 'D': 'table0.C1'}, []),
  1783                      [self.cards, self.system], None, {'X': 'table0.C0', 'X.modification_date': 'table0.C1', 'D': 'table0.C1'}, []),
  1728                     ('FetchStep', [('Any X,D WHERE X modification_date D, X is CWUser',
  1784                     ('FetchStep', [('Any X,D WHERE X modification_date D, X is CWUser',
  1729                                     [{'X': 'CWUser', 'D': 'Datetime'}])],
  1785                                     [{'X': 'CWUser', 'D': 'Datetime'}])],
  1730                      [self.ldap, self.system], None, {'X': 'table1.C0', 'X.modification_date': 'table1.C1', 'D': 'table1.C1'}, []),
  1786                      [self.ldap, self.system], None, {'X': 'table1.C0', 'X.modification_date': 'table1.C1', 'D': 'table1.C1'}, []),
  1731                     ('AggrStep', 'Any X ORDERBY D DESC', None, None, 'table2', None, [
  1787                     ('AggrStep', 'SELECT table2.C0 FROM table2 ORDER BY table2.C1 DESC', None, [
  1732                         ('FetchStep', [('Any X,D WHERE E eid %s, E wf_info_for X, X modification_date D, E is TrInfo, X is Affaire'%treid,
  1788                         ('FetchStep', [('Any X,D WHERE E eid %s, E wf_info_for X, X modification_date D, E is TrInfo, X is Affaire'%treid,
  1733                                         [{'X': 'Affaire', 'E': 'TrInfo', 'D': 'Datetime'}])],
  1789                                         [{'X': 'Affaire', 'E': 'TrInfo', 'D': 'Datetime'}])],
  1734                          [self.system],
  1790                          [self.system],
  1735                          {},
  1791                          {},
  1736                          {'X': 'table2.C0', 'X.modification_date': 'table2.C1', 'D': 'table2.C1', 'E.wf_info_for': 'table2.C0'}, []),
  1792                          {'X': 'table2.C0', 'X.modification_date': 'table2.C1', 'D': 'table2.C1', 'E.wf_info_for': 'table2.C0'}, []),
  1869         self._test('Any X ORDERBY Z DESC WHERE X modification_date Z, E eid %(x)s, E see_also X',
  1925         self._test('Any X ORDERBY Z DESC WHERE X modification_date Z, E eid %(x)s, E see_also X',
  1870                    [('FetchStep', [('Any X,Z WHERE X modification_date Z, X is Note',
  1926                    [('FetchStep', [('Any X,Z WHERE X modification_date Z, X is Note',
  1871                                     [{'X': 'Note', 'Z': 'Datetime'}])],
  1927                                     [{'X': 'Note', 'Z': 'Datetime'}])],
  1872                      [self.cards, self.system], None, {'X': 'table0.C0', 'X.modification_date': 'table0.C1', 'Z': 'table0.C1'},
  1928                      [self.cards, self.system], None, {'X': 'table0.C0', 'X.modification_date': 'table0.C1', 'Z': 'table0.C1'},
  1873                      []),
  1929                      []),
  1874                     ('AggrStep', 'Any X ORDERBY Z DESC',
  1930                     ('AggrStep', 'SELECT table1.C0 FROM table1 ORDER BY table1.C1 DESC', None,
  1875                      None, None, 'table1', None,
       
  1876                      [('FetchStep', [('Any X,Z WHERE X modification_date Z, 999999 see_also X, X is Bookmark',
  1931                      [('FetchStep', [('Any X,Z WHERE X modification_date Z, 999999 see_also X, X is Bookmark',
  1877                                       [{'X': 'Bookmark', 'Z': 'Datetime'}])],
  1932                                       [{'X': 'Bookmark', 'Z': 'Datetime'}])],
  1878                        [self.system], {},   {'X': 'table1.C0', 'X.modification_date': 'table1.C1',
  1933                        [self.system], {},   {'X': 'table1.C0', 'X.modification_date': 'table1.C1',
  1879                                              'Z': 'table1.C1'},
  1934                                              'Z': 'table1.C1'},
  1880                        []),
  1935                        []),