server/test/unittest_querier.py
branch3.5
changeset 2920 64322aa83a1d
parent 2608 21856eda34f6
child 2921 8e2544e78a5e
equal deleted inserted replaced
2919:662f35236d1c 2920:64322aa83a1d
   107                                        'I': 'Affaire',
   107                                        'I': 'Affaire',
   108                                        'J': 'Affaire',
   108                                        'J': 'Affaire',
   109                                        'X': 'Affaire',
   109                                        'X': 'Affaire',
   110                                        'ET': 'CWEType', 'ETN': 'String'}])
   110                                        'ET': 'CWEType', 'ETN': 'String'}])
   111         rql, solutions = partrqls[1]
   111         rql, solutions = partrqls[1]
   112         self.assertEquals(rql,  'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, '
   112         self.assertEquals(rql,  'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, X is IN(BaseTransition, Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, ExternalUri, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, SubWorkflowExitPoint, Tag, TrInfo, Transition, Workflow, WorkflowTransition)')
   113                           'X is IN(Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, ExternalUri, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, Tag, TrInfo, Transition)')
       
   114         self.assertListEquals(sorted(solutions),
   113         self.assertListEquals(sorted(solutions),
   115                               sorted([{'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'},
   114                               sorted([{'X': 'BaseTransition', 'ETN': 'String', 'ET': 'CWEType'},
       
   115                                       {'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'},
   116                                       {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'},
   116                                       {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'},
   117                                       {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'},
   117                                       {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'},
   118                                       {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'},
   118                                       {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'},
   119                                       {'X': 'CWCache', 'ETN': 'String', 'ET': 'CWEType'},
   119                                       {'X': 'CWCache', 'ETN': 'String', 'ET': 'CWEType'},
   120                                       {'X': 'CWConstraint', 'ETN': 'String', 'ET': 'CWEType'},
   120                                       {'X': 'CWConstraint', 'ETN': 'String', 'ET': 'CWEType'},
   139                                       {'X': 'Personne', 'ETN': 'String', 'ET': 'CWEType'},
   139                                       {'X': 'Personne', 'ETN': 'String', 'ET': 'CWEType'},
   140                                       {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'CWEType'},
   140                                       {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'CWEType'},
   141                                       {'X': 'Societe', 'ETN': 'String', 'ET': 'CWEType'},
   141                                       {'X': 'Societe', 'ETN': 'String', 'ET': 'CWEType'},
   142                                       {'X': 'State', 'ETN': 'String', 'ET': 'CWEType'},
   142                                       {'X': 'State', 'ETN': 'String', 'ET': 'CWEType'},
   143                                       {'X': 'SubDivision', 'ETN': 'String', 'ET': 'CWEType'},
   143                                       {'X': 'SubDivision', 'ETN': 'String', 'ET': 'CWEType'},
       
   144                                       {'X': 'SubWorkflowExitPoint', 'ETN': 'String', 'ET': 'CWEType'},
   144                                       {'X': 'Tag', 'ETN': 'String', 'ET': 'CWEType'},
   145                                       {'X': 'Tag', 'ETN': 'String', 'ET': 'CWEType'},
   145                                       {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'},
   146                                       {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'},
   146                                       {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'}]))
   147                                       {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'},
       
   148                                       {'X': 'Workflow', 'ETN': 'String', 'ET': 'CWEType'},
       
   149                                       {'X': 'WorkflowTransition', 'ETN': 'String', 'ET': 'CWEType'}]))
   147         rql, solutions = partrqls[2]
   150         rql, solutions = partrqls[2]
   148         self.assertEquals(rql,
   151         self.assertEquals(rql,
   149                           'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), '
   152                           'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), '
   150                           'ET is CWEType, X is Basket')
   153                           'ET is CWEType, X is Basket')
   151         self.assertEquals(solutions, [{'ET': 'CWEType',
   154         self.assertEquals(solutions, [{'ET': 'CWEType',
   283         self.assert_(('Personne',) in rset.description)
   286         self.assert_(('Personne',) in rset.description)
   284         rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")')
   287         rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")')
   285         self.assert_(('Personne',) in rset.description)
   288         self.assert_(('Personne',) in rset.description)
   286 
   289 
   287     def test_select_not_attr(self):
   290     def test_select_not_attr(self):
   288         self.execute("INSERT Personne X: X nom 'bidule'")
   291         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   289         self.execute("INSERT Societe X: X nom 'chouette'")
   292         seid = self.execute("INSERT Societe X: X nom 'chouette'")[0][0]
   290         rset = self.execute('Personne X WHERE NOT X nom "bidule"')
   293         rset = self.execute('Personne X WHERE NOT X nom "bidule"')
   291         self.assertEquals(len(rset.rows), 0, rset.rows)
   294         self.assertEquals(len(rset.rows), 0, rset.rows)
   292         rset = self.execute('Personne X WHERE NOT X nom "bid"')
   295         rset = self.execute('Personne X WHERE NOT X nom "bid"')
   293         self.assertEquals(len(rset.rows), 1, rset.rows)
   296         self.assertEquals(len(rset.rows), 1, rset.rows)
   294         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   297         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   348         rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?',
   351         rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?',
   349                             {'x':peid1}, 'x')
   352                             {'x':peid1}, 'x')
   350         self.assertEquals(rset.rows, [[peid1]])
   353         self.assertEquals(rset.rows, [[peid1]])
   351 
   354 
   352     def test_select_left_outer_join(self):
   355     def test_select_left_outer_join(self):
   353         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G "
   356         rset = self.execute('DISTINCT Any G WHERE U? in_group G')
   354                             "WHERE G name 'users'")[0][0]
   357         self.assertEquals(len(rset), 4)
   355         self.commit()
   358         rset = self.execute('DISTINCT Any G WHERE U? in_group G, U eid %(x)s',
   356         try:
   359                             {'x': self.session.user.eid}, 'x')
   357             rset = self.execute('Any FS,TS,C,D,U ORDERBY D DESC '
   360         self.assertEquals(len(rset), 4)
   358                                 'WHERE WF wf_info_for X,'
       
   359                                 'WF from_state FS?, WF to_state TS, WF comment C,'
       
   360                                 'WF creation_date D, WF owned_by U, X eid %(x)s',
       
   361                                 {'x': ueid}, 'x')
       
   362             self.assertEquals(len(rset), 1)
       
   363             self.execute('SET X in_state S WHERE X eid %(x)s, S name "deactivated"',
       
   364                          {'x': ueid}, 'x')
       
   365             rset = self.execute('Any FS,TS,C,D,U ORDERBY D DESC '
       
   366                                 'WHERE WF wf_info_for X,'
       
   367                                 'WF from_state FS?, WF to_state TS, WF comment C,'
       
   368                                 'WF creation_date D, WF owned_by U, X eid %(x)s',
       
   369                                 {'x': ueid}, 'x')
       
   370             self.assertEquals(len(rset), 2)
       
   371         finally:
       
   372             self.execute('DELETE CWUser X WHERE X eid %s' % ueid)
       
   373             self.commit()
       
   374 
   361 
   375     def test_select_ambigous_outer_join(self):
   362     def test_select_ambigous_outer_join(self):
   376         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
   363         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
   377         self.execute("INSERT Tag X: X name 'tagbis'")[0][0]
   364         self.execute("INSERT Tag X: X name 'tagbis'")[0][0]
   378         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
   365         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
   464     def test_select_aggregat_having(self):
   451     def test_select_aggregat_having(self):
   465         rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N '
   452         rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N '
   466                             'WHERE RT name N, RDEF relation_type RT '
   453                             'WHERE RT name N, RDEF relation_type RT '
   467                             'HAVING COUNT(RDEF) > 10')
   454                             'HAVING COUNT(RDEF) > 10')
   468         self.assertListEquals(rset.rows,
   455         self.assertListEquals(rset.rows,
   469                               [[u'description', 11],
   456                               [[u'description_format', 13],
   470                                [u'name', 13], [u'created_by', 34],
   457                                [u'description', 14],
   471                                [u'creation_date', 34], [u'cwuri', 34],
   458                                [u'name', 16],
   472                                ['in_basket', 34],
   459                                [u'created_by', 38],
   473                                [u'is', 34], [u'is_instance_of', 34],
   460                                [u'creation_date', 38],
   474                                [u'modification_date', 34], [u'owned_by', 34]])
   461                                [u'cwuri', 38],
       
   462                                [u'in_basket', 38],
       
   463                                [u'is', 38],
       
   464                                [u'is_instance_of', 38],
       
   465                                [u'modification_date', 38],
       
   466                                [u'owned_by', 38]])
   475 
   467 
   476     def test_select_aggregat_having_dumb(self):
   468     def test_select_aggregat_having_dumb(self):
   477         # dumb but should not raise an error
   469         # dumb but should not raise an error
   478         rset = self.execute('Any U,COUNT(X) GROUPBY U '
   470         rset = self.execute('Any U,COUNT(X) GROUPBY U '
   479                             'WHERE U eid %(x)s, X owned_by U '
   471                             'WHERE U eid %(x)s, X owned_by U '
   719                               ('String', 'String', 'String', 'String',)))
   711                               ('String', 'String', 'String', 'String',)))
   720         rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN')
   712         rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN')
   721 
   713 
   722     def test_select_union(self):
   714     def test_select_union(self):
   723         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   715         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   724                             '((Any X,N WHERE X name N, X transition_of E, E name %(name)s)'
   716                             '((Any X,N WHERE X name N, X transition_of WF, WF workflow_of E, E name %(name)s)'
   725                             ' UNION '
   717                             ' UNION '
   726                             '(Any X,N WHERE X name N, X state_of E, E name %(name)s))',
   718                             '(Any X,N WHERE X name N, X state_of WF, WF workflow_of E, E name %(name)s))',
   727                             {'name': 'CWUser'})
   719                             {'name': 'CWUser'})
   728         self.assertEquals([x[1] for x in rset.rows],
   720         self.assertEquals([x[1] for x in rset.rows],
   729                           ['activate', 'activated', 'deactivate', 'deactivated'])
   721                           ['activate', 'activated', 'deactivate', 'deactivated'])
   730         self.assertEquals(rset.description,
   722         self.assertEquals(rset.description,
   731                           [('Transition', 'String'), ('State', 'String'),
   723                           [('Transition', 'String'), ('State', 'String'),
   993         self.assertEquals(rset.rows, [])
   985         self.assertEquals(rset.rows, [])
   994 
   986 
   995     # update queries tests ####################################################
   987     # update queries tests ####################################################
   996 
   988 
   997     def test_update_1(self):
   989     def test_update_1(self):
   998         self.execute("INSERT Personne Y: Y nom 'toto'")
   990         peid = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0]
   999         rset = self.execute('Personne X WHERE X nom "toto"')
   991         rset = self.execute('Personne X WHERE X nom "toto"')
  1000         self.assertEqual(len(rset.rows), 1)
   992         self.assertEqual(len(rset.rows), 1)
  1001         self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'")
   993         self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'")
  1002         rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z')
   994         rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z')
  1003         self.assertEqual(tuplify(rset.rows), [('tutu', 'original')])
   995         self.assertEqual(tuplify(rset.rows), [('tutu', 'original')])
  1128 
  1120 
  1129     def test_nonregr_3(self):
  1121     def test_nonregr_3(self):
  1130         """bad sql generated on the second query (destination_state is not
  1122         """bad sql generated on the second query (destination_state is not
  1131         detected as an inlined relation)
  1123         detected as an inlined relation)
  1132         """
  1124         """
  1133         rset = self.execute('Any S,ES,T WHERE S state_of ET, ET name "CWUser",'
  1125         rset = self.execute('Any S,ES,T WHERE S state_of WF, WF workflow_of ET, ET name "CWUser",'
  1134                              'ES allowed_transition T, T destination_state S')
  1126                              'ES allowed_transition T, T destination_state S')
  1135         self.assertEquals(len(rset.rows), 2)
  1127         self.assertEquals(len(rset.rows), 2)
  1136 
  1128 
  1137     def test_nonregr_4(self):
  1129     def test_nonregr_4(self):
  1138         # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3
  1130         # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3
  1258         # huum, psycopg specific
  1250         # huum, psycopg specific
  1259         self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()})
  1251         self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()})
  1260 
  1252 
  1261     def test_nonregr_set_query(self):
  1253     def test_nonregr_set_query(self):
  1262         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0]
  1254         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0]
  1263         self.execute("SET E in_group G, E in_state S, "
  1255         self.execute("SET E in_group G, E firstname %(firstname)s, E surname %(surname)s "
  1264                       "E firstname %(firstname)s, E surname %(surname)s "
  1256                       "WHERE E eid %(x)s, G name 'users'",
  1265                       "WHERE E eid %(x)s, G name 'users', S name 'activated'",
       
  1266                       {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x')
  1257                       {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x')
  1267 
  1258 
  1268     def test_nonregr_u_owned_by_u(self):
  1259     def test_nonregr_u_owned_by_u(self):
  1269         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G "
  1260         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G "
  1270                              "WHERE G name 'users'")[0][0]
  1261                              "WHERE G name 'users'")[0][0]