server/test/unittest_querier.py
changeset 1808 aa09e20dd8c0
parent 1787 71c143c0ada3
child 1949 9064102b4deb
equal deleted inserted replaced
1693:49075f57cf2c 1808:aa09e20dd8c0
     1 # -*- coding: iso-8859-1 -*-
     1 # -*- coding: iso-8859-1 -*-
     2 """unit tests for modules cubicweb.server.querier and cubicweb.server.querier_steps
     2 """unit tests for modules cubicweb.server.querier and cubicweb.server.querier_steps
     3 """
     3 """
       
     4 from datetime import date, datetime
     4 
     5 
     5 from logilab.common.testlib import TestCase, unittest_main
     6 from logilab.common.testlib import TestCase, unittest_main
     6 from cubicweb.devtools import init_test_database
       
     7 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC
       
     8 from unittest_session import Variable
       
     9 
       
    10 from mx.DateTime import today, now, DateTimeType
       
    11 from rql import BadRQLQuery, RQLSyntaxError
     7 from rql import BadRQLQuery, RQLSyntaxError
    12 
     8 
    13 from cubicweb import QueryError, Unauthorized
     9 from cubicweb import QueryError, Unauthorized
    14 from cubicweb.server.sqlutils import SQL_PREFIX
    10 from cubicweb.server.sqlutils import SQL_PREFIX
    15 from cubicweb.server.utils import crypt_password
    11 from cubicweb.server.utils import crypt_password
    16 from cubicweb.server.sources.native import make_schema
    12 from cubicweb.server.sources.native import make_schema
       
    13 from cubicweb.devtools import init_test_database
       
    14 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC
       
    15 
       
    16 from unittest_session import Variable
    17 
    17 
    18 
    18 
    19 # register priority/severity sorting registered procedure
    19 # register priority/severity sorting registered procedure
    20 from rql.utils import register_function, FunctionDescr
    20 from rql.utils import register_function, FunctionDescr
    21 
    21 
    37 from logilab.common.adbh import _GenericAdvFuncHelper
    37 from logilab.common.adbh import _GenericAdvFuncHelper
    38 TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING
    38 TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING
    39 
    39 
    40 class MakeSchemaTC(TestCase):
    40 class MakeSchemaTC(TestCase):
    41     def test_known_values(self):
    41     def test_known_values(self):
    42         solution = {'A': 'String', 'B': 'EUser'}
    42         solution = {'A': 'String', 'B': 'CWUser'}
    43         self.assertEquals(make_schema((Variable('A'), Variable('B')), solution, 
    43         self.assertEquals(make_schema((Variable('A'), Variable('B')), solution,
    44                                       'table0', TYPEMAP),
    44                                       'table0', TYPEMAP),
    45                           ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'}))
    45                           ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'}))
    46         
    46 
    47 
    47 
    48 repo, cnx = init_test_database('sqlite')
    48 repo, cnx = init_test_database('sqlite')
    49 
    49 
    50 
    50 
    51 
    51 
    52 class UtilsTC(BaseQuerierTC):
    52 class UtilsTC(BaseQuerierTC):
    53     repo = repo
    53     repo = repo
    54     
    54 
    55     def get_max_eid(self):
    55     def get_max_eid(self):
    56         # no need for cleanup here
    56         # no need for cleanup here
    57         return None
    57         return None
    58     def cleanup(self):
    58     def cleanup(self):
    59         # no need for cleanup here
    59         # no need for cleanup here
    60         pass
    60         pass
    61     
    61 
    62     def test_preprocess_1(self):
    62     def test_preprocess_1(self):
    63         reid = self.execute('Any X WHERE X is ERType, X name "owned_by"')[0][0]
    63         reid = self.execute('Any X WHERE X is CWRType, X name "owned_by"')[0][0]
    64         rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid})
    64         rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid})
    65         self.assertEquals(rqlst.solutions, [{'RDEF': 'EFRDef'}, {'RDEF': 'ENFRDef'}])
    65         self.assertEquals(rqlst.solutions, [{'RDEF': 'CWAttribute'}, {'RDEF': 'CWRelation'}])
    66         
    66 
    67     def test_preprocess_2(self):
    67     def test_preprocess_2(self):
    68         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
    68         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
    69         #geid = self.execute("EGroup G WHERE G name 'users'")[0][0]
    69         #geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
    70         #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
    70         #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
    71         #             {'g': geid, 't': teid}, 'g')
    71         #             {'g': geid, 't': teid}, 'g')
    72         rqlst = self._prepare('Any X WHERE E eid %(x)s, E tags X', {'x': teid})
    72         rqlst = self._prepare('Any X WHERE E eid %(x)s, E tags X', {'x': teid})
    73         # the query may be optimized, should keep only one solution
    73         # the query may be optimized, should keep only one solution
    74         # (any one, etype will be discarded)
    74         # (any one, etype will be discarded)
    75         self.assertEquals(len(rqlst.solutions), 1)
    75         self.assertEquals(len(rqlst.solutions), 1)
    76         
    76 
    77     def test_preprocess_security(self):
    77     def test_preprocess_security(self):
    78         plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN '
    78         plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN '
    79                                   'WHERE X is ET, ET name ETN')
    79                                   'WHERE X is ET, ET name ETN')
    80         plan.session = self._user_session(('users',))[1]
    80         plan.session = self._user_session(('users',))[1]
    81         union = plan.rqlst
    81         union = plan.rqlst
    94                           'Any ETN,X WHERE X is ET, ET name ETN, (EXISTS(X owned_by %(B)s))'
    94                           'Any ETN,X WHERE X is ET, ET name ETN, (EXISTS(X owned_by %(B)s))'
    95                           ' OR ((((EXISTS(D concerne C?, C owned_by %(B)s, X identity D, C is Division, D is Affaire))'
    95                           ' OR ((((EXISTS(D concerne C?, C owned_by %(B)s, X identity D, C is Division, D is Affaire))'
    96                           ' OR (EXISTS(H concerne G?, G owned_by %(B)s, G is SubDivision, X identity H, H is Affaire)))'
    96                           ' OR (EXISTS(H concerne G?, G owned_by %(B)s, G is SubDivision, X identity H, H is Affaire)))'
    97                           ' OR (EXISTS(I concerne F?, F owned_by %(B)s, F is Societe, X identity I, I is Affaire)))'
    97                           ' OR (EXISTS(I concerne F?, F owned_by %(B)s, F is Societe, X identity I, I is Affaire)))'
    98                           ' OR (EXISTS(J concerne E?, E owned_by %(B)s, E is Note, X identity J, J is Affaire)))'
    98                           ' OR (EXISTS(J concerne E?, E owned_by %(B)s, E is Note, X identity J, J is Affaire)))'
    99                           ', ET is EEType, X is Affaire')
    99                           ', ET is CWEType, X is Affaire')
   100         self.assertEquals(solutions, [{'C': 'Division',
   100         self.assertEquals(solutions, [{'C': 'Division',
   101                                        'D': 'Affaire',
   101                                        'D': 'Affaire',
   102                                        'E': 'Note',
   102                                        'E': 'Note',
   103                                        'F': 'Societe',
   103                                        'F': 'Societe',
   104                                        'G': 'SubDivision',
   104                                        'G': 'SubDivision',
   105                                        'H': 'Affaire',
   105                                        'H': 'Affaire',
   106                                        'I': 'Affaire',
   106                                        'I': 'Affaire',
   107                                        'J': 'Affaire',
   107                                        'J': 'Affaire',
   108                                        'X': 'Affaire',
   108                                        'X': 'Affaire',
   109                                        'ET': 'EEType', 'ETN': 'String'}])
   109                                        'ET': 'CWEType', 'ETN': 'String'}])
   110         rql, solutions = partrqls[1]
   110         rql, solutions = partrqls[1]
   111         self.assertEquals(rql,  'Any ETN,X WHERE X is ET, ET name ETN, ET is EEType, '
   111         self.assertEquals(rql,  'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, '
   112                           'X is IN(Bookmark, Card, Comment, Division, ECache, EConstraint, EConstraintType, EEType, EFRDef, EGroup, ENFRDef, EPermission, EProperty, ERType, EUser, Email, EmailAddress, EmailPart, EmailThread, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, Tag, TrInfo, Transition)')
   112                           'X is IN(Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, Tag, TrInfo, Transition)')
   113         self.assertListEquals(sorted(solutions),
   113         self.assertListEquals(sorted(solutions),
   114                               sorted([{'X': 'Bookmark', 'ETN': 'String', 'ET': 'EEType'},
   114                               sorted([{'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'},
   115                                       {'X': 'Card', 'ETN': 'String', 'ET': 'EEType'},
   115                                       {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'},
   116                                       {'X': 'Comment', 'ETN': 'String', 'ET': 'EEType'},
   116                                       {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'},
   117                                       {'X': 'Division', 'ETN': 'String', 'ET': 'EEType'},
   117                                       {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'},
   118                                       {'X': 'ECache', 'ETN': 'String', 'ET': 'EEType'},
   118                                       {'X': 'CWCache', 'ETN': 'String', 'ET': 'CWEType'},
   119                                       {'X': 'EConstraint', 'ETN': 'String', 'ET': 'EEType'},
   119                                       {'X': 'CWConstraint', 'ETN': 'String', 'ET': 'CWEType'},
   120                                       {'X': 'EConstraintType', 'ETN': 'String', 'ET': 'EEType'},
   120                                       {'X': 'CWConstraintType', 'ETN': 'String', 'ET': 'CWEType'},
   121                                       {'X': 'EEType', 'ETN': 'String', 'ET': 'EEType'},
   121                                       {'X': 'CWEType', 'ETN': 'String', 'ET': 'CWEType'},
   122                                       {'X': 'EFRDef', 'ETN': 'String', 'ET': 'EEType'},
   122                                       {'X': 'CWAttribute', 'ETN': 'String', 'ET': 'CWEType'},
   123                                       {'X': 'EGroup', 'ETN': 'String', 'ET': 'EEType'},
   123                                       {'X': 'CWGroup', 'ETN': 'String', 'ET': 'CWEType'},
   124                                       {'X': 'Email', 'ETN': 'String', 'ET': 'EEType'},
   124                                       {'X': 'Email', 'ETN': 'String', 'ET': 'CWEType'},
   125                                       {'X': 'EmailAddress', 'ETN': 'String', 'ET': 'EEType'},
   125                                       {'X': 'EmailAddress', 'ETN': 'String', 'ET': 'CWEType'},
   126                                       {'X': 'EmailPart', 'ETN': 'String', 'ET': 'EEType'},
   126                                       {'X': 'EmailPart', 'ETN': 'String', 'ET': 'CWEType'},
   127                                       {'X': 'EmailThread', 'ETN': 'String', 'ET': 'EEType'},
   127                                       {'X': 'EmailThread', 'ETN': 'String', 'ET': 'CWEType'},
   128                                       {'X': 'ENFRDef', 'ETN': 'String', 'ET': 'EEType'},
   128                                       {'X': 'CWRelation', 'ETN': 'String', 'ET': 'CWEType'},
   129                                       {'X': 'EPermission', 'ETN': 'String', 'ET': 'EEType'},
   129                                       {'X': 'CWPermission', 'ETN': 'String', 'ET': 'CWEType'},
   130                                       {'X': 'EProperty', 'ETN': 'String', 'ET': 'EEType'},
   130                                       {'X': 'CWProperty', 'ETN': 'String', 'ET': 'CWEType'},
   131                                       {'X': 'ERType', 'ETN': 'String', 'ET': 'EEType'},
   131                                       {'X': 'CWRType', 'ETN': 'String', 'ET': 'CWEType'},
   132                                       {'X': 'EUser', 'ETN': 'String', 'ET': 'EEType'},
   132                                       {'X': 'CWUser', 'ETN': 'String', 'ET': 'CWEType'},
   133                                       {'X': 'File', 'ETN': 'String', 'ET': 'EEType'},
   133                                       {'X': 'File', 'ETN': 'String', 'ET': 'CWEType'},
   134                                       {'X': 'Folder', 'ETN': 'String', 'ET': 'EEType'},
   134                                       {'X': 'Folder', 'ETN': 'String', 'ET': 'CWEType'},
   135                                       {'X': 'Image', 'ETN': 'String', 'ET': 'EEType'},
   135                                       {'X': 'Image', 'ETN': 'String', 'ET': 'CWEType'},
   136                                       {'X': 'Note', 'ETN': 'String', 'ET': 'EEType'},
   136                                       {'X': 'Note', 'ETN': 'String', 'ET': 'CWEType'},
   137                                       {'X': 'Personne', 'ETN': 'String', 'ET': 'EEType'},
   137                                       {'X': 'Personne', 'ETN': 'String', 'ET': 'CWEType'},
   138                                       {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'EEType'},
   138                                       {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'CWEType'},
   139                                       {'X': 'Societe', 'ETN': 'String', 'ET': 'EEType'},
   139                                       {'X': 'Societe', 'ETN': 'String', 'ET': 'CWEType'},
   140                                       {'X': 'State', 'ETN': 'String', 'ET': 'EEType'},
   140                                       {'X': 'State', 'ETN': 'String', 'ET': 'CWEType'},
   141                                       {'X': 'SubDivision', 'ETN': 'String', 'ET': 'EEType'},
   141                                       {'X': 'SubDivision', 'ETN': 'String', 'ET': 'CWEType'},
   142                                       {'X': 'Tag', 'ETN': 'String', 'ET': 'EEType'},
   142                                       {'X': 'Tag', 'ETN': 'String', 'ET': 'CWEType'},
   143                                       {'X': 'Transition', 'ETN': 'String', 'ET': 'EEType'},
   143                                       {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'},
   144                                       {'X': 'TrInfo', 'ETN': 'String', 'ET': 'EEType'}]))
   144                                       {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'}]))
   145         rql, solutions = partrqls[2]
   145         rql, solutions = partrqls[2]
   146         self.assertEquals(rql,
   146         self.assertEquals(rql,
   147                           'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), '
   147                           'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), '
   148                           'ET is EEType, X is Basket')
   148                           'ET is CWEType, X is Basket')
   149         self.assertEquals(solutions, [{'ET': 'EEType',
   149         self.assertEquals(solutions, [{'ET': 'CWEType',
   150                                        'X': 'Basket',
   150                                        'X': 'Basket',
   151                                        'ETN': 'String',
   151                                        'ETN': 'String',
   152                                        }])
   152                                        }])
   153 
   153 
   154     def test_preprocess_security_aggregat(self):
   154     def test_preprocess_security_aggregat(self):
   160         self.assertEquals(len(union.children[0].with_), 1)
   160         self.assertEquals(len(union.children[0].with_), 1)
   161         subq = union.children[0].with_[0].query
   161         subq = union.children[0].with_[0].query
   162         self.assertEquals(len(subq.children), 3)
   162         self.assertEquals(len(subq.children), 3)
   163         self.assertEquals([t.as_string() for t in union.children[0].selection],
   163         self.assertEquals([t.as_string() for t in union.children[0].selection],
   164                           ['MAX(X)'])
   164                           ['MAX(X)'])
   165         
   165 
   166     def test_preprocess_nonregr(self):
   166     def test_preprocess_nonregr(self):
   167         rqlst = self._prepare('Any S ORDERBY SI WHERE NOT S ecrit_par O, S para SI')
   167         rqlst = self._prepare('Any S ORDERBY SI WHERE NOT S ecrit_par O, S para SI')
   168         self.assertEquals(len(rqlst.solutions), 1)
   168         self.assertEquals(len(rqlst.solutions), 1)
   169     
   169 
   170     def test_build_description(self):
   170     def test_build_description(self):
   171         # should return an empty result set
   171         # should return an empty result set
   172         rset = self.execute('Any X WHERE X eid %(x)s', {'x': self.session.user.eid})
   172         rset = self.execute('Any X WHERE X eid %(x)s', {'x': self.session.user.eid})
   173         self.assertEquals(rset.description[0][0], 'EUser')
   173         self.assertEquals(rset.description[0][0], 'CWUser')
   174         rset = self.execute('Any 1')
   174         rset = self.execute('Any 1')
   175         self.assertEquals(rset.description[0][0], 'Int')
   175         self.assertEquals(rset.description[0][0], 'Int')
   176         rset = self.execute('Any TRUE')
   176         rset = self.execute('Any TRUE')
   177         self.assertEquals(rset.description[0][0], 'Boolean')
   177         self.assertEquals(rset.description[0][0], 'Boolean')
   178         rset = self.execute('Any "hop"')
   178         rset = self.execute('Any "hop"')
   187         self.assertEquals(rset.description[0][0], 'Int')
   187         self.assertEquals(rset.description[0][0], 'Int')
   188         rset = self.execute('Any %(x)s', {'x': True})
   188         rset = self.execute('Any %(x)s', {'x': True})
   189         self.assertEquals(rset.description[0][0], 'Boolean')
   189         self.assertEquals(rset.description[0][0], 'Boolean')
   190         rset = self.execute('Any %(x)s', {'x': 1.0})
   190         rset = self.execute('Any %(x)s', {'x': 1.0})
   191         self.assertEquals(rset.description[0][0], 'Float')
   191         self.assertEquals(rset.description[0][0], 'Float')
   192         rset = self.execute('Any %(x)s', {'x': now()})
   192         rset = self.execute('Any %(x)s', {'x': datetime.now()})
   193         self.assertEquals(rset.description[0][0], 'Datetime')
   193         self.assertEquals(rset.description[0][0], 'Datetime')
   194         rset = self.execute('Any %(x)s', {'x': 'str'})
   194         rset = self.execute('Any %(x)s', {'x': 'str'})
   195         self.assertEquals(rset.description[0][0], 'String')
   195         self.assertEquals(rset.description[0][0], 'String')
   196         rset = self.execute('Any %(x)s', {'x': u'str'})
   196         rset = self.execute('Any %(x)s', {'x': u'str'})
   197         self.assertEquals(rset.description[0][0], 'String')
   197         self.assertEquals(rset.description[0][0], 'String')
   200 class QuerierTC(BaseQuerierTC):
   200 class QuerierTC(BaseQuerierTC):
   201     repo = repo
   201     repo = repo
   202 
   202 
   203     def test_encoding_pb(self):
   203     def test_encoding_pb(self):
   204         self.assertRaises(RQLSyntaxError, self.execute,
   204         self.assertRaises(RQLSyntaxError, self.execute,
   205                           'Any X WHERE X is ERType, X name "öwned_by"')
   205                           'Any X WHERE X is CWRType, X name "öwned_by"')
   206 
   206 
   207     def test_unknown_eid(self):
   207     def test_unknown_eid(self):
   208         # should return an empty result set
   208         # should return an empty result set
   209         self.failIf(self.execute('Any X WHERE X eid 99999999'))
   209         self.failIf(self.execute('Any X WHERE X eid 99999999'))
   210         
   210 
   211     # selection queries tests #################################################
   211     # selection queries tests #################################################
   212     
   212 
   213     def test_select_1(self):
   213     def test_select_1(self):
   214         rset = self.execute('Any X ORDERBY X WHERE X is EGroup')
   214         rset = self.execute('Any X ORDERBY X WHERE X is CWGroup')
   215         result, descr = rset.rows, rset.description
   215         result, descr = rset.rows, rset.description
   216         self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,)])
   216         self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,)])
   217         self.assertEquals(descr, [('EGroup',), ('EGroup',), ('EGroup',), ('EGroup',)])
   217         self.assertEquals(descr, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)])
   218         
   218 
   219     def test_select_2(self):
   219     def test_select_2(self):
   220         rset = self.execute('Any X ORDERBY N WHERE X is EGroup, X name N')
   220         rset = self.execute('Any X ORDERBY N WHERE X is CWGroup, X name N')
   221         self.assertEquals(tuplify(rset.rows), [(3,), (1,), (4,), (2,)])
   221         self.assertEquals(tuplify(rset.rows), [(3,), (1,), (4,), (2,)])
   222         self.assertEquals(rset.description, [('EGroup',), ('EGroup',), ('EGroup',), ('EGroup',)])
   222         self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)])
   223         rset = self.execute('Any X ORDERBY N DESC WHERE X is EGroup, X name N')
   223         rset = self.execute('Any X ORDERBY N DESC WHERE X is CWGroup, X name N')
   224         self.assertEquals(tuplify(rset.rows), [(2,), (4,), (1,), (3,)])
   224         self.assertEquals(tuplify(rset.rows), [(2,), (4,), (1,), (3,)])
   225         
   225 
   226     def test_select_3(self):
   226     def test_select_3(self):
   227         rset = self.execute('Any N GROUPBY N WHERE X is EGroup, X name N')
   227         rset = self.execute('Any N GROUPBY N WHERE X is CWGroup, X name N')
   228         result, descr = rset.rows, rset.description
   228         result, descr = rset.rows, rset.description
   229         result.sort()
   229         result.sort()
   230         self.assertEquals(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)])
   230         self.assertEquals(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)])
   231         self.assertEquals(descr, [('String',), ('String',), ('String',), ('String',)])
   231         self.assertEquals(descr, [('String',), ('String',), ('String',), ('String',)])
   232         
   232 
   233     def test_select_is(self):
   233     def test_select_is(self):
   234         rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN')
   234         rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN')
   235         result, descr = rset.rows, rset.description
   235         result, descr = rset.rows, rset.description
   236         self.assertEquals(result[0][1], descr[0][0])
   236         self.assertEquals(result[0][1], descr[0][0])
   237         
   237 
   238     def test_select_is_aggr(self):
   238     def test_select_is_aggr(self):
   239         rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN')
   239         rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN')
   240         result, descr = rset.rows, rset.description
   240         result, descr = rset.rows, rset.description
   241         self.assertEquals(descr[0][0], 'String')
   241         self.assertEquals(descr[0][0], 'String')
   242         self.assertEquals(descr[0][1], 'Int')
   242         self.assertEquals(descr[0][1], 'Int')
   243         self.assertEquals(result[0][0], 'ENFRDef')
   243         self.assertEquals(result[0][0], 'CWRelation')
   244         
   244 
   245     def test_select_groupby_orderby(self):
   245     def test_select_groupby_orderby(self):
   246         rset = self.execute('Any N GROUPBY N ORDERBY N WHERE X is EGroup, X name N')
   246         rset = self.execute('Any N GROUPBY N ORDERBY N WHERE X is CWGroup, X name N')
   247         self.assertEquals(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)])
   247         self.assertEquals(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)])
   248         self.assertEquals(rset.description, [('String',), ('String',), ('String',), ('String',)])
   248         self.assertEquals(rset.description, [('String',), ('String',), ('String',), ('String',)])
   249         
   249 
   250     def test_select_complex_groupby(self):
   250     def test_select_complex_groupby(self):
   251         rset = self.execute('Any N GROUPBY N WHERE X name N')
   251         rset = self.execute('Any N GROUPBY N WHERE X name N')
   252         rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D')
   252         rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D')
   253         
   253 
   254     def test_select_inlined_groupby(self):
   254     def test_select_inlined_groupby(self):
   255         seid = self.execute('State X WHERE X name "deactivated"')[0][0]
   255         seid = self.execute('State X WHERE X name "deactivated"')[0][0]
   256         rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid)
   256         rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid)
   257         
   257 
   258     def test_select_complex_orderby(self):
   258     def test_select_complex_orderby(self):
   259         rset1 = self.execute('Any N ORDERBY N WHERE X name N')
   259         rset1 = self.execute('Any N ORDERBY N WHERE X name N')
   260         self.assertEquals(sorted(rset1.rows), rset1.rows)
   260         self.assertEquals(sorted(rset1.rows), rset1.rows)
   261         rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N')
   261         rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N')
   262         self.assertEquals(rset.rows[0][0], rset1.rows[1][0]) 
   262         self.assertEquals(rset.rows[0][0], rset1.rows[1][0])
   263         self.assertEquals(len(rset), 5)
   263         self.assertEquals(len(rset), 5)
   264         
   264 
   265     def test_select_5(self):
   265     def test_select_5(self):
   266         rset = self.execute('Any X, TMP ORDERBY TMP WHERE X name TMP, X is EGroup')
   266         rset = self.execute('Any X, TMP ORDERBY TMP WHERE X name TMP, X is CWGroup')
   267         self.assertEquals(tuplify(rset.rows), [(3, 'guests',), (1, 'managers',), (4, 'owners',), (2, 'users',)])
   267         self.assertEquals(tuplify(rset.rows), [(3, 'guests',), (1, 'managers',), (4, 'owners',), (2, 'users',)])
   268         self.assertEquals(rset.description, [('EGroup', 'String',), ('EGroup', 'String',), ('EGroup', 'String',), ('EGroup', 'String',)])
   268         self.assertEquals(rset.description, [('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',)])
   269         
   269 
   270     def test_select_6(self):
   270     def test_select_6(self):
   271         self.execute("INSERT Personne X: X nom 'bidule'")[0]
   271         self.execute("INSERT Personne X: X nom 'bidule'")[0]
   272         rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")')
   272         rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")')
   273         #self.assertEquals(rset.description, [('Personne',), ('Personne',)])
   273         #self.assertEquals(rset.description, [('Personne',), ('Personne',)])
   274         self.assert_(('Personne',) in rset.description)
   274         self.assert_(('Personne',) in rset.description)
   275         rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")')
   275         rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")')
   276         self.assert_(('Personne',) in rset.description)
   276         self.assert_(('Personne',) in rset.description)
   277         
   277 
   278     def test_select_not_attr(self):
   278     def test_select_not_attr(self):
   279         self.execute("INSERT Personne X: X nom 'bidule'")
   279         self.execute("INSERT Personne X: X nom 'bidule'")
   280         self.execute("INSERT Societe X: X nom 'chouette'")
   280         self.execute("INSERT Societe X: X nom 'chouette'")
   281         rset = self.execute('Personne X WHERE NOT X nom "bidule"')
   281         rset = self.execute('Personne X WHERE NOT X nom "bidule"')
   282         self.assertEquals(len(rset.rows), 0, rset.rows)
   282         self.assertEquals(len(rset.rows), 0, rset.rows)
   283         rset = self.execute('Personne X WHERE NOT X nom "bid"')
   283         rset = self.execute('Personne X WHERE NOT X nom "bid"')
   284         self.assertEquals(len(rset.rows), 1, rset.rows)
   284         self.assertEquals(len(rset.rows), 1, rset.rows)
   285         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   285         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   286         rset = self.execute('Personne X WHERE NOT X travaille S')
   286         rset = self.execute('Personne X WHERE NOT X travaille S')
   287         self.assertEquals(len(rset.rows), 0, rset.rows)
   287         self.assertEquals(len(rset.rows), 0, rset.rows)
   288         
   288 
   289     def test_select_is_in(self):
   289     def test_select_is_in(self):
   290         self.execute("INSERT Personne X: X nom 'bidule'")
   290         self.execute("INSERT Personne X: X nom 'bidule'")
   291         self.execute("INSERT Societe X: X nom 'chouette'")
   291         self.execute("INSERT Societe X: X nom 'chouette'")
   292         self.assertEquals(len(self.execute("Any X WHERE X is IN (Personne, Societe)")),
   292         self.assertEquals(len(self.execute("Any X WHERE X is IN (Personne, Societe)")),
   293                           2)
   293                           2)
   294         
   294 
   295     def test_select_not_rel(self):
   295     def test_select_not_rel(self):
   296         self.execute("INSERT Personne X: X nom 'bidule'")
   296         self.execute("INSERT Personne X: X nom 'bidule'")
   297         self.execute("INSERT Societe X: X nom 'chouette'")
   297         self.execute("INSERT Societe X: X nom 'chouette'")
   298         self.execute("INSERT Personne X: X nom 'autre'")
   298         self.execute("INSERT Personne X: X nom 'autre'")
   299         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   299         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   300         rset = self.execute('Personne X WHERE NOT X travaille S')
   300         rset = self.execute('Personne X WHERE NOT X travaille S')
   301         self.assertEquals(len(rset.rows), 1, rset.rows)
   301         self.assertEquals(len(rset.rows), 1, rset.rows)
   302         rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"')
   302         rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"')
   303         self.assertEquals(len(rset.rows), 1, rset.rows)
   303         self.assertEquals(len(rset.rows), 1, rset.rows)
   304         
   304 
   305     def test_select_nonregr_inlined(self):
   305     def test_select_nonregr_inlined(self):
   306         self.execute("INSERT Note X: X para 'bidule'")
   306         self.execute("INSERT Note X: X para 'bidule'")
   307         self.execute("INSERT Personne X: X nom 'chouette'")
   307         self.execute("INSERT Personne X: X nom 'chouette'")
   308         self.execute("INSERT Personne X: X nom 'autre'")
   308         self.execute("INSERT Personne X: X nom 'autre'")
   309         self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'")
   309         self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'")
   310         rset = self.execute('Any U,T ORDERBY T DESC WHERE U is EUser, '
   310         rset = self.execute('Any U,T ORDERBY T DESC WHERE U is CWUser, '
   311                             'N ecrit_par U, N type T')#, {'x': self.ueid})
   311                             'N ecrit_par U, N type T')#, {'x': self.ueid})
   312         self.assertEquals(len(rset.rows), 0)
   312         self.assertEquals(len(rset.rows), 0)
   313         
   313 
   314     def test_select_nonregr_edition_not(self):
   314     def test_select_nonregr_edition_not(self):
   315         groupeids = set((1, 2, 3))
   315         groupeids = set((1, 2, 3))
   316         groupreadperms = set(r[0] for r in self.execute('Any Y WHERE X name "EGroup", Y eid IN(1, 2, 3), X read_permission Y'))
   316         groupreadperms = set(r[0] for r in self.execute('Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), X read_permission Y'))
   317         rset = self.execute('DISTINCT Any Y WHERE X is EEType, X name "EGroup", Y eid IN(1, 2, 3), NOT X read_permission Y')
   317         rset = self.execute('DISTINCT Any Y WHERE X is CWEType, X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y')
   318         self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))
   318         self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))
   319         rset = self.execute('DISTINCT Any Y WHERE X name "EGroup", Y eid IN(1, 2, 3), NOT X read_permission Y')
   319         rset = self.execute('DISTINCT Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y')
   320         self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))
   320         self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))
   321                      
   321 
   322     def test_select_outer_join(self):
   322     def test_select_outer_join(self):
   323         peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   323         peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   324         peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0]
   324         peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0]
   325         seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0]
   325         seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0]
   326         seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0]
   326         seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0]
   329         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   329         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   330         rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?')
   330         rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?')
   331         self.assertEquals(rset.rows, [[peid1, seid1], [peid2, None]])
   331         self.assertEquals(rset.rows, [[peid1, seid1], [peid2, None]])
   332         rset = self.execute('Any S,X ORDERBY S WHERE X? travaille S')
   332         rset = self.execute('Any S,X ORDERBY S WHERE X? travaille S')
   333         self.assertEquals(rset.rows, [[seid1, peid1], [seid2, None]])
   333         self.assertEquals(rset.rows, [[seid1, peid1], [seid2, None]])
   334         
   334 
   335     def test_select_outer_join_optimized(self):
   335     def test_select_outer_join_optimized(self):
   336         peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   336         peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   337         rset = self.execute('Any X WHERE X eid %(x)s, P? connait X', {'x':peid1}, 'x')
   337         rset = self.execute('Any X WHERE X eid %(x)s, P? connait X', {'x':peid1}, 'x')
   338         self.assertEquals(rset.rows, [[peid1]])
   338         self.assertEquals(rset.rows, [[peid1]])
   339         rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', {'x':peid1}, 'x')
   339         rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', {'x':peid1}, 'x')
   340         self.assertEquals(rset.rows, [[peid1]])
   340         self.assertEquals(rset.rows, [[peid1]])
   341 
   341 
   342     def test_select_left_outer_join(self):
   342     def test_select_left_outer_join(self):
   343         ueid = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto', X in_group G "
   343         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G "
   344                             "WHERE G name 'users'")[0][0]
   344                             "WHERE G name 'users'")[0][0]
   345         self.commit()
   345         self.commit()
   346         try:
   346         try:
   347             rset = self.execute('Any FS,TS,C,D,U ORDERBY D DESC '
   347             rset = self.execute('Any FS,TS,C,D,U ORDERBY D DESC '
   348                                 'WHERE WF wf_info_for X,'
   348                                 'WHERE WF wf_info_for X,'
   357                                 'WF from_state FS?, WF to_state TS, WF comment C,'
   357                                 'WF from_state FS?, WF to_state TS, WF comment C,'
   358                                 'WF creation_date D, WF owned_by U, X eid %(x)s',
   358                                 'WF creation_date D, WF owned_by U, X eid %(x)s',
   359                                 {'x': ueid}, 'x')
   359                                 {'x': ueid}, 'x')
   360             self.assertEquals(len(rset), 2)
   360             self.assertEquals(len(rset), 2)
   361         finally:
   361         finally:
   362             self.execute('DELETE EUser X WHERE X eid %s' % ueid)
   362             self.execute('DELETE CWUser X WHERE X eid %s' % ueid)
   363             self.commit()
   363             self.commit()
   364 
   364 
   365     def test_select_ambigous_outer_join(self):
   365     def test_select_ambigous_outer_join(self):
   366         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
   366         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
   367         self.execute("INSERT Tag X: X name 'tagbis'")[0][0]
   367         self.execute("INSERT Tag X: X name 'tagbis'")[0][0]
   368         geid = self.execute("EGroup G WHERE G name 'users'")[0][0]
   368         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
   369         self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
   369         self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
   370                      {'g': geid, 't': teid}, 'g')
   370                      {'g': geid, 't': teid}, 'g')
   371         rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN")
   371         rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN")
   372         self.failUnless(['users', 'tag'] in rset.rows)
   372         self.failUnless(['users', 'tag'] in rset.rows)
   373         self.failUnless(['activated', None] in rset.rows)
   373         self.failUnless(['activated', None] in rset.rows)
   374         rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN")
   374         rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN")
   375         self.assertEquals(rset.rows, [[None, 'tagbis'], ['users', 'tag']])            
   375         self.assertEquals(rset.rows, [[None, 'tagbis'], ['users', 'tag']])
   376         
   376 
   377     def test_select_not_inline_rel(self):
   377     def test_select_not_inline_rel(self):
   378         self.execute("INSERT Personne X: X nom 'bidule'")
   378         self.execute("INSERT Personne X: X nom 'bidule'")
   379         self.execute("INSERT Note X: X type 'a'")
   379         self.execute("INSERT Note X: X type 'a'")
   380         self.execute("INSERT Note X: X type 'b'")
   380         self.execute("INSERT Note X: X type 'b'")
   381         self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'")
   381         self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'")
   382         rset = self.execute('Note X WHERE NOT X ecrit_par P')
   382         rset = self.execute('Note X WHERE NOT X ecrit_par P')
   383         self.assertEquals(len(rset.rows), 1, rset.rows)
   383         self.assertEquals(len(rset.rows), 1, rset.rows)
   384         
   384 
   385     def test_select_not_unlinked_multiple_solutions(self):
   385     def test_select_not_unlinked_multiple_solutions(self):
   386         self.execute("INSERT Personne X: X nom 'bidule'")
   386         self.execute("INSERT Personne X: X nom 'bidule'")
   387         self.execute("INSERT Note X: X type 'a'")
   387         self.execute("INSERT Note X: X type 'a'")
   388         self.execute("INSERT Note X: X type 'b'")
   388         self.execute("INSERT Note X: X type 'b'")
   389         self.execute("SET Y evaluee X WHERE X type 'a', Y nom 'bidule'")
   389         self.execute("SET Y evaluee X WHERE X type 'a', Y nom 'bidule'")
   393     def test_select_aggregat_count(self):
   393     def test_select_aggregat_count(self):
   394         rset = self.execute('Any COUNT(X)')
   394         rset = self.execute('Any COUNT(X)')
   395         self.assertEquals(len(rset.rows), 1)
   395         self.assertEquals(len(rset.rows), 1)
   396         self.assertEquals(len(rset.rows[0]), 1)
   396         self.assertEquals(len(rset.rows[0]), 1)
   397         self.assertEquals(rset.description, [('Int',)])
   397         self.assertEquals(rset.description, [('Int',)])
   398         
   398 
   399     def test_select_aggregat_sum(self):
   399     def test_select_aggregat_sum(self):
   400         rset = self.execute('Any SUM(O) WHERE X ordernum O')
   400         rset = self.execute('Any SUM(O) WHERE X ordernum O')
   401         self.assertEquals(len(rset.rows), 1)
   401         self.assertEquals(len(rset.rows), 1)
   402         self.assertEquals(len(rset.rows[0]), 1)
   402         self.assertEquals(len(rset.rows[0]), 1)
   403         self.assertEquals(rset.description, [('Int',)])
   403         self.assertEquals(rset.description, [('Int',)])
   404         
   404 
   405     def test_select_aggregat_min(self):
   405     def test_select_aggregat_min(self):
   406         rset = self.execute('Any MIN(X) WHERE X is Personne')
   406         rset = self.execute('Any MIN(X) WHERE X is Personne')
   407         self.assertEquals(len(rset.rows), 1)
   407         self.assertEquals(len(rset.rows), 1)
   408         self.assertEquals(len(rset.rows[0]), 1)
   408         self.assertEquals(len(rset.rows[0]), 1)
   409         self.assertEquals(rset.description, [('Personne',)])
   409         self.assertEquals(rset.description, [('Personne',)])
   410         rset = self.execute('Any MIN(O) WHERE X ordernum O')
   410         rset = self.execute('Any MIN(O) WHERE X ordernum O')
   411         self.assertEquals(len(rset.rows), 1)
   411         self.assertEquals(len(rset.rows), 1)
   412         self.assertEquals(len(rset.rows[0]), 1)
   412         self.assertEquals(len(rset.rows[0]), 1)
   413         self.assertEquals(rset.description, [('Int',)])
   413         self.assertEquals(rset.description, [('Int',)])
   414         
   414 
   415     def test_select_aggregat_max(self):
   415     def test_select_aggregat_max(self):
   416         rset = self.execute('Any MAX(X) WHERE X is Personne')
   416         rset = self.execute('Any MAX(X) WHERE X is Personne')
   417         self.assertEquals(len(rset.rows), 1)
   417         self.assertEquals(len(rset.rows), 1)
   418         self.assertEquals(len(rset.rows[0]), 1)
   418         self.assertEquals(len(rset.rows[0]), 1)
   419         self.assertEquals(rset.description, [('Personne',)])
   419         self.assertEquals(rset.description, [('Personne',)])
   421         self.assertEquals(len(rset.rows), 1)
   421         self.assertEquals(len(rset.rows), 1)
   422         self.assertEquals(len(rset.rows[0]), 1)
   422         self.assertEquals(len(rset.rows[0]), 1)
   423         self.assertEquals(rset.description, [('Int',)])
   423         self.assertEquals(rset.description, [('Int',)])
   424 
   424 
   425     def test_select_custom_aggregat_concat_string(self):
   425     def test_select_custom_aggregat_concat_string(self):
   426         rset = self.execute('Any CONCAT_STRINGS(N) WHERE X is EGroup, X name N')
   426         rset = self.execute('Any CONCAT_STRINGS(N) WHERE X is CWGroup, X name N')
   427         self.failUnless(rset)
   427         self.failUnless(rset)
   428         self.failUnlessEqual(sorted(rset[0][0].split(', ')), ['guests', 'managers',
   428         self.failUnlessEqual(sorted(rset[0][0].split(', ')), ['guests', 'managers',
   429                                                              'owners', 'users'])
   429                                                              'owners', 'users'])
   430 
   430 
   431     def test_select_custom_regproc_limit_size(self):
   431     def test_select_custom_regproc_limit_size(self):
   432         rset = self.execute('Any TEXT_LIMIT_SIZE(N, 3) WHERE X is EGroup, X name N, X name "managers"')
   432         rset = self.execute('Any TEXT_LIMIT_SIZE(N, 3) WHERE X is CWGroup, X name N, X name "managers"')
   433         self.failUnless(rset)
   433         self.failUnless(rset)
   434         self.failUnlessEqual(rset[0][0], 'man...')
   434         self.failUnlessEqual(rset[0][0], 'man...')
   435         self.execute("INSERT Basket X: X name 'bidule', X description '<b>hop hop</b>', X description_format 'text/html'")
   435         self.execute("INSERT Basket X: X name 'bidule', X description '<b>hop hop</b>', X description_format 'text/html'")
   436         rset = self.execute('Any LIMIT_SIZE(D, DF, 3) WHERE X is Basket, X description D, X description_format DF')
   436         rset = self.execute('Any LIMIT_SIZE(D, DF, 3) WHERE X is Basket, X description D, X description_format DF')
   437         self.failUnless(rset)
   437         self.failUnless(rset)
   438         self.failUnlessEqual(rset[0][0], 'hop...')
   438         self.failUnlessEqual(rset[0][0], 'hop...')
   439 
   439 
   440     def test_select_regproc_orderby(self):
   440     def test_select_regproc_orderby(self):
   441         rset = self.execute('DISTINCT Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is EGroup, X name N, X name "managers"')
   441         rset = self.execute('DISTINCT Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, X name "managers"')
   442         self.failUnlessEqual(len(rset), 1)
   442         self.failUnlessEqual(len(rset), 1)
   443         self.failUnlessEqual(rset[0][1], 'managers')
   443         self.failUnlessEqual(rset[0][1], 'managers')
   444         rset = self.execute('Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is EGroup, X name N, NOT U in_group X, U login "admin"')
   444         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"')
   445         self.failUnlessEqual(len(rset), 3)
   445         self.failUnlessEqual(len(rset), 3)
   446         self.failUnlessEqual(rset[0][1], 'owners')
   446         self.failUnlessEqual(rset[0][1], 'owners')
   447         
   447 
   448     def test_select_aggregat_sort(self):
   448     def test_select_aggregat_sort(self):
   449         rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G')
   449         rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G')
   450         self.assertEquals(len(rset.rows), 2)
   450         self.assertEquals(len(rset.rows), 2)
   451         self.assertEquals(len(rset.rows[0]), 2)
   451         self.assertEquals(len(rset.rows[0]), 2)
   452         self.assertEquals(rset.description[0], ('EGroup', 'Int',))
   452         self.assertEquals(rset.description[0], ('CWGroup', 'Int',))
   453 
   453 
   454     def test_select_aggregat_having(self):
   454     def test_select_aggregat_having(self):
   455         rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N '
   455         rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N '
   456                             'WHERE RT name N, RDEF relation_type RT '
   456                             'WHERE RT name N, RDEF relation_type RT '
   457                             'HAVING COUNT(RDEF) > 10')
   457                             'HAVING COUNT(RDEF) > 10')
   473         self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed')
   473         self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed')
   474         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')
   475         result = rset.rows
   475         result = rset.rows
   476         result.sort()
   476         result.sort()
   477         self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,), (5,)])
   477         self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,), (5,)])
   478         
   478 
   479     def test_select_upper(self):
   479     def test_select_upper(self):
   480         rset = self.execute('Any X, UPPER(L) ORDERBY L WHERE X is EUser, X login L')
   480         rset = self.execute('Any X, UPPER(L) ORDERBY L WHERE X is CWUser, X login L')
   481         self.assertEquals(len(rset.rows), 2)
   481         self.assertEquals(len(rset.rows), 2)
   482         self.assertEquals(rset.rows[0][1], 'ADMIN')
   482         self.assertEquals(rset.rows[0][1], 'ADMIN')
   483         self.assertEquals(rset.description[0], ('EUser', 'String',))
   483         self.assertEquals(rset.description[0], ('CWUser', 'String',))
   484         self.assertEquals(rset.rows[1][1], 'ANON')
   484         self.assertEquals(rset.rows[1][1], 'ANON')
   485         self.assertEquals(rset.description[1], ('EUser', 'String',))
   485         self.assertEquals(rset.description[1], ('CWUser', 'String',))
   486         eid = rset.rows[0][0]
   486         eid = rset.rows[0][0]
   487         rset = self.execute('Any UPPER(L) WHERE X eid %s, X login L'%eid)
   487         rset = self.execute('Any UPPER(L) WHERE X eid %s, X login L'%eid)
   488         self.assertEquals(rset.rows[0][0], 'ADMIN')
   488         self.assertEquals(rset.rows[0][0], 'ADMIN')
   489         self.assertEquals(rset.description, [('String',)])
   489         self.assertEquals(rset.description, [('String',)])
   490 
   490 
   492 ##         ueid = self.session.user.eid
   492 ##         ueid = self.session.user.eid
   493 ##         rset = self.execute('Any L WHERE %s login L'%ueid)
   493 ##         rset = self.execute('Any L WHERE %s login L'%ueid)
   494 ##         self.assertEquals(rset.rows[0][0], 'admin')
   494 ##         self.assertEquals(rset.rows[0][0], 'admin')
   495 ##         rset = self.execute('Any L WHERE %(x)s login L', {'x':ueid})
   495 ##         rset = self.execute('Any L WHERE %(x)s login L', {'x':ueid})
   496 ##         self.assertEquals(rset.rows[0][0], 'admin')
   496 ##         self.assertEquals(rset.rows[0][0], 'admin')
   497         
   497 
   498     def test_select_searchable_text_1(self):
   498     def test_select_searchable_text_1(self):
   499         rset = self.execute(u"INSERT Personne X: X nom 'bidüle'")
   499         rset = self.execute(u"INSERT Personne X: X nom 'bidüle'")
   500         rset = self.execute(u"INSERT Societe X: X nom 'bidüle'")
   500         rset = self.execute(u"INSERT Societe X: X nom 'bidüle'")
   501         rset = self.execute("INSERT Societe X: X nom 'chouette'")
   501         rset = self.execute("INSERT Societe X: X nom 'chouette'")
   502         self.commit()
   502         self.commit()
   507         biduleeids = [r[0] for r in rset.rows]
   507         biduleeids = [r[0] for r in rset.rows]
   508         rset = self.execute(u'Any N where NOT N has_text "bidüle"')
   508         rset = self.execute(u'Any N where NOT N has_text "bidüle"')
   509         self.failIf([r[0] for r in rset.rows if r[0] in biduleeids])
   509         self.failIf([r[0] for r in rset.rows if r[0] in biduleeids])
   510         # duh?
   510         # duh?
   511         rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'})
   511         rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'})
   512         
   512 
   513     def test_select_searchable_text_2(self):
   513     def test_select_searchable_text_2(self):
   514         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   514         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   515         rset = self.execute("INSERT Personne X: X nom 'chouette'")
   515         rset = self.execute("INSERT Personne X: X nom 'chouette'")
   516         rset = self.execute("INSERT Societe X: X nom 'bidule'")
   516         rset = self.execute("INSERT Societe X: X nom 'bidule'")
   517         self.commit()
   517         self.commit()
   518         rset = self.execute('Personne N where N has_text "bidule"')
   518         rset = self.execute('Personne N where N has_text "bidule"')
   519         self.assertEquals(len(rset.rows), 1, rset.rows)
   519         self.assertEquals(len(rset.rows), 1, rset.rows)
   520         
   520 
   521     def test_select_searchable_text_3(self):
   521     def test_select_searchable_text_3(self):
   522         rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'M'")
   522         rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'M'")
   523         rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'F'")
   523         rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'F'")
   524         rset = self.execute("INSERT Societe X: X nom 'bidule'")
   524         rset = self.execute("INSERT Societe X: X nom 'bidule'")
   525         self.commit()
   525         self.commit()
   526         rset = self.execute('Any X where X has_text "bidule" and X sexe "M"')
   526         rset = self.execute('Any X where X has_text "bidule" and X sexe "M"')
   527         self.assertEquals(len(rset.rows), 1, rset.rows)
   527         self.assertEquals(len(rset.rows), 1, rset.rows)
   528         
   528 
   529     def test_select_multiple_searchable_text(self):
   529     def test_select_multiple_searchable_text(self):
   530         self.execute(u"INSERT Personne X: X nom 'bidüle'")
   530         self.execute(u"INSERT Personne X: X nom 'bidüle'")
   531         self.execute("INSERT Societe X: X nom 'chouette', S travaille X")
   531         self.execute("INSERT Societe X: X nom 'chouette', S travaille X")
   532         self.execute(u"INSERT Personne X: X nom 'bidüle'")
   532         self.execute(u"INSERT Personne X: X nom 'bidüle'")
   533         self.commit()
   533         self.commit()
   534         rset = self.execute('Personne X WHERE X has_text %(text)s, X travaille S, S has_text %(text2)s',
   534         rset = self.execute('Personne X WHERE X has_text %(text)s, X travaille S, S has_text %(text2)s',
   535                             {'text': u'bidüle',
   535                             {'text': u'bidüle',
   536                              'text2': u'chouette',}
   536                              'text2': u'chouette',}
   537                             )
   537                             )
   538         self.assertEquals(len(rset.rows), 1, rset.rows)
   538         self.assertEquals(len(rset.rows), 1, rset.rows)
   539         
   539 
   540     def test_select_no_descr(self):
   540     def test_select_no_descr(self):
   541         rset = self.execute('Any X WHERE X is EGroup', build_descr=0)
   541         rset = self.execute('Any X WHERE X is CWGroup', build_descr=0)
   542         rset.rows.sort()
   542         rset.rows.sort()
   543         self.assertEquals(tuplify(rset.rows), [(1,), (2,), (3,), (4,)])
   543         self.assertEquals(tuplify(rset.rows), [(1,), (2,), (3,), (4,)])
   544         self.assertEquals(rset.description, ())
   544         self.assertEquals(rset.description, ())
   545 
   545 
   546     def test_select_limit_offset(self):
   546     def test_select_limit_offset(self):
   547         rset = self.execute('EGroup X ORDERBY N LIMIT 2 WHERE X name N')
   547         rset = self.execute('CWGroup X ORDERBY N LIMIT 2 WHERE X name N')
   548         self.assertEquals(tuplify(rset.rows), [(3,), (1,)])
   548         self.assertEquals(tuplify(rset.rows), [(3,), (1,)])
   549         self.assertEquals(rset.description, [('EGroup',), ('EGroup',)])
   549         self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',)])
   550         rset = self.execute('EGroup X ORDERBY N LIMIT 2 OFFSET 2 WHERE X name N')
   550         rset = self.execute('CWGroup X ORDERBY N LIMIT 2 OFFSET 2 WHERE X name N')
   551         self.assertEquals(tuplify(rset.rows), [(4,), (2,)])
   551         self.assertEquals(tuplify(rset.rows), [(4,), (2,)])
   552         
   552 
   553     def test_select_symetric(self):
   553     def test_select_symetric(self):
   554         self.execute("INSERT Personne X: X nom 'machin'")
   554         self.execute("INSERT Personne X: X nom 'machin'")
   555         self.execute("INSERT Personne X: X nom 'bidule'")
   555         self.execute("INSERT Personne X: X nom 'bidule'")
   556         self.execute("INSERT Personne X: X nom 'chouette'")
   556         self.execute("INSERT Personne X: X nom 'chouette'")
   557         self.execute("INSERT Personne X: X nom 'trucmuche'")
   557         self.execute("INSERT Personne X: X nom 'trucmuche'")
   567         self.assertEquals(len(rset.rows), 1, rset.rows)
   567         self.assertEquals(len(rset.rows), 1, rset.rows)
   568         rset = self.execute('Any P where P connait P2, P2 nom "chouette"')
   568         rset = self.execute('Any P where P connait P2, P2 nom "chouette"')
   569         self.assertEquals(len(rset.rows), 2, rset.rows)
   569         self.assertEquals(len(rset.rows), 2, rset.rows)
   570         rset = self.execute('Any P where P2 connait P, P2 nom "chouette"')
   570         rset = self.execute('Any P where P2 connait P, P2 nom "chouette"')
   571         self.assertEquals(len(rset.rows), 2, rset.rows)
   571         self.assertEquals(len(rset.rows), 2, rset.rows)
   572         
   572 
   573     def test_select_inline(self):
   573     def test_select_inline(self):
   574         self.execute("INSERT Personne X: X nom 'bidule'")
   574         self.execute("INSERT Personne X: X nom 'bidule'")
   575         self.execute("INSERT Note X: X type 'a'")
   575         self.execute("INSERT Note X: X type 'a'")
   576         self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'")
   576         self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'")
   577         rset = self.execute('Any N where N ecrit_par X, X nom "bidule"')
   577         rset = self.execute('Any N where N ecrit_par X, X nom "bidule"')
   578         self.assertEquals(len(rset.rows), 1, rset.rows)
   578         self.assertEquals(len(rset.rows), 1, rset.rows)
   579         
   579 
   580     def test_select_creation_date(self):
   580     def test_select_creation_date(self):
   581         self.execute("INSERT Personne X: X nom 'bidule'")
   581         self.execute("INSERT Personne X: X nom 'bidule'")
   582         rset = self.execute('Any D WHERE X nom "bidule", X creation_date D')
   582         rset = self.execute('Any D WHERE X nom "bidule", X creation_date D')
   583         self.assertEqual(len(rset.rows), 1)
   583         self.assertEqual(len(rset.rows), 1)
   584 
   584 
   591         rset = self.execute('DISTINCT Any P WHERE P travaille S1 OR P travaille S2, S1 nom "logilab", S2 nom "caesium"')
   591         rset = self.execute('DISTINCT Any P WHERE P travaille S1 OR P travaille S2, S1 nom "logilab", S2 nom "caesium"')
   592         self.assertEqual(len(rset.rows), 1)
   592         self.assertEqual(len(rset.rows), 1)
   593         self.execute("SET P travaille S WHERE P nom 'chouette', S nom 'caesium'")
   593         self.execute("SET P travaille S WHERE P nom 'chouette', S nom 'caesium'")
   594         rset = self.execute('DISTINCT Any P WHERE P travaille S1 OR P travaille S2, S1 nom "logilab", S2 nom "caesium"')
   594         rset = self.execute('DISTINCT Any P WHERE P travaille S1 OR P travaille S2, S1 nom "logilab", S2 nom "caesium"')
   595         self.assertEqual(len(rset.rows), 2)
   595         self.assertEqual(len(rset.rows), 2)
   596         
   596 
   597     def test_select_or_sym_relation(self):
   597     def test_select_or_sym_relation(self):
   598         self.execute("INSERT Personne X: X nom 'bidule'")
   598         self.execute("INSERT Personne X: X nom 'bidule'")
   599         self.execute("INSERT Personne X: X nom 'chouette'")
   599         self.execute("INSERT Personne X: X nom 'chouette'")
   600         self.execute("INSERT Personne X: X nom 'truc'")
   600         self.execute("INSERT Personne X: X nom 'truc'")
   601         self.execute("SET P connait S WHERE P nom 'bidule', S nom 'chouette'")
   601         self.execute("SET P connait S WHERE P nom 'bidule', S nom 'chouette'")
   606         self.execute("SET P connait S WHERE P nom 'chouette', S nom 'truc'")
   606         self.execute("SET P connait S WHERE P nom 'chouette', S nom 'truc'")
   607         rset = self.execute('DISTINCT Any P WHERE S connait P, S nom "chouette"')
   607         rset = self.execute('DISTINCT Any P WHERE S connait P, S nom "chouette"')
   608         self.assertEqual(len(rset.rows), 2, rset.rows)
   608         self.assertEqual(len(rset.rows), 2, rset.rows)
   609         rset = self.execute('DISTINCT Any P WHERE P connait S OR S connait P, S nom "chouette"')
   609         rset = self.execute('DISTINCT Any P WHERE P connait S OR S connait P, S nom "chouette"')
   610         self.assertEqual(len(rset.rows), 2, rset.rows)
   610         self.assertEqual(len(rset.rows), 2, rset.rows)
   611             
   611 
   612     def test_select_follow_relation(self):
   612     def test_select_follow_relation(self):
   613         self.execute("INSERT Affaire X: X sujet 'cool'")
   613         self.execute("INSERT Affaire X: X sujet 'cool'")
   614         self.execute("INSERT Societe X: X nom 'chouette'")
   614         self.execute("INSERT Societe X: X nom 'chouette'")
   615         self.execute("SET A concerne S WHERE A is Affaire, S is Societe")
   615         self.execute("SET A concerne S WHERE A is Affaire, S is Societe")
   616         self.execute("INSERT Note X: X para 'truc'")
   616         self.execute("INSERT Note X: X para 'truc'")
   633         self.execute("INSERT Affaire X: X sujet 'normal'")
   633         self.execute("INSERT Affaire X: X sujet 'normal'")
   634         self.execute("INSERT Affaire X: X sujet 'zou'")
   634         self.execute("INSERT Affaire X: X sujet 'zou'")
   635         self.execute("INSERT Affaire X: X sujet 'abcd'")
   635         self.execute("INSERT Affaire X: X sujet 'abcd'")
   636         rset = self.execute('DISTINCT Any S ORDERBY S WHERE A is Affaire, A sujet S')
   636         rset = self.execute('DISTINCT Any S ORDERBY S WHERE A is Affaire, A sujet S')
   637         self.assertEqual(rset.rows, [['abcd'], ['important'], ['minor'], ['normal'], ['zou']])
   637         self.assertEqual(rset.rows, [['abcd'], ['important'], ['minor'], ['normal'], ['zou']])
   638         
   638 
   639     def test_select_ordered_distinct_3(self):
   639     def test_select_ordered_distinct_3(self):
   640         rset = self.execute('DISTINCT Any N ORDERBY GROUP_SORT_VALUE(N) WHERE X is EGroup, X name N')
   640         rset = self.execute('DISTINCT Any N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N')
   641         self.assertEqual(rset.rows, [['owners'], ['guests'], ['users'], ['managers']])
   641         self.assertEqual(rset.rows, [['owners'], ['guests'], ['users'], ['managers']])
   642 
   642 
   643     def test_select_or_value(self):
   643     def test_select_or_value(self):
   644         rset = self.execute('Any U WHERE U in_group G, G name "owners" OR G name "users"')
   644         rset = self.execute('Any U WHERE U in_group G, G name "owners" OR G name "users"')
   645         self.assertEqual(len(rset.rows), 0)
   645         self.assertEqual(len(rset.rows), 0)
   648 
   648 
   649     def test_select_explicit_eid(self):
   649     def test_select_explicit_eid(self):
   650         rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid})
   650         rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid})
   651         self.failUnless(rset)
   651         self.failUnless(rset)
   652         self.assertEquals(rset.description[0][1], 'Int')
   652         self.assertEquals(rset.description[0][1], 'Int')
   653         
   653 
   654 #     def test_select_rewritten_optional(self):
   654 #     def test_select_rewritten_optional(self):
   655 #         eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
   655 #         eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
   656 #         rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)',
   656 #         rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)',
   657 #                             {'x': eid}, 'x')
   657 #                             {'x': eid}, 'x')
   658 #         self.assertEquals(rset.rows, [[eid]])
   658 #         self.assertEquals(rset.rows, [[eid]])
   659         
   659 
   660     def test_today_bug(self):
   660     def test_today_bug(self):
   661         self.execute("INSERT Tag X: X name 'bidule', X creation_date TODAY")
   661         self.execute("INSERT Tag X: X name 'bidule', X creation_date NOW")
   662         self.execute("INSERT Tag Y: Y name 'toto'")
   662         self.execute("INSERT Tag Y: Y name 'toto'")
   663         rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D")
   663         rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D")
   664         self.assert_(isinstance(rset.rows[0][0], DateTimeType), rset.rows)
   664         self.assert_(isinstance(rset.rows[0][0], datetime), rset.rows)
   665         rset = self.execute('Tag X WHERE X creation_date TODAY')
   665         rset = self.execute('Tag X WHERE X creation_date TODAY')
   666         self.assertEqual(len(rset.rows), 2)
   666         self.assertEqual(len(rset.rows), 2)
   667         rset = self.execute('Any MAX(D) WHERE X is Tag, X creation_date D')
   667         rset = self.execute('Any MAX(D) WHERE X is Tag, X creation_date D')
   668         self.failUnless(isinstance(rset[0][0], DateTimeType), type(rset[0][0]))
   668         self.failUnless(isinstance(rset[0][0], datetime), type(rset[0][0]))
   669 
   669 
   670     def test_today(self):
   670     def test_today(self):
   671         self.execute("INSERT Tag X: X name 'bidule', X creation_date TODAY")
   671         self.execute("INSERT Tag X: X name 'bidule', X creation_date TODAY")
   672         self.execute("INSERT Tag Y: Y name 'toto'")
   672         self.execute("INSERT Tag Y: Y name 'toto'")
   673         rset = self.execute('Tag X WHERE X creation_date TODAY')
   673         rset = self.execute('Tag X WHERE X creation_date TODAY')
   674         self.assertEqual(len(rset.rows), 2)
   674         self.assertEqual(len(rset.rows), 2)
   675 
   675 
   676     def test_select_boolean(self):
   676     def test_select_boolean(self):
   677         rset = self.execute('Any N WHERE X is EEType, X name N, X final %(val)s',
   677         rset = self.execute('Any N WHERE X is CWEType, X name N, X final %(val)s',
   678                             {'val': True})
   678                             {'val': True})
   679         self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes',
   679         self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes',
   680                                                             'Date', 'Datetime',
   680                                                             'Date', 'Datetime',
   681                                                             'Decimal', 'Float',
   681                                                             'Decimal', 'Float',
   682                                                             'Int', 'Interval',
   682                                                             'Int', 'Interval',
   683                                                             'Password', 'String',
   683                                                             'Password', 'String',
   684                                                             'Time'])
   684                                                             'Time'])
   685         rset = self.execute('Any N WHERE X is EEType, X name N, X final TRUE')
   685         rset = self.execute('Any N WHERE X is CWEType, X name N, X final TRUE')
   686         self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes',
   686         self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes',
   687                                                             'Date', 'Datetime',
   687                                                             'Date', 'Datetime',
   688                                                             'Decimal', 'Float',
   688                                                             'Decimal', 'Float',
   689                                                             'Int', 'Interval',
   689                                                             'Int', 'Interval',
   690                                                             'Password', 'String',
   690                                                             'Password', 'String',
   691                                                             'Time'])
   691                                                             'Time'])
   692         
   692 
   693     def test_select_constant(self):
   693     def test_select_constant(self):
   694         rset = self.execute('Any X, "toto" ORDERBY X WHERE X is EGroup')
   694         rset = self.execute('Any X, "toto" ORDERBY X WHERE X is CWGroup')
   695         self.assertEquals(rset.rows,
   695         self.assertEquals(rset.rows,
   696                           map(list, zip((1,2,3,4), ('toto','toto','toto','toto',))))
   696                           map(list, zip((1,2,3,4), ('toto','toto','toto','toto',))))
   697         self.assertIsInstance(rset[0][1], unicode)
   697         self.assertIsInstance(rset[0][1], unicode)
   698         self.assertEquals(rset.description,
   698         self.assertEquals(rset.description,
   699                           zip(('EGroup', 'EGroup', 'EGroup', 'EGroup'),
   699                           zip(('CWGroup', 'CWGroup', 'CWGroup', 'CWGroup'),
   700                               ('String', 'String', 'String', 'String',)))
   700                               ('String', 'String', 'String', 'String',)))
   701         rset = self.execute('Any X, %(value)s ORDERBY X WHERE X is EGroup', {'value': 'toto'})
   701         rset = self.execute('Any X, %(value)s ORDERBY X WHERE X is CWGroup', {'value': 'toto'})
   702         self.assertEquals(rset.rows,
   702         self.assertEquals(rset.rows,
   703                           map(list, zip((1,2,3,4), ('toto','toto','toto','toto',))))
   703                           map(list, zip((1,2,3,4), ('toto','toto','toto','toto',))))
   704         self.assertIsInstance(rset[0][1], unicode)
   704         self.assertIsInstance(rset[0][1], unicode)
   705         self.assertEquals(rset.description,
   705         self.assertEquals(rset.description,
   706                           zip(('EGroup', 'EGroup', 'EGroup', 'EGroup'),
   706                           zip(('CWGroup', 'CWGroup', 'CWGroup', 'CWGroup'),
   707                               ('String', 'String', 'String', 'String',)))
   707                               ('String', 'String', 'String', 'String',)))
   708         rset = self.execute('Any X,GN WHERE X is EUser, G is EGroup, X login "syt", X in_group G, G name GN')
   708         rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN')
   709 
   709 
   710     def test_select_union(self):
   710     def test_select_union(self):
   711         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   711         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   712                             '((Any X,N WHERE X name N, X transition_of E, E name %(name)s)'
   712                             '((Any X,N WHERE X name N, X transition_of E, E name %(name)s)'
   713                             ' UNION '
   713                             ' UNION '
   714                             '(Any X,N WHERE X name N, X state_of E, E name %(name)s))',
   714                             '(Any X,N WHERE X name N, X state_of E, E name %(name)s))',
   715                             {'name': 'EUser'})
   715                             {'name': 'CWUser'})
   716         self.assertEquals([x[1] for x in rset.rows],
   716         self.assertEquals([x[1] for x in rset.rows],
   717                           ['activate', 'activated', 'deactivate', 'deactivated'])
   717                           ['activate', 'activated', 'deactivate', 'deactivated'])
   718         self.assertEquals(rset.description,
   718         self.assertEquals(rset.description,
   719                           [('Transition', 'String'), ('State', 'String'),
   719                           [('Transition', 'String'), ('State', 'String'),
   720                            ('Transition', 'String'), ('State', 'String')])
   720                            ('Transition', 'String'), ('State', 'String')])
   721         
   721 
   722     def test_select_union_aggregat(self):
   722     def test_select_union_aggregat(self):
   723         # meaningless, the goal in to have group by done on different attribute
   723         # meaningless, the goal in to have group by done on different attribute
   724         # for each sub-query
   724         # for each sub-query
   725         self.execute('(Any N,COUNT(X) GROUPBY N WHERE X name N, X is State)'
   725         self.execute('(Any N,COUNT(X) GROUPBY N WHERE X name N, X is State)'
   726                      ' UNION '
   726                      ' UNION '
   727                      '(Any N,COUNT(X) GROUPBY N ORDERBY 2 WHERE X login N)')
   727                      '(Any N,COUNT(X) GROUPBY N ORDERBY 2 WHERE X login N)')
   728         
   728 
   729     def test_select_union_aggregat_independant_group(self):
   729     def test_select_union_aggregat_independant_group(self):
   730         self.execute('INSERT State X: X name "hop"')
   730         self.execute('INSERT State X: X name "hop"')
   731         self.execute('INSERT State X: X name "hop"')
   731         self.execute('INSERT State X: X name "hop"')
   732         self.execute('INSERT Transition X: X name "hop"')
   732         self.execute('INSERT Transition X: X name "hop"')
   733         self.execute('INSERT Transition X: X name "hop"')
   733         self.execute('INSERT Transition X: X name "hop"')
   734         rset = self.execute('Any N,NX ORDERBY 2 WITH N,NX BEING '
   734         rset = self.execute('Any N,NX ORDERBY 2 WITH N,NX BEING '
   735                             '((Any N,COUNT(X) GROUPBY N WHERE X name N, X is State HAVING COUNT(X)>1)'
   735                             '((Any N,COUNT(X) GROUPBY N WHERE X name N, X is State HAVING COUNT(X)>1)'
   736                             ' UNION '
   736                             ' UNION '
   737                             '(Any N,COUNT(X) GROUPBY N WHERE X name N, X is Transition HAVING COUNT(X)>1))')
   737                             '(Any N,COUNT(X) GROUPBY N WHERE X name N, X is Transition HAVING COUNT(X)>1))')
   738         self.assertEquals(rset.rows, [[u'hop', 2], [u'hop', 2]])
   738         self.assertEquals(rset.rows, [[u'hop', 2], [u'hop', 2]])
   739         
   739 
   740     def test_select_union_selection_with_diff_variables(self):
   740     def test_select_union_selection_with_diff_variables(self):
   741         rset = self.execute('(Any N WHERE X name N, X is State)'
   741         rset = self.execute('(Any N WHERE X name N, X is State)'
   742                             ' UNION '
   742                             ' UNION '
   743                             '(Any NN WHERE XX name NN, XX is Transition)')
   743                             '(Any NN WHERE XX name NN, XX is Transition)')
   744         self.assertEquals(sorted(r[0] for r in rset.rows),
   744         self.assertEquals(sorted(r[0] for r in rset.rows),
   745                           ['abort', 'activate', 'activated', 'ben non',
   745                           ['abort', 'activate', 'activated', 'ben non',
   746                            'deactivate', 'deactivated', 'done', 'en cours',
   746                            'deactivate', 'deactivated', 'done', 'en cours',
   747                            'end', 'finie', 'markasdone', 'pitetre', 'redoit',
   747                            'end', 'finie', 'markasdone', 'pitetre', 'redoit',
   748                            'start', 'todo'])
   748                            'start', 'todo'])
   749         
   749 
   750     def test_exists(self):
   750     def test_exists(self):
   751         geid = self.execute("INSERT EGroup X: X name 'lulufanclub'")[0][0]
   751         geid = self.execute("INSERT CWGroup X: X name 'lulufanclub'")[0][0]
   752         self.execute("SET U in_group G WHERE G name 'lulufanclub'")
   752         self.execute("SET U in_group G WHERE G name 'lulufanclub'")
   753         peid = self.execute("INSERT Personne X: X prenom 'lulu', X nom 'petit'")[0][0]
   753         peid = self.execute("INSERT Personne X: X prenom 'lulu', X nom 'petit'")[0][0]
   754         rset = self.execute("Any X WHERE X prenom 'lulu',"
   754         rset = self.execute("Any X WHERE X prenom 'lulu',"
   755                             "EXISTS (U in_group G, G name 'lulufanclub' OR G name 'managers');")
   755                             "EXISTS (U in_group G, G name 'lulufanclub' OR G name 'managers');")
   756         self.assertEquals(rset.rows, [[peid]])
   756         self.assertEquals(rset.rows, [[peid]])
   762         self.assertEquals(eid, 1)
   762         self.assertEquals(eid, 1)
   763         login = self.execute('Any L WHERE X login "admin", X identity Y, Y login L')[0][0]
   763         login = self.execute('Any L WHERE X login "admin", X identity Y, Y login L')[0][0]
   764         self.assertEquals(login, 'admin')
   764         self.assertEquals(login, 'admin')
   765 
   765 
   766     def test_select_date_mathexp(self):
   766     def test_select_date_mathexp(self):
   767         rset = self.execute('Any X, TODAY - CD WHERE X is EUser, X creation_date CD')
   767         rset = self.execute('Any X, TODAY - CD WHERE X is CWUser, X creation_date CD')
   768         self.failUnless(rset)
   768         self.failUnless(rset)
   769         self.failUnlessEqual(rset.description[0][1], 'Interval')
   769         self.failUnlessEqual(rset.description[0][1], 'Interval')
   770         eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0]
   770         eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0]
   771         rset = self.execute('Any X, NOW - CD WHERE X is Personne, X creation_date CD')
   771         rset = self.execute('Any X, NOW - CD WHERE X is Personne, X creation_date CD')
   772         self.failUnlessEqual(rset.description[0][1], 'Interval')
   772         self.failUnlessEqual(rset.description[0][1], 'Interval')
   773         # sqlite bug
       
   774         #from mx.DateTime import DateTimeDeltaType
       
   775         #self.assertIsInstance(rset[0][1], DateTimeDeltaType) 
       
   776         #self.failUnless(rset[0][1].seconds > 0)
       
   777 
   773 
   778     def test_select_subquery_aggregat(self):
   774     def test_select_subquery_aggregat(self):
   779         # percent users by groups
   775         # percent users by groups
   780         self.execute('SET X in_group G WHERE G name "users"')
   776         self.execute('SET X in_group G WHERE G name "users"')
   781         rset = self.execute('Any GN, COUNT(X)*100/T GROUPBY GN ORDERBY 2,1'
   777         rset = self.execute('Any GN, COUNT(X)*100/T GROUPBY GN ORDERBY 2,1'
   782                             ' WHERE G name GN, X in_group G'
   778                             ' WHERE G name GN, X in_group G'
   783                             ' WITH T BEING (Any COUNT(U) WHERE U is EUser)')
   779                             ' WITH T BEING (Any COUNT(U) WHERE U is CWUser)')
   784         self.assertEquals(rset.rows, [[u'guests', 50], [u'managers', 50], [u'users', 100]])
   780         self.assertEquals(rset.rows, [[u'guests', 50], [u'managers', 50], [u'users', 100]])
   785         self.assertEquals(rset.description, [('String', 'Int'), ('String', 'Int'), ('String', 'Int')])
   781         self.assertEquals(rset.description, [('String', 'Int'), ('String', 'Int'), ('String', 'Int')])
   786 
   782 
   787     def test_select_subquery_const(self):
   783     def test_select_subquery_const(self):
   788         rset = self.execute('Any X WITH X BEING ((Any NULL) UNION (Any "toto"))')
   784         rset = self.execute('Any X WITH X BEING ((Any NULL) UNION (Any "toto"))')
   789         self.assertEquals(rset.rows, [[None], ['toto']])
   785         self.assertEquals(rset.rows, [[None], ['toto']])
   790         self.assertEquals(rset.description, [(None,), ('String',)])
   786         self.assertEquals(rset.description, [(None,), ('String',)])
   791                           
   787 
   792     # insertion queries tests #################################################
   788     # insertion queries tests #################################################
   793     
   789 
   794     def test_insert_is(self):
   790     def test_insert_is(self):
   795         eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0]
   791         eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0]
   796         etype, = self.execute("Any TN WHERE X is T, X eid %s, T name TN" % eid)[0]
   792         etype, = self.execute("Any TN WHERE X is T, X eid %s, T name TN" % eid)[0]
   797         self.assertEquals(etype, 'Personne')
   793         self.assertEquals(etype, 'Personne')
   798         self.execute("INSERT Personne X: X nom 'managers'")
   794         self.execute("INSERT Personne X: X nom 'managers'")
   799     
   795 
   800     def test_insert_1(self):
   796     def test_insert_1(self):
   801         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   797         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   802         self.assertEquals(len(rset.rows), 1)
   798         self.assertEquals(len(rset.rows), 1)
   803         self.assertEquals(rset.description, [('Personne',)])
   799         self.assertEquals(rset.description, [('Personne',)])
   804         rset = self.execute('Personne X WHERE X nom "bidule"')
   800         rset = self.execute('Personne X WHERE X nom "bidule"')
   821 
   817 
   822     def test_insert_3(self):
   818     def test_insert_3(self):
   823         self.execute("INSERT Personne X: X nom Y WHERE U login 'admin', U login Y")
   819         self.execute("INSERT Personne X: X nom Y WHERE U login 'admin', U login Y")
   824         rset = self.execute('Personne X WHERE X nom "admin"')
   820         rset = self.execute('Personne X WHERE X nom "admin"')
   825         self.assert_(rset.rows)
   821         self.assert_(rset.rows)
   826         self.assertEquals(rset.description, [('Personne',)])        
   822         self.assertEquals(rset.description, [('Personne',)])
   827 
   823 
   828     def test_insert_4(self):
   824     def test_insert_4(self):
   829         self.execute("INSERT Societe Y: Y nom 'toto'")
   825         self.execute("INSERT Societe Y: Y nom 'toto'")
   830         self.execute("INSERT Personne X: X nom 'bidule', X travaille Y WHERE Y nom 'toto'")
   826         self.execute("INSERT Personne X: X nom 'bidule', X travaille Y WHERE Y nom 'toto'")
   831         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   827         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   832         self.assert_(rset.rows)
   828         self.assert_(rset.rows)
   833         self.assertEquals(rset.description, [('Personne', 'Societe',)])
   829         self.assertEquals(rset.description, [('Personne', 'Societe',)])
   834         
   830 
   835     def test_insert_4bis(self):
   831     def test_insert_4bis(self):
   836         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   832         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   837         seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s",
   833         seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s",
   838                              {'x': str(peid)})[0][0]
   834                              {'x': str(peid)})[0][0]
   839         self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 1)
   835         self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 1)
   840         self.execute("INSERT Personne X: X nom 'chouette', X travaille Y WHERE Y eid %(x)s",
   836         self.execute("INSERT Personne X: X nom 'chouette', X travaille Y WHERE Y eid %(x)s",
   841                       {'x': str(seid)})
   837                       {'x': str(seid)})
   842         self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 2)
   838         self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 2)
   843         
   839 
   844     def test_insert_4ter(self):
   840     def test_insert_4ter(self):
   845         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   841         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   846         seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s",
   842         seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s",
   847                              {'x': unicode(peid)})[0][0]
   843                              {'x': unicode(peid)})[0][0]
   848         self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 1)
   844         self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 1)
   886                           self.execute,
   882                           self.execute,
   887                           "INSERT Personne X: X nom 'toto', X has_text 'tutu'")
   883                           "INSERT Personne X: X nom 'toto', X has_text 'tutu'")
   888 
   884 
   889         self.assertRaises(QueryError,
   885         self.assertRaises(QueryError,
   890                           self.execute,
   886                           self.execute,
   891                           "INSERT EUser X: X login 'toto', X eid %s" % cnx.user(self.session).eid)
   887                           "INSERT CWUser X: X login 'toto', X eid %s" % cnx.user(self.session).eid)
   892 
   888 
   893     def test_insertion_description_with_where(self):
   889     def test_insertion_description_with_where(self):
   894         rset = self.execute('INSERT EUser E, EmailAddress EM: E login "X", E upassword "X", '
   890         rset = self.execute('INSERT CWUser E, EmailAddress EM: E login "X", E upassword "X", '
   895                             'E primary_email EM, EM address "X", E in_group G '
   891                             'E primary_email EM, EM address "X", E in_group G '
   896                             'WHERE G name "managers"')
   892                             'WHERE G name "managers"')
   897         self.assertEquals(list(rset.description[0]), ['EUser', 'EmailAddress'])
   893         self.assertEquals(list(rset.description[0]), ['CWUser', 'EmailAddress'])
   898     
   894 
   899     # deletion queries tests ##################################################
   895     # deletion queries tests ##################################################
   900 
   896 
   901     def test_delete_1(self):
   897     def test_delete_1(self):
   902         self.execute("INSERT Personne Y: Y nom 'toto'")
   898         self.execute("INSERT Personne Y: Y nom 'toto'")
   903         rset = self.execute('Personne X WHERE X nom "toto"')
   899         rset = self.execute('Personne X WHERE X nom "toto"')
   904         self.assertEqual(len(rset.rows), 1)
   900         self.assertEqual(len(rset.rows), 1)
   905         self.execute("DELETE Personne Y WHERE Y nom 'toto'")
   901         self.execute("DELETE Personne Y WHERE Y nom 'toto'")
   906         rset = self.execute('Personne X WHERE X nom "toto"')
   902         rset = self.execute('Personne X WHERE X nom "toto"')
   907         self.assertEqual(len(rset.rows), 0)
   903         self.assertEqual(len(rset.rows), 0)
   908         
   904 
   909     def test_delete_2(self):
   905     def test_delete_2(self):
   910         rset = self.execute("INSERT Personne X, Personne Y, Societe Z : X nom 'syt', Y nom 'adim', Z nom 'Logilab', X travaille Z, Y travaille Z")
   906         rset = self.execute("INSERT Personne X, Personne Y, Societe Z : X nom 'syt', Y nom 'adim', Z nom 'Logilab', X travaille Z, Y travaille Z")
   911         self.assertEquals(len(rset), 1)
   907         self.assertEquals(len(rset), 1)
   912         self.assertEquals(len(rset[0]), 3)
   908         self.assertEquals(len(rset[0]), 3)
   913         self.assertEquals(rset.description[0], ('Personne', 'Personne', 'Societe'))
   909         self.assertEquals(rset.description[0], ('Personne', 'Personne', 'Societe'))
   962         sqlc = s.pool['system']
   958         sqlc = s.pool['system']
   963         sqlc.execute('SELECT * FROM recipients_relation')
   959         sqlc.execute('SELECT * FROM recipients_relation')
   964         self.assertEquals(len(sqlc.fetchall()), 0)
   960         self.assertEquals(len(sqlc.fetchall()), 0)
   965         sqlc.execute('SELECT * FROM owned_by_relation WHERE eid_from=%s'%eeid)
   961         sqlc.execute('SELECT * FROM owned_by_relation WHERE eid_from=%s'%eeid)
   966         self.assertEquals(len(sqlc.fetchall()), 0)
   962         self.assertEquals(len(sqlc.fetchall()), 0)
   967             
   963 
   968     def test_nonregr_delete_cache2(self):
   964     def test_nonregr_delete_cache2(self):
   969         eid = self.execute("INSERT Folder T: T name 'toto'")[0][0]
   965         eid = self.execute("INSERT Folder T: T name 'toto'")[0][0]
   970         self.commit()
   966         self.commit()
   971         # fill the cache
   967         # fill the cache
   972         self.execute("Any X WHERE X eid %(x)s", {'x': eid}, 'x')
   968         self.execute("Any X WHERE X eid %(x)s", {'x': eid}, 'x')
   981         self.assertEquals(rset.rows, [])
   977         self.assertEquals(rset.rows, [])
   982         rset = self.execute("Folder X WHERE X eid %(x)s", {'x': eid}, 'x')
   978         rset = self.execute("Folder X WHERE X eid %(x)s", {'x': eid}, 'x')
   983         self.assertEquals(rset.rows, [])
   979         self.assertEquals(rset.rows, [])
   984         rset = self.execute("Folder X WHERE X eid %s" %eid)
   980         rset = self.execute("Folder X WHERE X eid %s" %eid)
   985         self.assertEquals(rset.rows, [])
   981         self.assertEquals(rset.rows, [])
   986         
   982 
   987     # update queries tests ####################################################
   983     # update queries tests ####################################################
   988 
   984 
   989     def test_update_1(self):
   985     def test_update_1(self):
   990         self.execute("INSERT Personne Y: Y nom 'toto'")
   986         self.execute("INSERT Personne Y: Y nom 'toto'")
   991         rset = self.execute('Personne X WHERE X nom "toto"')
   987         rset = self.execute('Personne X WHERE X nom "toto"')
   992         self.assertEqual(len(rset.rows), 1)
   988         self.assertEqual(len(rset.rows), 1)
   993         self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'")
   989         self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'")
   994         rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z')
   990         rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z')
   995         self.assertEqual(tuplify(rset.rows), [('tutu', 'original')])
   991         self.assertEqual(tuplify(rset.rows), [('tutu', 'original')])
   996         
   992 
   997     def test_update_2(self):
   993     def test_update_2(self):
   998         self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
   994         self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
   999         #rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto"')
   995         #rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto"')
  1000         #self.assertEqual(len(rset.rows), 1)
   996         #self.assertEqual(len(rset.rows), 1)
  1001         #rset = self.execute('Any X, Y WHERE X travaille Y')
   997         #rset = self.execute('Any X, Y WHERE X travaille Y')
  1002         #self.assertEqual(len(rset.rows), 0)
   998         #self.assertEqual(len(rset.rows), 0)
  1003         self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'")
   999         self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'")
  1004         rset = self.execute('Any X, Y WHERE X travaille Y')
  1000         rset = self.execute('Any X, Y WHERE X travaille Y')
  1005         self.assertEqual(len(rset.rows), 1)
  1001         self.assertEqual(len(rset.rows), 1)
  1006         
  1002 
  1007     def test_update_2bis(self):
  1003     def test_update_2bis(self):
  1008         rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
  1004         rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
  1009         eid1, eid2 = rset[0][0], rset[0][1]
  1005         eid1, eid2 = rset[0][0], rset[0][1]
  1010         self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s",
  1006         self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s",
  1011                       {'x': str(eid1), 'y': str(eid2)})
  1007                       {'x': str(eid1), 'y': str(eid2)})
  1012         rset = self.execute('Any X, Y WHERE X travaille Y')
  1008         rset = self.execute('Any X, Y WHERE X travaille Y')
  1013         self.assertEqual(len(rset.rows), 1)
  1009         self.assertEqual(len(rset.rows), 1)
  1014         
  1010 
  1015     def test_update_2ter(self):
  1011     def test_update_2ter(self):
  1016         rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
  1012         rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
  1017         eid1, eid2 = rset[0][0], rset[0][1]
  1013         eid1, eid2 = rset[0][0], rset[0][1]
  1018         self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s",
  1014         self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s",
  1019                       {'x': unicode(eid1), 'y': unicode(eid2)})
  1015                       {'x': unicode(eid1), 'y': unicode(eid2)})
  1020         rset = self.execute('Any X, Y WHERE X travaille Y')
  1016         rset = self.execute('Any X, Y WHERE X travaille Y')
  1021         self.assertEqual(len(rset.rows), 1)
  1017         self.assertEqual(len(rset.rows), 1)
  1022         
  1018 
  1023 ##     def test_update_4(self):
  1019 ##     def test_update_4(self):
  1024 ##         self.execute("SET X know Y WHERE X ami Y")
  1020 ##         self.execute("SET X know Y WHERE X ami Y")
  1025         
  1021 
  1026     def test_update_multiple1(self):
  1022     def test_update_multiple1(self):
  1027         peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1023         peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1028         peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0]
  1024         peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0]
  1029         self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'")
  1025         self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'")
  1030         self.assertEquals(self.execute('Any X WHERE X nom "toto"').rows, [[peid1]])
  1026         self.assertEquals(self.execute('Any X WHERE X nom "toto"').rows, [[peid1]])
  1031         self.assertEquals(self.execute('Any X WHERE X nom "tutu"').rows, [[peid2]])
  1027         self.assertEquals(self.execute('Any X WHERE X nom "tutu"').rows, [[peid2]])
  1032 
  1028 
  1033     def test_update_multiple2(self):
  1029     def test_update_multiple2(self):
  1034         ueid = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto'")[0][0]
  1030         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0]
  1035         peid1 = self.execute("INSERT Personne Y: Y nom 'turlu'")[0][0]
  1031         peid1 = self.execute("INSERT Personne Y: Y nom 'turlu'")[0][0]
  1036         peid2 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1032         peid2 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1037         self.execute('SET P1 owned_by U, P2 owned_by U '
  1033         self.execute('SET P1 owned_by U, P2 owned_by U '
  1038                      'WHERE P1 eid %s, P2 eid %s, U eid %s' % (peid1, peid2, ueid))
  1034                      'WHERE P1 eid %s, P2 eid %s, U eid %s' % (peid1, peid2, ueid))
  1039         self.failUnless(self.execute('Any X WHERE X eid %s, X owned_by U, U eid %s'
  1035         self.failUnless(self.execute('Any X WHERE X eid %s, X owned_by U, U eid %s'
  1056     def test_update_string_concat(self):
  1052     def test_update_string_concat(self):
  1057         beid = self.execute("INSERT Bookmark Y: Y title 'toto', Y path '/view'")[0][0]
  1053         beid = self.execute("INSERT Bookmark Y: Y title 'toto', Y path '/view'")[0][0]
  1058         self.execute('SET X title XN + %(suffix)s WHERE X is Bookmark, X title XN', {'suffix': u'-moved'})
  1054         self.execute('SET X title XN + %(suffix)s WHERE X is Bookmark, X title XN', {'suffix': u'-moved'})
  1059         newname = self.execute('Any XN WHERE X eid %(x)s, X title XN', {'x': beid}, 'x')[0][0]
  1055         newname = self.execute('Any XN WHERE X eid %(x)s, X title XN', {'x': beid}, 'x')[0][0]
  1060         self.assertEquals(newname, 'toto-moved')
  1056         self.assertEquals(newname, 'toto-moved')
  1061                        
  1057 
  1062     def test_update_query_error(self):
  1058     def test_update_query_error(self):
  1063         self.execute("INSERT Personne Y: Y nom 'toto'")
  1059         self.execute("INSERT Personne Y: Y nom 'toto'")
  1064         self.assertRaises(Exception, self.execute, "SET X nom 'toto', X is Personne")
  1060         self.assertRaises(Exception, self.execute, "SET X nom 'toto', X is Personne")
  1065         self.assertRaises(QueryError, self.execute, "SET X nom 'toto', X has_text 'tutu' WHERE X is Personne")
  1061         self.assertRaises(QueryError, self.execute, "SET X nom 'toto', X has_text 'tutu' WHERE X is Personne")
  1066         self.assertRaises(QueryError, self.execute, "SET X login 'tutu', X eid %s" % cnx.user(self.session).eid)
  1062         self.assertRaises(QueryError, self.execute, "SET X login 'tutu', X eid %s" % cnx.user(self.session).eid)
  1067 
  1063 
  1068        
  1064 
  1069     # upassword encryption tests #################################################
  1065     # upassword encryption tests #################################################
  1070     
  1066 
  1071     def test_insert_upassword(self):
  1067     def test_insert_upassword(self):
  1072         rset = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto'")
  1068         rset = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")
  1073         self.assertEquals(len(rset.rows), 1)
  1069         self.assertEquals(len(rset.rows), 1)
  1074         self.assertEquals(rset.description, [('EUser',)])
  1070         self.assertEquals(rset.description, [('CWUser',)])
  1075         self.assertRaises(Unauthorized,
  1071         self.assertRaises(Unauthorized,
  1076                           self.execute, "Any P WHERE X is EUser, X login 'bob', X upassword P")
  1072                           self.execute, "Any P WHERE X is CWUser, X login 'bob', X upassword P")
  1077         cursor = self.pool['system']
  1073         cursor = self.pool['system']
  1078         cursor.execute("SELECT %supassword from %sEUser WHERE %slogin='bob'"
  1074         cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'"
  1079                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1075                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1080         passwd = cursor.fetchone()[0].getvalue()
  1076         passwd = cursor.fetchone()[0].getvalue()
  1081         self.assertEquals(passwd, crypt_password('toto', passwd[:2])) 
  1077         self.assertEquals(passwd, crypt_password('toto', passwd[:2]))
  1082         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 CWUser, X login 'bob', X upassword '%s'" % passwd)
  1083         self.assertEquals(len(rset.rows), 1)
  1079         self.assertEquals(len(rset.rows), 1)
  1084         self.assertEquals(rset.description, [('EUser',)])
  1080         self.assertEquals(rset.description, [('CWUser',)])
  1085         
  1081 
  1086     def test_update_upassword(self):
  1082     def test_update_upassword(self):
  1087         cursor = self.pool['system']
  1083         cursor = self.pool['system']
  1088         rset = self.execute("INSERT EUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'})
  1084         rset = self.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'})
  1089         self.assertEquals(rset.description[0][0], 'EUser')
  1085         self.assertEquals(rset.description[0][0], 'CWUser')
  1090         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 CWUser, X login 'bob'",
  1091                             {'pwd': 'tutu'})
  1087                             {'pwd': 'tutu'})
  1092         cursor.execute("SELECT %supassword from %sEUser WHERE %slogin='bob'"
  1088         cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'"
  1093                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1089                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1094         passwd = cursor.fetchone()[0].getvalue()
  1090         passwd = cursor.fetchone()[0].getvalue()
  1095         self.assertEquals(passwd, crypt_password('tutu', passwd[:2])) 
  1091         self.assertEquals(passwd, crypt_password('tutu', passwd[:2]))
  1096         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 CWUser, X login 'bob', X upassword '%s'" % passwd)
  1097         self.assertEquals(len(rset.rows), 1)
  1093         self.assertEquals(len(rset.rows), 1)
  1098         self.assertEquals(rset.description, [('EUser',)])
  1094         self.assertEquals(rset.description, [('CWUser',)])
  1099 
  1095 
  1100     # non regression tests ####################################################
  1096     # non regression tests ####################################################
  1101     
  1097 
  1102     def test_nonregr_1(self):
  1098     def test_nonregr_1(self):
  1103         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
  1099         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
  1104         self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'")
  1100         self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'")
  1105         rset = self.execute('Any X WHERE T tags X')
  1101         rset = self.execute('Any X WHERE T tags X')
  1106         self.assertEquals(len(rset.rows), 1, rset.rows)
  1102         self.assertEquals(len(rset.rows), 1, rset.rows)
  1109         rset = self.execute('Any T WHERE T tags X')
  1105         rset = self.execute('Any T WHERE T tags X')
  1110         self.assertEquals(rset.rows, [[teid]])
  1106         self.assertEquals(rset.rows, [[teid]])
  1111 
  1107 
  1112     def test_nonregr_2(self):
  1108     def test_nonregr_2(self):
  1113         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
  1109         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
  1114         geid = self.execute("EGroup G WHERE G name 'users'")[0][0]
  1110         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
  1115         self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
  1111         self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
  1116                        {'g': geid, 't': teid})
  1112                        {'g': geid, 't': teid})
  1117         rset = self.execute('Any X WHERE E eid %(x)s, E tags X',
  1113         rset = self.execute('Any X WHERE E eid %(x)s, E tags X',
  1118                               {'x': teid})
  1114                               {'x': teid})
  1119         self.assertEquals(rset.rows, [[geid]])
  1115         self.assertEquals(rset.rows, [[geid]])
  1120         
  1116 
  1121     def test_nonregr_3(self):
  1117     def test_nonregr_3(self):
  1122         """bad sql generated on the second query (destination_state is not
  1118         """bad sql generated on the second query (destination_state is not
  1123         detected as an inlined relation)
  1119         detected as an inlined relation)
  1124         """
  1120         """
  1125         rset = self.execute('Any S,ES,T WHERE S state_of ET, ET name "EUser",'
  1121         rset = self.execute('Any S,ES,T WHERE S state_of ET, ET name "CWUser",'
  1126                              'ES allowed_transition T, T destination_state S')
  1122                              'ES allowed_transition T, T destination_state S')
  1127         self.assertEquals(len(rset.rows), 2)
  1123         self.assertEquals(len(rset.rows), 2)
  1128 
  1124 
  1129     def test_nonregr_4(self):
  1125     def test_nonregr_4(self):
  1130         # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3
  1126         # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3
  1131         # union queries and that make for instance a 266Ko sql query which is refused
  1127         # union queries and that make for instance a 266Ko sql query which is refused
  1132         # by the server (or client lib)
  1128         # by the server (or client lib)
  1133         rset = self.execute('Any ER,SE,OE WHERE SE name "Comment", ER name "comments", OE name "Comment",'
  1129         rset = self.execute('Any ER,SE,OE WHERE SE name "Comment", ER name "comments", OE name "Comment",'
  1134                             'ER is ERType, SE is EEType, OE is EEType')
  1130                             'ER is CWRType, SE is CWEType, OE is CWEType')
  1135         self.assertEquals(len(rset), 1)
  1131         self.assertEquals(len(rset), 1)
  1136 
  1132 
  1137     def test_nonregr_5(self):
  1133     def test_nonregr_5(self):
  1138         # jpl #15505: equivalent queries returning different result sets
  1134         # jpl #15505: equivalent queries returning different result sets
  1139         teid1 = self.execute("INSERT Folder X: X name 'hop'")[0][0]
  1135         teid1 = self.execute("INSERT Folder X: X name 'hop'")[0][0]
  1153                              'N filed_under T, W concerne N,'
  1149                              'N filed_under T, W concerne N,'
  1154                              'W filed_under A, A eid %s' % (teid1, teid2))
  1150                              'W filed_under A, A eid %s' % (teid1, teid2))
  1155         self.assertEquals(rset1.rows, rset2.rows)
  1151         self.assertEquals(rset1.rows, rset2.rows)
  1156         self.assertEquals(rset1.rows, rset3.rows)
  1152         self.assertEquals(rset1.rows, rset3.rows)
  1157         self.assertEquals(rset1.rows, rset4.rows)
  1153         self.assertEquals(rset1.rows, rset4.rows)
  1158         
  1154 
  1159     def test_nonregr_6(self):
  1155     def test_nonregr_6(self):
  1160         self.execute('Any N,COUNT(S) GROUPBY N ORDERBY COUNT(N) WHERE S name N, S is State')
  1156         self.execute('Any N,COUNT(S) GROUPBY N ORDERBY COUNT(N) WHERE S name N, S is State')
  1161         
  1157 
  1162     def test_sqlite_encoding(self):
  1158     def test_sqlite_encoding(self):
  1163         """XXX this test was trying to show a bug on use of lower which only
  1159         """XXX this test was trying to show a bug on use of lower which only
  1164         occurs with non ascii string and misconfigured locale
  1160         occurs with non ascii string and misconfigured locale
  1165         """
  1161         """
  1166         self.execute("INSERT Tag X: X name %(name)s,"
  1162         self.execute("INSERT Tag X: X name %(name)s,"
  1217         cause: old variable ref inserted into a fresh rqlst copy
  1213         cause: old variable ref inserted into a fresh rqlst copy
  1218         (in RQLSpliter._complex_select_plan)
  1214         (in RQLSpliter._complex_select_plan)
  1219         """
  1215         """
  1220         self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed')
  1216         self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed')
  1221         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')
  1222     
  1218 
  1223     def test_nonregr_extra_joins(self):
  1219     def test_nonregr_extra_joins(self):
  1224         ueid = self.session.user.eid
  1220         ueid = self.session.user.eid
  1225         teid1 = self.execute("INSERT Folder X: X name 'folder1'")[0][0]
  1221         teid1 = self.execute("INSERT Folder X: X name 'folder1'")[0][0]
  1226         teid2 = self.execute("INSERT Folder X: X name 'folder2'")[0][0]
  1222         teid2 = self.execute("INSERT Folder X: X name 'folder2'")[0][0]
  1227         neid1 = self.execute("INSERT Note X: X para 'note1'")[0][0]
  1223         neid1 = self.execute("INSERT Note X: X para 'note1'")[0][0]
  1245         self.assertEquals(len(rset), 1)
  1241         self.assertEquals(len(rset), 1)
  1246 
  1242 
  1247 
  1243 
  1248     def test_nonregr_set_datetime(self):
  1244     def test_nonregr_set_datetime(self):
  1249         # huum, psycopg specific
  1245         # huum, psycopg specific
  1250         self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': today()})
  1246         self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()})
  1251 
  1247 
  1252     def test_nonregr_set_query(self):
  1248     def test_nonregr_set_query(self):
  1253         ueid = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto'")[0][0]
  1249         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0]
  1254         self.execute("SET E in_group G, E in_state S, "
  1250         self.execute("SET E in_group G, E in_state S, "
  1255                       "E firstname %(firstname)s, E surname %(surname)s "
  1251                       "E firstname %(firstname)s, E surname %(surname)s "
  1256                       "WHERE E eid %(x)s, G name 'users', S name 'activated'",
  1252                       "WHERE E eid %(x)s, G name 'users', S name 'activated'",
  1257                       {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x')
  1253                       {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x')
  1258         
  1254 
  1259     def test_nonregr_u_owned_by_u(self):
  1255     def test_nonregr_u_owned_by_u(self):
  1260         ueid = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto', X in_group G "
  1256         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G "
  1261                              "WHERE G name 'users'")[0][0]
  1257                              "WHERE G name 'users'")[0][0]
  1262         rset = self.execute("EUser U")
  1258         rset = self.execute("CWUser U")
  1263         self.assertEquals(len(rset), 3) # bob + admin + anon
  1259         self.assertEquals(len(rset), 3) # bob + admin + anon
  1264         rset = self.execute("Any U WHERE NOT U owned_by U")
  1260         rset = self.execute("Any U WHERE NOT U owned_by U")
  1265         self.assertEquals(len(rset), 0) # even admin created at repo initialization time should belong to itself
  1261         self.assertEquals(len(rset), 0) # even admin created at repo initialization time should belong to itself
  1266 
  1262 
  1267     def test_nonreg_update_index(self):
  1263     def test_nonreg_update_index(self):
  1268         # this is the kind of queries generated by "cubicweb-ctl db-check -ry"
  1264         # this is the kind of queries generated by "cubicweb-ctl db-check -ry"
  1269         self.execute("SET X description D WHERE X is State, X description D")
  1265         self.execute("SET X description D WHERE X is State, X description D")
  1270 
  1266 
  1271     def test_nonregr_is(self):
  1267     def test_nonregr_is(self):
  1272         uteid = self.execute('Any ET WHERE ET name "EUser"')[0][0]
  1268         uteid = self.execute('Any ET WHERE ET name "CWUser"')[0][0]
  1273         self.execute('Any X, ET WHERE X is ET, ET eid %s' % uteid)
  1269         self.execute('Any X, ET WHERE X is ET, ET eid %s' % uteid)
  1274 
  1270 
  1275     def test_nonregr_orderby(self):
  1271     def test_nonregr_orderby(self):
  1276         seid = self.execute('Any X WHERE X name "activated"')[0][0]
  1272         seid = self.execute('Any X WHERE X name "activated"')[0][0]
  1277         self.execute('Any X,S, MAX(T) GROUPBY X,S ORDERBY S WHERE X is EUser, T tags X, S eid IN(%s), X in_state S' % seid)
  1273         self.execute('Any X,S, MAX(T) GROUPBY X,S ORDERBY S WHERE X is CWUser, T tags X, S eid IN(%s), X in_state S' % seid)
  1278 
  1274 
  1279     def test_nonregr_solution_cache(self):
  1275     def test_nonregr_solution_cache(self):
  1280         self.skip('XXX should be fixed or documented') # (doesn't occur if cache key is provided.)
  1276         self.skip('XXX should be fixed or documented') # (doesn't occur if cache key is provided.)
  1281         rset = self.execute('Any X WHERE X is EUser, X eid %(x)s', {'x':self.ueid})
  1277         rset = self.execute('Any X WHERE X is CWUser, X eid %(x)s', {'x':self.ueid})
  1282         self.assertEquals(len(rset), 1)
  1278         self.assertEquals(len(rset), 1)
  1283         rset = self.execute('Any X WHERE X is EUser, X eid %(x)s', {'x':12345})
  1279         rset = self.execute('Any X WHERE X is CWUser, X eid %(x)s', {'x':12345})
  1284         self.assertEquals(len(rset), 0)
  1280         self.assertEquals(len(rset), 0)
  1285 
  1281 
  1286     def test_nonregr_final_norestr(self):
  1282     def test_nonregr_final_norestr(self):
  1287         self.assertRaises(BadRQLQuery, self.execute, 'Date X')
  1283         self.assertRaises(BadRQLQuery, self.execute, 'Date X')
  1288 
  1284