server/test/unittest_querier.py
changeset 7791 31bb51ea5485
parent 7713 467c52ff9557
child 7907 08320697ca1a
equal deleted inserted replaced
7790:7e16e056eecb 7791:31bb51ea5485
   248         self.assertRaises(RQLSyntaxError, self.execute,
   248         self.assertRaises(RQLSyntaxError, self.execute,
   249                           'Any X WHERE X is CWRType, X name "öwned_by"')
   249                           'Any X WHERE X is CWRType, X name "öwned_by"')
   250 
   250 
   251     def test_unknown_eid(self):
   251     def test_unknown_eid(self):
   252         # should return an empty result set
   252         # should return an empty result set
   253         self.failIf(self.execute('Any X WHERE X eid 99999999'))
   253         self.assertFalse(self.execute('Any X WHERE X eid 99999999'))
   254 
   254 
   255     def test_typed_eid(self):
   255     def test_typed_eid(self):
   256         # should return an empty result set
   256         # should return an empty result set
   257         rset = self.execute('Any X WHERE X eid %(x)s', {'x': '1'})
   257         rset = self.execute('Any X WHERE X eid %(x)s', {'x': '1'})
   258         self.assertIsInstance(rset[0][0], (int, long))
   258         self.assertIsInstance(rset[0][0], (int, long))
   416         self.execute("INSERT Tag X: X name 'tagbis'")[0][0]
   416         self.execute("INSERT Tag X: X name 'tagbis'")[0][0]
   417         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
   417         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
   418         self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
   418         self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
   419                      {'g': geid, 't': teid})
   419                      {'g': geid, 't': teid})
   420         rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN")
   420         rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN")
   421         self.failUnless(['users', 'tag'] in rset.rows)
   421         self.assertTrue(['users', 'tag'] in rset.rows)
   422         self.failUnless(['activated', None] in rset.rows)
   422         self.assertTrue(['activated', None] in rset.rows)
   423         rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN")
   423         rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN")
   424         self.assertEqual(rset.rows, [[None, 'tagbis'], ['users', 'tag']])
   424         self.assertEqual(rset.rows, [[None, 'tagbis'], ['users', 'tag']])
   425 
   425 
   426     def test_select_not_inline_rel(self):
   426     def test_select_not_inline_rel(self):
   427         self.execute("INSERT Personne X: X nom 'bidule'")
   427         self.execute("INSERT Personne X: X nom 'bidule'")
   492         self.assertEqual(len(rset.rows[0]), 1)
   492         self.assertEqual(len(rset.rows[0]), 1)
   493         self.assertEqual(rset.description, [('Int',)])
   493         self.assertEqual(rset.description, [('Int',)])
   494 
   494 
   495     def test_select_custom_aggregat_concat_string(self):
   495     def test_select_custom_aggregat_concat_string(self):
   496         rset = self.execute('Any GROUP_CONCAT(N) WHERE X is CWGroup, X name N')
   496         rset = self.execute('Any GROUP_CONCAT(N) WHERE X is CWGroup, X name N')
   497         self.failUnless(rset)
   497         self.assertTrue(rset)
   498         self.failUnlessEqual(sorted(rset[0][0].split(', ')), ['guests', 'managers',
   498         self.assertEqual(sorted(rset[0][0].split(', ')), ['guests', 'managers',
   499                                                              'owners', 'users'])
   499                                                              'owners', 'users'])
   500 
   500 
   501     def test_select_custom_regproc_limit_size(self):
   501     def test_select_custom_regproc_limit_size(self):
   502         rset = self.execute('Any TEXT_LIMIT_SIZE(N, 3) WHERE X is CWGroup, X name N, X name "managers"')
   502         rset = self.execute('Any TEXT_LIMIT_SIZE(N, 3) WHERE X is CWGroup, X name N, X name "managers"')
   503         self.failUnless(rset)
   503         self.assertTrue(rset)
   504         self.failUnlessEqual(rset[0][0], 'man...')
   504         self.assertEqual(rset[0][0], 'man...')
   505         self.execute("INSERT Basket X: X name 'bidule', X description '<b>hop hop</b>', X description_format 'text/html'")
   505         self.execute("INSERT Basket X: X name 'bidule', X description '<b>hop hop</b>', X description_format 'text/html'")
   506         rset = self.execute('Any LIMIT_SIZE(D, DF, 3) WHERE X is Basket, X description D, X description_format DF')
   506         rset = self.execute('Any LIMIT_SIZE(D, DF, 3) WHERE X is Basket, X description D, X description_format DF')
   507         self.failUnless(rset)
   507         self.assertTrue(rset)
   508         self.failUnlessEqual(rset[0][0], 'hop...')
   508         self.assertEqual(rset[0][0], 'hop...')
   509 
   509 
   510     def test_select_regproc_orderby(self):
   510     def test_select_regproc_orderby(self):
   511         rset = self.execute('DISTINCT Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, X name "managers"')
   511         rset = self.execute('DISTINCT Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, X name "managers"')
   512         self.failUnlessEqual(len(rset), 1)
   512         self.assertEqual(len(rset), 1)
   513         self.failUnlessEqual(rset[0][1], 'managers')
   513         self.assertEqual(rset[0][1], 'managers')
   514         rset = self.execute('Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, NOT U in_group X, U login "admin"')
   514         rset = self.execute('Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, NOT U in_group X, U login "admin"')
   515         self.failUnlessEqual(len(rset), 3)
   515         self.assertEqual(len(rset), 3)
   516         self.failUnlessEqual(rset[0][1], 'owners')
   516         self.assertEqual(rset[0][1], 'owners')
   517 
   517 
   518     def test_select_aggregat_sort(self):
   518     def test_select_aggregat_sort(self):
   519         rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G')
   519         rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G')
   520         self.assertEqual(len(rset.rows), 2)
   520         self.assertEqual(len(rset.rows), 2)
   521         self.assertEqual(len(rset.rows[0]), 2)
   521         self.assertEqual(len(rset.rows[0]), 2)
   617         self.assertEqual(len(rset.rows), 2, rset.rows)
   617         self.assertEqual(len(rset.rows), 2, rset.rows)
   618         rset = self.execute(u'Any N where N has_text "bidüle"')
   618         rset = self.execute(u'Any N where N has_text "bidüle"')
   619         self.assertEqual(len(rset.rows), 2, rset.rows)
   619         self.assertEqual(len(rset.rows), 2, rset.rows)
   620         biduleeids = [r[0] for r in rset.rows]
   620         biduleeids = [r[0] for r in rset.rows]
   621         rset = self.execute(u'Any N where NOT N has_text "bidüle"')
   621         rset = self.execute(u'Any N where NOT N has_text "bidüle"')
   622         self.failIf([r[0] for r in rset.rows if r[0] in biduleeids])
   622         self.assertFalse([r[0] for r in rset.rows if r[0] in biduleeids])
   623         # duh?
   623         # duh?
   624         rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'})
   624         rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'})
   625 
   625 
   626     def test_select_searchable_text_2(self):
   626     def test_select_searchable_text_2(self):
   627         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   627         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   755         rset = self.execute('Any U WHERE U in_group G, G name "guests" OR G name "managers"')
   755         rset = self.execute('Any U WHERE U in_group G, G name "guests" OR G name "managers"')
   756         self.assertEqual(len(rset.rows), 2)
   756         self.assertEqual(len(rset.rows), 2)
   757 
   757 
   758     def test_select_explicit_eid(self):
   758     def test_select_explicit_eid(self):
   759         rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid})
   759         rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid})
   760         self.failUnless(rset)
   760         self.assertTrue(rset)
   761         self.assertEqual(rset.description[0][1], 'Int')
   761         self.assertEqual(rset.description[0][1], 'Int')
   762 
   762 
   763 #     def test_select_rewritten_optional(self):
   763 #     def test_select_rewritten_optional(self):
   764 #         eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
   764 #         eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
   765 #         rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)',
   765 #         rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)',
   772         rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D")
   772         rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D")
   773         self.assert_(isinstance(rset.rows[0][0], datetime), rset.rows)
   773         self.assert_(isinstance(rset.rows[0][0], datetime), rset.rows)
   774         rset = self.execute('Tag X WHERE X creation_date TODAY')
   774         rset = self.execute('Tag X WHERE X creation_date TODAY')
   775         self.assertEqual(len(rset.rows), 2)
   775         self.assertEqual(len(rset.rows), 2)
   776         rset = self.execute('Any MAX(D) WHERE X is Tag, X creation_date D')
   776         rset = self.execute('Any MAX(D) WHERE X is Tag, X creation_date D')
   777         self.failUnless(isinstance(rset[0][0], datetime), (rset[0][0], type(rset[0][0])))
   777         self.assertTrue(isinstance(rset[0][0], datetime), (rset[0][0], type(rset[0][0])))
   778 
   778 
   779     def test_today(self):
   779     def test_today(self):
   780         self.execute("INSERT Tag X: X name 'bidule', X creation_date TODAY")
   780         self.execute("INSERT Tag X: X name 'bidule', X creation_date TODAY")
   781         self.execute("INSERT Tag Y: Y name 'toto'")
   781         self.execute("INSERT Tag Y: Y name 'toto'")
   782         rset = self.execute('Tag X WHERE X creation_date TODAY')
   782         rset = self.execute('Tag X WHERE X creation_date TODAY')
   889         login = self.execute('Any L WHERE X login "admin", X identity Y, Y login L')[0][0]
   889         login = self.execute('Any L WHERE X login "admin", X identity Y, Y login L')[0][0]
   890         self.assertEqual(login, 'admin')
   890         self.assertEqual(login, 'admin')
   891 
   891 
   892     def test_select_date_mathexp(self):
   892     def test_select_date_mathexp(self):
   893         rset = self.execute('Any X, TODAY - CD WHERE X is CWUser, X creation_date CD')
   893         rset = self.execute('Any X, TODAY - CD WHERE X is CWUser, X creation_date CD')
   894         self.failUnless(rset)
   894         self.assertTrue(rset)
   895         self.failUnlessEqual(rset.description[0][1], 'Interval')
   895         self.assertEqual(rset.description[0][1], 'Interval')
   896         eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0]
   896         eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0]
   897         rset = self.execute('Any X, NOW - CD WHERE X is Personne, X creation_date CD')
   897         rset = self.execute('Any X, NOW - CD WHERE X is Personne, X creation_date CD')
   898         self.failUnlessEqual(rset.description[0][1], 'Interval')
   898         self.assertEqual(rset.description[0][1], 'Interval')
   899 
   899 
   900     def test_select_subquery_aggregat_1(self):
   900     def test_select_subquery_aggregat_1(self):
   901         # percent users by groups
   901         # percent users by groups
   902         self.execute('SET X in_group G WHERE G name "users"')
   902         self.execute('SET X in_group G WHERE G name "users"')
   903         rset = self.execute('Any GN, COUNT(X)*100/T GROUPBY GN ORDERBY 2,1'
   903         rset = self.execute('Any GN, COUNT(X)*100/T GROUPBY GN ORDERBY 2,1'
  1171         self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s",
  1171         self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s",
  1172                       {'x': str(eid1), 'y': str(eid2)})
  1172                       {'x': str(eid1), 'y': str(eid2)})
  1173         rset = self.execute('Any X, Y WHERE X travaille Y')
  1173         rset = self.execute('Any X, Y WHERE X travaille Y')
  1174         self.assertEqual(len(rset.rows), 1)
  1174         self.assertEqual(len(rset.rows), 1)
  1175         # test add of an existant relation but with NOT X rel Y protection
  1175         # test add of an existant relation but with NOT X rel Y protection
  1176         self.failIf(self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s,"
  1176         self.assertFalse(self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s,"
  1177                                  "NOT X travaille Y",
  1177                                  "NOT X travaille Y",
  1178                                  {'x': str(eid1), 'y': str(eid2)}))
  1178                                  {'x': str(eid1), 'y': str(eid2)}))
  1179 
  1179 
  1180     def test_update_2ter(self):
  1180     def test_update_2ter(self):
  1181         rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
  1181         rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
  1196         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0]
  1196         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0]
  1197         peid1 = self.execute("INSERT Personne Y: Y nom 'turlu'")[0][0]
  1197         peid1 = self.execute("INSERT Personne Y: Y nom 'turlu'")[0][0]
  1198         peid2 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1198         peid2 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1199         self.execute('SET P1 owned_by U, P2 owned_by U '
  1199         self.execute('SET P1 owned_by U, P2 owned_by U '
  1200                      'WHERE P1 eid %s, P2 eid %s, U eid %s' % (peid1, peid2, ueid))
  1200                      'WHERE P1 eid %s, P2 eid %s, U eid %s' % (peid1, peid2, ueid))
  1201         self.failUnless(self.execute('Any X WHERE X eid %s, X owned_by U, U eid %s'
  1201         self.assertTrue(self.execute('Any X WHERE X eid %s, X owned_by U, U eid %s'
  1202                                        % (peid1, ueid)))
  1202                                        % (peid1, ueid)))
  1203         self.failUnless(self.execute('Any X WHERE X eid %s, X owned_by U, U eid %s'
  1203         self.assertTrue(self.execute('Any X WHERE X eid %s, X owned_by U, U eid %s'
  1204                                        % (peid2, ueid)))
  1204                                        % (peid2, ueid)))
  1205 
  1205 
  1206     def test_update_math_expr(self):
  1206     def test_update_math_expr(self):
  1207         orders = [r[0] for r in self.execute('Any O ORDERBY O WHERE ST name "Personne", X from_entity ST, X ordernum O')]
  1207         orders = [r[0] for r in self.execute('Any O ORDERBY O WHERE ST name "Personne", X from_entity ST, X ordernum O')]
  1208         for i,v in enumerate(orders):
  1208         for i,v in enumerate(orders):