server/test/unittest_querier.py
branchtls-sprint
changeset 1263 01152fffd593
parent 1016 26387b836099
parent 1251 af40e615dc89
child 1398 5fe84a5f7035
equal deleted inserted replaced
1246:76b3cd5d4f31 1263:01152fffd593
     5 
     5 
     6 from logilab.common.testlib import TestCase, unittest_main
     6 from logilab.common.testlib import TestCase, unittest_main
     7 from rql import BadRQLQuery, RQLSyntaxError
     7 from rql import BadRQLQuery, RQLSyntaxError
     8 
     8 
     9 from cubicweb import QueryError, Unauthorized
     9 from cubicweb import QueryError, Unauthorized
       
    10 from cubicweb.server.sqlutils import SQL_PREFIX
    10 from cubicweb.server.utils import crypt_password
    11 from cubicweb.server.utils import crypt_password
    11 from cubicweb.server.sources.native import make_schema
    12 from cubicweb.server.sources.native import make_schema
    12 from cubicweb.devtools import init_test_database
    13 from cubicweb.devtools import init_test_database
    13 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC
    14 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC
    14 
    15 
   467                             'HAVING COUNT(X) > 10', {'x': self.ueid})
   468                             'HAVING COUNT(X) > 10', {'x': self.ueid})
   468         self.assertEquals(len(rset.rows), 1)
   469         self.assertEquals(len(rset.rows), 1)
   469         self.assertEquals(rset.rows[0][0], self.ueid)
   470         self.assertEquals(rset.rows[0][0], self.ueid)
   470 
   471 
   471     def test_select_complex_sort(self):
   472     def test_select_complex_sort(self):
       
   473         self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed')
   472         rset = self.execute('Any X ORDERBY X,D LIMIT 5 WHERE X creation_date D')
   474         rset = self.execute('Any X ORDERBY X,D LIMIT 5 WHERE X creation_date D')
   473         result = rset.rows
   475         result = rset.rows
   474         result.sort()
   476         result.sort()
   475         self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,), (5,)])
   477         self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,), (5,)])
   476         
   478         
  1067         self.assertEquals(len(rset.rows), 1)
  1069         self.assertEquals(len(rset.rows), 1)
  1068         self.assertEquals(rset.description, [('EUser',)])
  1070         self.assertEquals(rset.description, [('EUser',)])
  1069         self.assertRaises(Unauthorized,
  1071         self.assertRaises(Unauthorized,
  1070                           self.execute, "Any P WHERE X is EUser, X login 'bob', X upassword P")
  1072                           self.execute, "Any P WHERE X is EUser, X login 'bob', X upassword P")
  1071         cursor = self.pool['system']
  1073         cursor = self.pool['system']
  1072         cursor.execute("SELECT upassword from EUser WHERE login='bob'")
  1074         cursor.execute("SELECT %supassword from %sEUser WHERE %slogin='bob'"
       
  1075                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1073         passwd = cursor.fetchone()[0].getvalue()
  1076         passwd = cursor.fetchone()[0].getvalue()
  1074         self.assertEquals(passwd, crypt_password('toto', passwd[:2])) 
  1077         self.assertEquals(passwd, crypt_password('toto', passwd[:2])) 
  1075         rset = self.execute("Any X WHERE X is EUser, X login 'bob', X upassword '%s'" % passwd)
  1078         rset = self.execute("Any X WHERE X is EUser, X login 'bob', X upassword '%s'" % passwd)
  1076         self.assertEquals(len(rset.rows), 1)
  1079         self.assertEquals(len(rset.rows), 1)
  1077         self.assertEquals(rset.description, [('EUser',)])
  1080         self.assertEquals(rset.description, [('EUser',)])
  1080         cursor = self.pool['system']
  1083         cursor = self.pool['system']
  1081         rset = self.execute("INSERT EUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'})
  1084         rset = self.execute("INSERT EUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'})
  1082         self.assertEquals(rset.description[0][0], 'EUser')
  1085         self.assertEquals(rset.description[0][0], 'EUser')
  1083         rset = self.execute("SET X upassword %(pwd)s WHERE X is EUser, X login 'bob'",
  1086         rset = self.execute("SET X upassword %(pwd)s WHERE X is EUser, X login 'bob'",
  1084                             {'pwd': 'tutu'})
  1087                             {'pwd': 'tutu'})
  1085         cursor.execute("SELECT upassword from EUser WHERE login='bob'")
  1088         cursor.execute("SELECT %supassword from %sEUser WHERE %slogin='bob'"
       
  1089                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1086         passwd = cursor.fetchone()[0].getvalue()
  1090         passwd = cursor.fetchone()[0].getvalue()
  1087         self.assertEquals(passwd, crypt_password('tutu', passwd[:2])) 
  1091         self.assertEquals(passwd, crypt_password('tutu', passwd[:2])) 
  1088         rset = self.execute("Any X WHERE X is EUser, X login 'bob', X upassword '%s'" % passwd)
  1092         rset = self.execute("Any X WHERE X is EUser, X login 'bob', X upassword '%s'" % passwd)
  1089         self.assertEquals(len(rset.rows), 1)
  1093         self.assertEquals(len(rset.rows), 1)
  1090         self.assertEquals(rset.description, [('EUser',)])
  1094         self.assertEquals(rset.description, [('EUser',)])
  1207         """Error: Variable has no attribute 'sql' in rql2sql.py (visit_variable)
  1211         """Error: Variable has no attribute 'sql' in rql2sql.py (visit_variable)
  1208 
  1212 
  1209         cause: old variable ref inserted into a fresh rqlst copy
  1213         cause: old variable ref inserted into a fresh rqlst copy
  1210         (in RQLSpliter._complex_select_plan)
  1214         (in RQLSpliter._complex_select_plan)
  1211         """
  1215         """
       
  1216         self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed')
  1212         self.execute('Any X ORDERBY D DESC WHERE X creation_date D')
  1217         self.execute('Any X ORDERBY D DESC WHERE X creation_date D')
  1213     
  1218     
  1214     def test_nonregr_extra_joins(self):
  1219     def test_nonregr_extra_joins(self):
  1215         ueid = self.session.user.eid
  1220         ueid = self.session.user.eid
  1216         teid1 = self.execute("INSERT Folder X: X name 'folder1'")[0][0]
  1221         teid1 = self.execute("INSERT Folder X: X name 'folder1'")[0][0]