server/test/unittest_querier.py
changeset 6366 1806148d6ce8
parent 6225 a176e68b7d0d
parent 6360 1edfc0f3bb1d
child 6427 c8a5ac2d1eaa
equal deleted inserted replaced
6333:e3994fcc21c3 6366:1806148d6ce8
    55 TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING
    55 TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING
    56 
    56 
    57 class MakeSchemaTC(TestCase):
    57 class MakeSchemaTC(TestCase):
    58     def test_known_values(self):
    58     def test_known_values(self):
    59         solution = {'A': 'String', 'B': 'CWUser'}
    59         solution = {'A': 'String', 'B': 'CWUser'}
    60         self.assertEquals(make_schema((Variable('A'), Variable('B')), solution,
    60         self.assertEqual(make_schema((Variable('A'), Variable('B')), solution,
    61                                       'table0', TYPEMAP),
    61                                       'table0', TYPEMAP),
    62                           ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'}))
    62                           ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'}))
    63 
    63 
    64 
    64 
    65 repo, cnx = init_test_database()
    65 repo, cnx = init_test_database()
    82         pass
    82         pass
    83 
    83 
    84     def test_preprocess_1(self):
    84     def test_preprocess_1(self):
    85         reid = self.execute('Any X WHERE X is CWRType, X name "owned_by"')[0][0]
    85         reid = self.execute('Any X WHERE X is CWRType, X name "owned_by"')[0][0]
    86         rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid})
    86         rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid})
    87         self.assertEquals(rqlst.solutions, [{'RDEF': 'CWAttribute'}, {'RDEF': 'CWRelation'}])
    87         self.assertEqual(rqlst.solutions, [{'RDEF': 'CWAttribute'}, {'RDEF': 'CWRelation'}])
    88 
    88 
    89     def test_preprocess_2(self):
    89     def test_preprocess_2(self):
    90         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
    90         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
    91         #geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
    91         #geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
    92         #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
    92         #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
    93         #             {'g': geid, 't': teid}, 'g')
    93         #             {'g': geid, 't': teid}, 'g')
    94         rqlst = self._prepare('Any X WHERE E eid %(x)s, E tags X', {'x': teid})
    94         rqlst = self._prepare('Any X WHERE E eid %(x)s, E tags X', {'x': teid})
    95         # the query may be optimized, should keep only one solution
    95         # the query may be optimized, should keep only one solution
    96         # (any one, etype will be discarded)
    96         # (any one, etype will be discarded)
    97         self.assertEquals(len(rqlst.solutions), 1)
    97         self.assertEqual(len(rqlst.solutions), 1)
    98 
    98 
    99     def test_preprocess_security(self):
    99     def test_preprocess_security(self):
   100         plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN '
   100         plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN '
   101                                   'WHERE X is ET, ET name ETN')
   101                                   'WHERE X is ET, ET name ETN')
   102         plan.session = self.user_groups_session('users')
   102         plan.session = self.user_groups_session('users')
   103         union = plan.rqlst
   103         union = plan.rqlst
   104         plan.preprocess(union)
   104         plan.preprocess(union)
   105         self.assertEquals(len(union.children), 1)
   105         self.assertEqual(len(union.children), 1)
   106         self.assertEquals(len(union.children[0].with_), 1)
   106         self.assertEqual(len(union.children[0].with_), 1)
   107         subq = union.children[0].with_[0].query
   107         subq = union.children[0].with_[0].query
   108         self.assertEquals(len(subq.children), 3)
   108         self.assertEqual(len(subq.children), 3)
   109         self.assertEquals([t.as_string() for t in union.children[0].selection],
   109         self.assertEqual([t.as_string() for t in union.children[0].selection],
   110                           ['ETN','COUNT(X)'])
   110                           ['ETN','COUNT(X)'])
   111         self.assertEquals([t.as_string() for t in union.children[0].groupby],
   111         self.assertEqual([t.as_string() for t in union.children[0].groupby],
   112                           ['ETN'])
   112                           ['ETN'])
   113         partrqls = sorted(((rqlst.as_string(), rqlst.solutions) for rqlst in subq.children))
   113         partrqls = sorted(((rqlst.as_string(), rqlst.solutions) for rqlst in subq.children))
   114         rql, solutions = partrqls[0]
   114         rql, solutions = partrqls[0]
   115         self.assertEquals(rql,
   115         self.assertEqual(rql,
   116                           'Any ETN,X WHERE X is ET, ET name ETN, (EXISTS(X owned_by %(B)s))'
   116                           'Any ETN,X WHERE X is ET, ET name ETN, (EXISTS(X owned_by %(B)s))'
   117                           ' OR ((((EXISTS(D concerne C?, C owned_by %(B)s, X identity D, C is Division, D is Affaire))'
   117                           ' OR ((((EXISTS(D concerne C?, C owned_by %(B)s, X identity D, C is Division, D is Affaire))'
   118                           ' OR (EXISTS(H concerne G?, G owned_by %(B)s, G is SubDivision, X identity H, H is Affaire)))'
   118                           ' OR (EXISTS(H concerne G?, G owned_by %(B)s, G is SubDivision, X identity H, H is Affaire)))'
   119                           ' OR (EXISTS(I concerne F?, F owned_by %(B)s, F is Societe, X identity I, I is Affaire)))'
   119                           ' OR (EXISTS(I concerne F?, F owned_by %(B)s, F is Societe, X identity I, I is Affaire)))'
   120                           ' OR (EXISTS(J concerne E?, E owned_by %(B)s, E is Note, X identity J, J is Affaire)))'
   120                           ' OR (EXISTS(J concerne E?, E owned_by %(B)s, E is Note, X identity J, J is Affaire)))'
   121                           ', ET is CWEType, X is Affaire')
   121                           ', ET is CWEType, X is Affaire')
   122         self.assertEquals(solutions, [{'C': 'Division',
   122         self.assertEqual(solutions, [{'C': 'Division',
   123                                        'D': 'Affaire',
   123                                        'D': 'Affaire',
   124                                        'E': 'Note',
   124                                        'E': 'Note',
   125                                        'F': 'Societe',
   125                                        'F': 'Societe',
   126                                        'G': 'SubDivision',
   126                                        'G': 'SubDivision',
   127                                        'H': 'Affaire',
   127                                        'H': 'Affaire',
   128                                        'I': 'Affaire',
   128                                        'I': 'Affaire',
   129                                        'J': 'Affaire',
   129                                        'J': 'Affaire',
   130                                        'X': 'Affaire',
   130                                        'X': 'Affaire',
   131                                        'ET': 'CWEType', 'ETN': 'String'}])
   131                                        'ET': 'CWEType', 'ETN': 'String'}])
   132         rql, solutions = partrqls[1]
   132         rql, solutions = partrqls[1]
   133         self.assertEquals(rql,  'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, X is IN(BaseTransition, Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUniqueTogetherConstraint, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, ExternalUri, File, Folder, Note, Personne, RQLExpression, Societe, State, SubDivision, SubWorkflowExitPoint, Tag, TrInfo, Transition, Workflow, WorkflowTransition)')
   133         self.assertEqual(rql,  'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, X is IN(BaseTransition, Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUniqueTogetherConstraint, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, ExternalUri, File, Folder, Note, Personne, RQLExpression, Societe, State, SubDivision, SubWorkflowExitPoint, Tag, TrInfo, Transition, Workflow, WorkflowTransition)')
   134         self.assertListEquals(sorted(solutions),
   134         self.assertListEqual(sorted(solutions),
   135                               sorted([{'X': 'BaseTransition', 'ETN': 'String', 'ET': 'CWEType'},
   135                               sorted([{'X': 'BaseTransition', 'ETN': 'String', 'ET': 'CWEType'},
   136                                       {'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'},
   136                                       {'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'},
   137                                       {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'},
   137                                       {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'},
   138                                       {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'},
   138                                       {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'},
   139                                       {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'},
   139                                       {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'},
   167                                       {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'},
   167                                       {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'},
   168                                       {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'},
   168                                       {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'},
   169                                       {'X': 'Workflow', 'ETN': 'String', 'ET': 'CWEType'},
   169                                       {'X': 'Workflow', 'ETN': 'String', 'ET': 'CWEType'},
   170                                       {'X': 'WorkflowTransition', 'ETN': 'String', 'ET': 'CWEType'}]))
   170                                       {'X': 'WorkflowTransition', 'ETN': 'String', 'ET': 'CWEType'}]))
   171         rql, solutions = partrqls[2]
   171         rql, solutions = partrqls[2]
   172         self.assertEquals(rql,
   172         self.assertEqual(rql,
   173                           'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), '
   173                           'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), '
   174                           'ET is CWEType, X is Basket')
   174                           'ET is CWEType, X is Basket')
   175         self.assertEquals(solutions, [{'ET': 'CWEType',
   175         self.assertEqual(solutions, [{'ET': 'CWEType',
   176                                        'X': 'Basket',
   176                                        'X': 'Basket',
   177                                        'ETN': 'String',
   177                                        'ETN': 'String',
   178                                        }])
   178                                        }])
   179 
   179 
   180     def test_preprocess_security_aggregat(self):
   180     def test_preprocess_security_aggregat(self):
   181         plan = self._prepare_plan('Any MAX(X)')
   181         plan = self._prepare_plan('Any MAX(X)')
   182         plan.session = self.user_groups_session('users')
   182         plan.session = self.user_groups_session('users')
   183         union = plan.rqlst
   183         union = plan.rqlst
   184         plan.preprocess(union)
   184         plan.preprocess(union)
   185         self.assertEquals(len(union.children), 1)
   185         self.assertEqual(len(union.children), 1)
   186         self.assertEquals(len(union.children[0].with_), 1)
   186         self.assertEqual(len(union.children[0].with_), 1)
   187         subq = union.children[0].with_[0].query
   187         subq = union.children[0].with_[0].query
   188         self.assertEquals(len(subq.children), 3)
   188         self.assertEqual(len(subq.children), 3)
   189         self.assertEquals([t.as_string() for t in union.children[0].selection],
   189         self.assertEqual([t.as_string() for t in union.children[0].selection],
   190                           ['MAX(X)'])
   190                           ['MAX(X)'])
   191 
   191 
   192     def test_preprocess_nonregr(self):
   192     def test_preprocess_nonregr(self):
   193         rqlst = self._prepare('Any S ORDERBY SI WHERE NOT S ecrit_par O, S para SI')
   193         rqlst = self._prepare('Any S ORDERBY SI WHERE NOT S ecrit_par O, S para SI')
   194         self.assertEquals(len(rqlst.solutions), 1)
   194         self.assertEqual(len(rqlst.solutions), 1)
   195 
   195 
   196     def test_build_description(self):
   196     def test_build_description(self):
   197         # should return an empty result set
   197         # should return an empty result set
   198         rset = self.execute('Any X WHERE X eid %(x)s', {'x': self.session.user.eid})
   198         rset = self.execute('Any X WHERE X eid %(x)s', {'x': self.session.user.eid})
   199         self.assertEquals(rset.description[0][0], 'CWUser')
   199         self.assertEqual(rset.description[0][0], 'CWUser')
   200         rset = self.execute('Any 1')
   200         rset = self.execute('Any 1')
   201         self.assertEquals(rset.description[0][0], 'Int')
   201         self.assertEqual(rset.description[0][0], 'Int')
   202         rset = self.execute('Any TRUE')
   202         rset = self.execute('Any TRUE')
   203         self.assertEquals(rset.description[0][0], 'Boolean')
   203         self.assertEqual(rset.description[0][0], 'Boolean')
   204         rset = self.execute('Any "hop"')
   204         rset = self.execute('Any "hop"')
   205         self.assertEquals(rset.description[0][0], 'String')
   205         self.assertEqual(rset.description[0][0], 'String')
   206         rset = self.execute('Any TODAY')
   206         rset = self.execute('Any TODAY')
   207         self.assertEquals(rset.description[0][0], 'Date')
   207         self.assertEqual(rset.description[0][0], 'Date')
   208         rset = self.execute('Any NOW')
   208         rset = self.execute('Any NOW')
   209         self.assertEquals(rset.description[0][0], 'Datetime')
   209         self.assertEqual(rset.description[0][0], 'Datetime')
   210         rset = self.execute('Any %(x)s', {'x': 1})
   210         rset = self.execute('Any %(x)s', {'x': 1})
   211         self.assertEquals(rset.description[0][0], 'Int')
   211         self.assertEqual(rset.description[0][0], 'Int')
   212         rset = self.execute('Any %(x)s', {'x': 1L})
   212         rset = self.execute('Any %(x)s', {'x': 1L})
   213         self.assertEquals(rset.description[0][0], 'Int')
   213         self.assertEqual(rset.description[0][0], 'Int')
   214         rset = self.execute('Any %(x)s', {'x': True})
   214         rset = self.execute('Any %(x)s', {'x': True})
   215         self.assertEquals(rset.description[0][0], 'Boolean')
   215         self.assertEqual(rset.description[0][0], 'Boolean')
   216         rset = self.execute('Any %(x)s', {'x': 1.0})
   216         rset = self.execute('Any %(x)s', {'x': 1.0})
   217         self.assertEquals(rset.description[0][0], 'Float')
   217         self.assertEqual(rset.description[0][0], 'Float')
   218         rset = self.execute('Any %(x)s', {'x': datetime.now()})
   218         rset = self.execute('Any %(x)s', {'x': datetime.now()})
   219         self.assertEquals(rset.description[0][0], 'Datetime')
   219         self.assertEqual(rset.description[0][0], 'Datetime')
   220         rset = self.execute('Any %(x)s', {'x': 'str'})
   220         rset = self.execute('Any %(x)s', {'x': 'str'})
   221         self.assertEquals(rset.description[0][0], 'String')
   221         self.assertEqual(rset.description[0][0], 'String')
   222         rset = self.execute('Any %(x)s', {'x': u'str'})
   222         rset = self.execute('Any %(x)s', {'x': u'str'})
   223         self.assertEquals(rset.description[0][0], 'String')
   223         self.assertEqual(rset.description[0][0], 'String')
   224 
   224 
   225 
   225 
   226 class QuerierTC(BaseQuerierTC):
   226 class QuerierTC(BaseQuerierTC):
   227     repo = repo
   227     repo = repo
   228 
   228 
   242     def test_bytes_storage(self):
   242     def test_bytes_storage(self):
   243         feid = self.execute('INSERT File X: X data_name "foo.pdf", X data_format "text/plain", X data %(data)s',
   243         feid = self.execute('INSERT File X: X data_name "foo.pdf", X data_format "text/plain", X data %(data)s',
   244                             {'data': Binary("xxx")})[0][0]
   244                             {'data': Binary("xxx")})[0][0]
   245         fdata = self.execute('Any D WHERE X data D, X eid %(x)s', {'x': feid})[0][0]
   245         fdata = self.execute('Any D WHERE X data D, X eid %(x)s', {'x': feid})[0][0]
   246         self.assertIsInstance(fdata, Binary)
   246         self.assertIsInstance(fdata, Binary)
   247         self.assertEquals(fdata.getvalue(), 'xxx')
   247         self.assertEqual(fdata.getvalue(), 'xxx')
   248 
   248 
   249     # selection queries tests #################################################
   249     # selection queries tests #################################################
   250 
   250 
   251     def test_select_1(self):
   251     def test_select_1(self):
   252         rset = self.execute('Any X ORDERBY X WHERE X is CWGroup')
   252         rset = self.execute('Any X ORDERBY X WHERE X is CWGroup')
   253         result, descr = rset.rows, rset.description
   253         result, descr = rset.rows, rset.description
   254         self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,)])
   254         self.assertEqual(tuplify(result), [(1,), (2,), (3,), (4,)])
   255         self.assertEquals(descr, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)])
   255         self.assertEqual(descr, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)])
   256 
   256 
   257     def test_select_2(self):
   257     def test_select_2(self):
   258         rset = self.execute('Any X ORDERBY N WHERE X is CWGroup, X name N')
   258         rset = self.execute('Any X ORDERBY N WHERE X is CWGroup, X name N')
   259         self.assertEquals(tuplify(rset.rows), [(1,), (2,), (3,), (4,)])
   259         self.assertEqual(tuplify(rset.rows), [(1,), (2,), (3,), (4,)])
   260         self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)])
   260         self.assertEqual(rset.description, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)])
   261         rset = self.execute('Any X ORDERBY N DESC WHERE X is CWGroup, X name N')
   261         rset = self.execute('Any X ORDERBY N DESC WHERE X is CWGroup, X name N')
   262         self.assertEquals(tuplify(rset.rows), [(4,), (3,), (2,), (1,)])
   262         self.assertEqual(tuplify(rset.rows), [(4,), (3,), (2,), (1,)])
   263 
   263 
   264     def test_select_3(self):
   264     def test_select_3(self):
   265         rset = self.execute('Any N GROUPBY N WHERE X is CWGroup, X name N')
   265         rset = self.execute('Any N GROUPBY N WHERE X is CWGroup, X name N')
   266         result, descr = rset.rows, rset.description
   266         result, descr = rset.rows, rset.description
   267         result.sort()
   267         result.sort()
   268         self.assertEquals(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)])
   268         self.assertEqual(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)])
   269         self.assertEquals(descr, [('String',), ('String',), ('String',), ('String',)])
   269         self.assertEqual(descr, [('String',), ('String',), ('String',), ('String',)])
   270 
   270 
   271     def test_select_is(self):
   271     def test_select_is(self):
   272         rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN')
   272         rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN')
   273         result, descr = rset.rows, rset.description
   273         result, descr = rset.rows, rset.description
   274         self.assertEquals(result[0][1], descr[0][0])
   274         self.assertEqual(result[0][1], descr[0][0])
   275 
   275 
   276     def test_select_is_aggr(self):
   276     def test_select_is_aggr(self):
   277         rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN')
   277         rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN')
   278         result, descr = rset.rows, rset.description
   278         result, descr = rset.rows, rset.description
   279         self.assertEquals(descr[0][0], 'String')
   279         self.assertEqual(descr[0][0], 'String')
   280         self.assertEquals(descr[0][1], 'Int')
   280         self.assertEqual(descr[0][1], 'Int')
   281         self.assertEquals(result[0][0], 'CWRelation') # XXX may change as schema evolve
   281         self.assertEqual(result[0][0], 'CWRelation') # XXX may change as schema evolve
   282 
   282 
   283     def test_select_groupby_orderby(self):
   283     def test_select_groupby_orderby(self):
   284         rset = self.execute('Any N GROUPBY N ORDERBY N WHERE X is CWGroup, X name N')
   284         rset = self.execute('Any N GROUPBY N ORDERBY N WHERE X is CWGroup, X name N')
   285         self.assertEquals(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)])
   285         self.assertEqual(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)])
   286         self.assertEquals(rset.description, [('String',), ('String',), ('String',), ('String',)])
   286         self.assertEqual(rset.description, [('String',), ('String',), ('String',), ('String',)])
   287 
   287 
   288     def test_select_complex_groupby(self):
   288     def test_select_complex_groupby(self):
   289         rset = self.execute('Any N GROUPBY N WHERE X name N')
   289         rset = self.execute('Any N GROUPBY N WHERE X name N')
   290         rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D')
   290         rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D')
   291 
   291 
   293         seid = self.execute('State X WHERE X name "deactivated"')[0][0]
   293         seid = self.execute('State X WHERE X name "deactivated"')[0][0]
   294         rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid)
   294         rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid)
   295 
   295 
   296     def test_select_complex_orderby(self):
   296     def test_select_complex_orderby(self):
   297         rset1 = self.execute('Any N ORDERBY N WHERE X name N')
   297         rset1 = self.execute('Any N ORDERBY N WHERE X name N')
   298         self.assertEquals(sorted(rset1.rows), rset1.rows)
   298         self.assertEqual(sorted(rset1.rows), rset1.rows)
   299         rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N')
   299         rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N')
   300         self.assertEquals(rset.rows[0][0], rset1.rows[1][0])
   300         self.assertEqual(rset.rows[0][0], rset1.rows[1][0])
   301         self.assertEquals(len(rset), 5)
   301         self.assertEqual(len(rset), 5)
   302 
   302 
   303     def test_select_5(self):
   303     def test_select_5(self):
   304         rset = self.execute('Any X, TMP ORDERBY TMP WHERE X name TMP, X is CWGroup')
   304         rset = self.execute('Any X, TMP ORDERBY TMP WHERE X name TMP, X is CWGroup')
   305         self.assertEquals(tuplify(rset.rows), [(1, 'guests',), (2, 'managers',), (3, 'owners',), (4, 'users',)])
   305         self.assertEqual(tuplify(rset.rows), [(1, 'guests',), (2, 'managers',), (3, 'owners',), (4, 'users',)])
   306         self.assertEquals(rset.description, [('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',)])
   306         self.assertEqual(rset.description, [('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',)])
   307 
   307 
   308     def test_select_6(self):
   308     def test_select_6(self):
   309         self.execute("INSERT Personne X: X nom 'bidule'")[0]
   309         self.execute("INSERT Personne X: X nom 'bidule'")[0]
   310         rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")')
   310         rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")')
   311         #self.assertEquals(rset.description, [('Personne',), ('Personne',)])
   311         #self.assertEqual(rset.description, [('Personne',), ('Personne',)])
   312         self.assert_(('Personne',) in rset.description)
   312         self.assert_(('Personne',) in rset.description)
   313         rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")')
   313         rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")')
   314         self.assert_(('Personne',) in rset.description)
   314         self.assert_(('Personne',) in rset.description)
   315 
   315 
   316     def test_select_not_attr(self):
   316     def test_select_not_attr(self):
   317         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   317         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   318         seid = self.execute("INSERT Societe X: X nom 'chouette'")[0][0]
   318         seid = self.execute("INSERT Societe X: X nom 'chouette'")[0][0]
   319         rset = self.execute('Personne X WHERE NOT X nom "bidule"')
   319         rset = self.execute('Personne X WHERE NOT X nom "bidule"')
   320         self.assertEquals(len(rset.rows), 0, rset.rows)
   320         self.assertEqual(len(rset.rows), 0, rset.rows)
   321         rset = self.execute('Personne X WHERE NOT X nom "bid"')
   321         rset = self.execute('Personne X WHERE NOT X nom "bid"')
   322         self.assertEquals(len(rset.rows), 1, rset.rows)
   322         self.assertEqual(len(rset.rows), 1, rset.rows)
   323         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   323         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   324         rset = self.execute('Personne X WHERE NOT X travaille S')
   324         rset = self.execute('Personne X WHERE NOT X travaille S')
   325         self.assertEquals(len(rset.rows), 0, rset.rows)
   325         self.assertEqual(len(rset.rows), 0, rset.rows)
   326 
   326 
   327     def test_select_is_in(self):
   327     def test_select_is_in(self):
   328         self.execute("INSERT Personne X: X nom 'bidule'")
   328         self.execute("INSERT Personne X: X nom 'bidule'")
   329         self.execute("INSERT Societe X: X nom 'chouette'")
   329         self.execute("INSERT Societe X: X nom 'chouette'")
   330         self.assertEquals(len(self.execute("Any X WHERE X is IN (Personne, Societe)")),
   330         self.assertEqual(len(self.execute("Any X WHERE X is IN (Personne, Societe)")),
   331                           2)
   331                           2)
   332 
   332 
   333     def test_select_not_rel(self):
   333     def test_select_not_rel(self):
   334         self.execute("INSERT Personne X: X nom 'bidule'")
   334         self.execute("INSERT Personne X: X nom 'bidule'")
   335         self.execute("INSERT Societe X: X nom 'chouette'")
   335         self.execute("INSERT Societe X: X nom 'chouette'")
   336         self.execute("INSERT Personne X: X nom 'autre'")
   336         self.execute("INSERT Personne X: X nom 'autre'")
   337         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   337         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   338         rset = self.execute('Personne X WHERE NOT X travaille S')
   338         rset = self.execute('Personne X WHERE NOT X travaille S')
   339         self.assertEquals(len(rset.rows), 1, rset.rows)
   339         self.assertEqual(len(rset.rows), 1, rset.rows)
   340         rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"')
   340         rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"')
   341         self.assertEquals(len(rset.rows), 1, rset.rows)
   341         self.assertEqual(len(rset.rows), 1, rset.rows)
   342 
   342 
   343     def test_select_nonregr_inlined(self):
   343     def test_select_nonregr_inlined(self):
   344         self.execute("INSERT Note X: X para 'bidule'")
   344         self.execute("INSERT Note X: X para 'bidule'")
   345         self.execute("INSERT Personne X: X nom 'chouette'")
   345         self.execute("INSERT Personne X: X nom 'chouette'")
   346         self.execute("INSERT Personne X: X nom 'autre'")
   346         self.execute("INSERT Personne X: X nom 'autre'")
   347         self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'")
   347         self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'")
   348         rset = self.execute('Any U,T ORDERBY T DESC WHERE U is CWUser, '
   348         rset = self.execute('Any U,T ORDERBY T DESC WHERE U is CWUser, '
   349                             'N ecrit_par U, N type T')#, {'x': self.ueid})
   349                             'N ecrit_par U, N type T')#, {'x': self.ueid})
   350         self.assertEquals(len(rset.rows), 0)
   350         self.assertEqual(len(rset.rows), 0)
   351 
   351 
   352     def test_select_nonregr_edition_not(self):
   352     def test_select_nonregr_edition_not(self):
   353         groupeids = set((1, 2, 3))
   353         groupeids = set((1, 2, 3))
   354         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'))
   354         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'))
   355         rset = self.execute('DISTINCT Any Y WHERE X is CWEType, X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y')
   355         rset = self.execute('DISTINCT Any Y WHERE X is CWEType, X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y')
   356         self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))
   356         self.assertEqual(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))
   357         rset = self.execute('DISTINCT Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y')
   357         rset = self.execute('DISTINCT Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y')
   358         self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))
   358         self.assertEqual(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))
   359 
   359 
   360     def test_select_outer_join(self):
   360     def test_select_outer_join(self):
   361         peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   361         peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   362         peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0]
   362         peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0]
   363         seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0]
   363         seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0]
   364         seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0]
   364         seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0]
   365         rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?')
   365         rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?')
   366         self.assertEquals(rset.rows, [[peid1, None], [peid2, None]])
   366         self.assertEqual(rset.rows, [[peid1, None], [peid2, None]])
   367         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   367         self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")
   368         rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?')
   368         rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?')
   369         self.assertEquals(rset.rows, [[peid1, seid1], [peid2, None]])
   369         self.assertEqual(rset.rows, [[peid1, seid1], [peid2, None]])
   370         rset = self.execute('Any S,X ORDERBY S WHERE X? travaille S')
   370         rset = self.execute('Any S,X ORDERBY S WHERE X? travaille S')
   371         self.assertEquals(rset.rows, [[seid1, peid1], [seid2, None]])
   371         self.assertEqual(rset.rows, [[seid1, peid1], [seid2, None]])
   372 
   372 
   373     def test_select_outer_join_optimized(self):
   373     def test_select_outer_join_optimized(self):
   374         peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   374         peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   375         rset = self.execute('Any X WHERE X eid %(x)s, P? connait X', {'x':peid1})
   375         rset = self.execute('Any X WHERE X eid %(x)s, P? connait X', {'x':peid1})
   376         self.assertEquals(rset.rows, [[peid1]])
   376         self.assertEqual(rset.rows, [[peid1]])
   377         rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?',
   377         rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?',
   378                             {'x':peid1})
   378                             {'x':peid1})
   379         self.assertEquals(rset.rows, [[peid1]])
   379         self.assertEqual(rset.rows, [[peid1]])
   380 
   380 
   381     def test_select_left_outer_join(self):
   381     def test_select_left_outer_join(self):
   382         rset = self.execute('DISTINCT Any G WHERE U? in_group G')
   382         rset = self.execute('DISTINCT Any G WHERE U? in_group G')
   383         self.assertEquals(len(rset), 4)
   383         self.assertEqual(len(rset), 4)
   384         rset = self.execute('DISTINCT Any G WHERE U? in_group G, U eid %(x)s',
   384         rset = self.execute('DISTINCT Any G WHERE U? in_group G, U eid %(x)s',
   385                             {'x': self.session.user.eid})
   385                             {'x': self.session.user.eid})
   386         self.assertEquals(len(rset), 4)
   386         self.assertEqual(len(rset), 4)
   387 
   387 
   388     def test_select_ambigous_outer_join(self):
   388     def test_select_ambigous_outer_join(self):
   389         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
   389         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
   390         self.execute("INSERT Tag X: X name 'tagbis'")[0][0]
   390         self.execute("INSERT Tag X: X name 'tagbis'")[0][0]
   391         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
   391         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
   393                      {'g': geid, 't': teid})
   393                      {'g': geid, 't': teid})
   394         rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN")
   394         rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN")
   395         self.failUnless(['users', 'tag'] in rset.rows)
   395         self.failUnless(['users', 'tag'] in rset.rows)
   396         self.failUnless(['activated', None] in rset.rows)
   396         self.failUnless(['activated', None] in rset.rows)
   397         rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN")
   397         rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN")
   398         self.assertEquals(rset.rows, [[None, 'tagbis'], ['users', 'tag']])
   398         self.assertEqual(rset.rows, [[None, 'tagbis'], ['users', 'tag']])
   399 
   399 
   400     def test_select_not_inline_rel(self):
   400     def test_select_not_inline_rel(self):
   401         self.execute("INSERT Personne X: X nom 'bidule'")
   401         self.execute("INSERT Personne X: X nom 'bidule'")
   402         self.execute("INSERT Note X: X type 'a'")
   402         self.execute("INSERT Note X: X type 'a'")
   403         self.execute("INSERT Note X: X type 'b'")
   403         self.execute("INSERT Note X: X type 'b'")
   404         self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'")
   404         self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'")
   405         rset = self.execute('Note X WHERE NOT X ecrit_par P')
   405         rset = self.execute('Note X WHERE NOT X ecrit_par P')
   406         self.assertEquals(len(rset.rows), 1, rset.rows)
   406         self.assertEqual(len(rset.rows), 1, rset.rows)
   407 
   407 
   408     def test_select_not_unlinked_multiple_solutions(self):
   408     def test_select_not_unlinked_multiple_solutions(self):
   409         self.execute("INSERT Personne X: X nom 'bidule'")
   409         self.execute("INSERT Personne X: X nom 'bidule'")
   410         self.execute("INSERT Note X: X type 'a'")
   410         self.execute("INSERT Note X: X type 'a'")
   411         self.execute("INSERT Note X: X type 'b'")
   411         self.execute("INSERT Note X: X type 'b'")
   412         self.execute("SET Y evaluee X WHERE X type 'a', Y nom 'bidule'")
   412         self.execute("SET Y evaluee X WHERE X type 'a', Y nom 'bidule'")
   413         rset = self.execute('Note X WHERE NOT Y evaluee X')
   413         rset = self.execute('Note X WHERE NOT Y evaluee X')
   414         self.assertEquals(len(rset.rows), 1, rset.rows)
   414         self.assertEqual(len(rset.rows), 1, rset.rows)
   415 
   415 
   416     def test_select_date_extraction(self):
   416     def test_select_date_extraction(self):
   417         self.execute("INSERT Personne X: X nom 'foo', X datenaiss %(d)s",
   417         self.execute("INSERT Personne X: X nom 'foo', X datenaiss %(d)s",
   418                      {'d': datetime(2001, 2,3, 12,13)})
   418                      {'d': datetime(2001, 2,3, 12,13)})
   419         test_data = [('YEAR', 2001), ('MONTH', 2), ('DAY', 3),
   419         test_data = [('YEAR', 2001), ('MONTH', 2), ('DAY', 3),
   420                      ('HOUR', 12), ('MINUTE', 13)]
   420                      ('HOUR', 12), ('MINUTE', 13)]
   421         for funcname, result in test_data:
   421         for funcname, result in test_data:
   422             rset = self.execute('Any %s(D) WHERE X is Personne, X datenaiss D'
   422             rset = self.execute('Any %s(D) WHERE X is Personne, X datenaiss D'
   423                                 % funcname)
   423                                 % funcname)
   424             self.assertEquals(len(rset.rows), 1)
   424             self.assertEqual(len(rset.rows), 1)
   425             self.assertEquals(rset.rows[0][0], result)
   425             self.assertEqual(rset.rows[0][0], result)
   426             self.assertEquals(rset.description, [('Int',)])
   426             self.assertEqual(rset.description, [('Int',)])
   427 
   427 
   428     def test_select_aggregat_count(self):
   428     def test_select_aggregat_count(self):
   429         rset = self.execute('Any COUNT(X)')
   429         rset = self.execute('Any COUNT(X)')
   430         self.assertEquals(len(rset.rows), 1)
   430         self.assertEqual(len(rset.rows), 1)
   431         self.assertEquals(len(rset.rows[0]), 1)
   431         self.assertEqual(len(rset.rows[0]), 1)
   432         self.assertEquals(rset.description, [('Int',)])
   432         self.assertEqual(rset.description, [('Int',)])
   433 
   433 
   434     def test_select_aggregat_sum(self):
   434     def test_select_aggregat_sum(self):
   435         rset = self.execute('Any SUM(O) WHERE X ordernum O')
   435         rset = self.execute('Any SUM(O) WHERE X ordernum O')
   436         self.assertEquals(len(rset.rows), 1)
   436         self.assertEqual(len(rset.rows), 1)
   437         self.assertEquals(len(rset.rows[0]), 1)
   437         self.assertEqual(len(rset.rows[0]), 1)
   438         self.assertEquals(rset.description, [('Int',)])
   438         self.assertEqual(rset.description, [('Int',)])
   439 
   439 
   440     def test_select_aggregat_min(self):
   440     def test_select_aggregat_min(self):
   441         rset = self.execute('Any MIN(X) WHERE X is Personne')
   441         rset = self.execute('Any MIN(X) WHERE X is Personne')
   442         self.assertEquals(len(rset.rows), 1)
   442         self.assertEqual(len(rset.rows), 1)
   443         self.assertEquals(len(rset.rows[0]), 1)
   443         self.assertEqual(len(rset.rows[0]), 1)
   444         self.assertEquals(rset.description, [('Personne',)])
   444         self.assertEqual(rset.description, [('Personne',)])
   445         rset = self.execute('Any MIN(O) WHERE X ordernum O')
   445         rset = self.execute('Any MIN(O) WHERE X ordernum O')
   446         self.assertEquals(len(rset.rows), 1)
   446         self.assertEqual(len(rset.rows), 1)
   447         self.assertEquals(len(rset.rows[0]), 1)
   447         self.assertEqual(len(rset.rows[0]), 1)
   448         self.assertEquals(rset.description, [('Int',)])
   448         self.assertEqual(rset.description, [('Int',)])
   449 
   449 
   450     def test_select_aggregat_max(self):
   450     def test_select_aggregat_max(self):
   451         rset = self.execute('Any MAX(X) WHERE X is Personne')
   451         rset = self.execute('Any MAX(X) WHERE X is Personne')
   452         self.assertEquals(len(rset.rows), 1)
   452         self.assertEqual(len(rset.rows), 1)
   453         self.assertEquals(len(rset.rows[0]), 1)
   453         self.assertEqual(len(rset.rows[0]), 1)
   454         self.assertEquals(rset.description, [('Personne',)])
   454         self.assertEqual(rset.description, [('Personne',)])
   455         rset = self.execute('Any MAX(O) WHERE X ordernum O')
   455         rset = self.execute('Any MAX(O) WHERE X ordernum O')
   456         self.assertEquals(len(rset.rows), 1)
   456         self.assertEqual(len(rset.rows), 1)
   457         self.assertEquals(len(rset.rows[0]), 1)
   457         self.assertEqual(len(rset.rows[0]), 1)
   458         self.assertEquals(rset.description, [('Int',)])
   458         self.assertEqual(rset.description, [('Int',)])
   459 
   459 
   460     def test_select_custom_aggregat_concat_string(self):
   460     def test_select_custom_aggregat_concat_string(self):
   461         rset = self.execute('Any GROUP_CONCAT(N) WHERE X is CWGroup, X name N')
   461         rset = self.execute('Any GROUP_CONCAT(N) WHERE X is CWGroup, X name N')
   462         self.failUnless(rset)
   462         self.failUnless(rset)
   463         self.failUnlessEqual(sorted(rset[0][0].split(', ')), ['guests', 'managers',
   463         self.failUnlessEqual(sorted(rset[0][0].split(', ')), ['guests', 'managers',
   480         self.failUnlessEqual(len(rset), 3)
   480         self.failUnlessEqual(len(rset), 3)
   481         self.failUnlessEqual(rset[0][1], 'owners')
   481         self.failUnlessEqual(rset[0][1], 'owners')
   482 
   482 
   483     def test_select_aggregat_sort(self):
   483     def test_select_aggregat_sort(self):
   484         rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G')
   484         rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G')
   485         self.assertEquals(len(rset.rows), 2)
   485         self.assertEqual(len(rset.rows), 2)
   486         self.assertEquals(len(rset.rows[0]), 2)
   486         self.assertEqual(len(rset.rows[0]), 2)
   487         self.assertEquals(rset.description[0], ('CWGroup', 'Int',))
   487         self.assertEqual(rset.description[0], ('CWGroup', 'Int',))
   488 
   488 
   489     def test_select_aggregat_having(self):
   489     def test_select_aggregat_having(self):
   490         rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N '
   490         rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N '
   491                             'WHERE RT name N, RDEF relation_type RT '
   491                             'WHERE RT name N, RDEF relation_type RT '
   492                             'HAVING COUNT(RDEF) > 10')
   492                             'HAVING COUNT(RDEF) > 10')
   493         self.assertListEquals(rset.rows,
   493         self.assertListEqual(rset.rows,
   494                               [[u'description_format', 12],
   494                               [[u'description_format', 12],
   495                                [u'description', 13],
   495                                [u'description', 13],
   496                                [u'name', 14],
   496                                [u'name', 14],
   497                                [u'created_by', 38],
   497                                [u'created_by', 38],
   498                                [u'creation_date', 38],
   498                                [u'creation_date', 38],
   506     def test_select_aggregat_having_dumb(self):
   506     def test_select_aggregat_having_dumb(self):
   507         # dumb but should not raise an error
   507         # dumb but should not raise an error
   508         rset = self.execute('Any U,COUNT(X) GROUPBY U '
   508         rset = self.execute('Any U,COUNT(X) GROUPBY U '
   509                             'WHERE U eid %(x)s, X owned_by U '
   509                             'WHERE U eid %(x)s, X owned_by U '
   510                             'HAVING COUNT(X) > 10', {'x': self.ueid})
   510                             'HAVING COUNT(X) > 10', {'x': self.ueid})
   511         self.assertEquals(len(rset.rows), 1)
   511         self.assertEqual(len(rset.rows), 1)
   512         self.assertEquals(rset.rows[0][0], self.ueid)
   512         self.assertEqual(rset.rows[0][0], self.ueid)
   513 
   513 
   514     def test_select_having_non_aggregat_1(self):
   514     def test_select_having_non_aggregat_1(self):
   515         rset = self.execute('Any L WHERE X login L, X creation_date CD '
   515         rset = self.execute('Any L WHERE X login L, X creation_date CD '
   516                             'HAVING YEAR(CD) = %s' % date.today().year)
   516                             'HAVING YEAR(CD) = %s' % date.today().year)
   517         self.assertListEquals(rset.rows,
   517         self.assertListEqual(rset.rows,
   518                               [[u'admin'],
   518                               [[u'admin'],
   519                                [u'anon']])
   519                                [u'anon']])
   520 
   520 
   521     def test_select_having_non_aggregat_2(self):
   521     def test_select_having_non_aggregat_2(self):
   522         rset = self.execute('Any L GROUPBY L WHERE X login L, X in_group G, '
   522         rset = self.execute('Any L GROUPBY L WHERE X login L, X in_group G, '
   523                             'X creation_date CD HAVING YEAR(CD) = %s OR COUNT(G) > 1'
   523                             'X creation_date CD HAVING YEAR(CD) = %s OR COUNT(G) > 1'
   524                             % date.today().year)
   524                             % date.today().year)
   525         self.assertListEquals(rset.rows,
   525         self.assertListEqual(rset.rows,
   526                               [[u'admin'],
   526                               [[u'admin'],
   527                                [u'anon']])
   527                                [u'anon']])
   528 
   528 
   529     def test_select_complex_sort(self):
   529     def test_select_complex_sort(self):
   530         """need sqlite including http://www.sqlite.org/cvstrac/tktview?tn=3773 fix"""
   530         """need sqlite including http://www.sqlite.org/cvstrac/tktview?tn=3773 fix"""
   531         rset = self.execute('Any X ORDERBY X,D LIMIT 5 WHERE X creation_date D')
   531         rset = self.execute('Any X ORDERBY X,D LIMIT 5 WHERE X creation_date D')
   532         result = rset.rows
   532         result = rset.rows
   533         result.sort()
   533         result.sort()
   534         self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,), (5,)])
   534         self.assertEqual(tuplify(result), [(1,), (2,), (3,), (4,), (5,)])
   535 
   535 
   536     def test_select_upper(self):
   536     def test_select_upper(self):
   537         rset = self.execute('Any X, UPPER(L) ORDERBY L WHERE X is CWUser, X login L')
   537         rset = self.execute('Any X, UPPER(L) ORDERBY L WHERE X is CWUser, X login L')
   538         self.assertEquals(len(rset.rows), 2)
   538         self.assertEqual(len(rset.rows), 2)
   539         self.assertEquals(rset.rows[0][1], 'ADMIN')
   539         self.assertEqual(rset.rows[0][1], 'ADMIN')
   540         self.assertEquals(rset.description[0], ('CWUser', 'String',))
   540         self.assertEqual(rset.description[0], ('CWUser', 'String',))
   541         self.assertEquals(rset.rows[1][1], 'ANON')
   541         self.assertEqual(rset.rows[1][1], 'ANON')
   542         self.assertEquals(rset.description[1], ('CWUser', 'String',))
   542         self.assertEqual(rset.description[1], ('CWUser', 'String',))
   543         eid = rset.rows[0][0]
   543         eid = rset.rows[0][0]
   544         rset = self.execute('Any UPPER(L) WHERE X eid %s, X login L'%eid)
   544         rset = self.execute('Any UPPER(L) WHERE X eid %s, X login L'%eid)
   545         self.assertEquals(rset.rows[0][0], 'ADMIN')
   545         self.assertEqual(rset.rows[0][0], 'ADMIN')
   546         self.assertEquals(rset.description, [('String',)])
   546         self.assertEqual(rset.description, [('String',)])
   547 
   547 
   548     def test_select_float_abs(self):
   548     def test_select_float_abs(self):
   549         # test positive number
   549         # test positive number
   550         eid = self.execute('INSERT Affaire A: A invoiced %(i)s', {'i': 1.2})[0][0]
   550         eid = self.execute('INSERT Affaire A: A invoiced %(i)s', {'i': 1.2})[0][0]
   551         rset = self.execute('Any ABS(I) WHERE X eid %(x)s, X invoiced I', {'x': eid})
   551         rset = self.execute('Any ABS(I) WHERE X eid %(x)s, X invoiced I', {'x': eid})
   552         self.assertEquals(rset.rows[0][0], 1.2)
   552         self.assertEqual(rset.rows[0][0], 1.2)
   553         # test negative number
   553         # test negative number
   554         eid = self.execute('INSERT Affaire A: A invoiced %(i)s', {'i': -1.2})[0][0]
   554         eid = self.execute('INSERT Affaire A: A invoiced %(i)s', {'i': -1.2})[0][0]
   555         rset = self.execute('Any ABS(I) WHERE X eid %(x)s, X invoiced I', {'x': eid})
   555         rset = self.execute('Any ABS(I) WHERE X eid %(x)s, X invoiced I', {'x': eid})
   556         self.assertEquals(rset.rows[0][0], 1.2)
   556         self.assertEqual(rset.rows[0][0], 1.2)
   557 
   557 
   558     def test_select_int_abs(self):
   558     def test_select_int_abs(self):
   559         # test positive number
   559         # test positive number
   560         eid = self.execute('INSERT Affaire A: A duration %(d)s', {'d': 12})[0][0]
   560         eid = self.execute('INSERT Affaire A: A duration %(d)s', {'d': 12})[0][0]
   561         rset = self.execute('Any ABS(D) WHERE X eid %(x)s, X duration D', {'x': eid})
   561         rset = self.execute('Any ABS(D) WHERE X eid %(x)s, X duration D', {'x': eid})
   562         self.assertEquals(rset.rows[0][0], 12)
   562         self.assertEqual(rset.rows[0][0], 12)
   563         # test negative number
   563         # test negative number
   564         eid = self.execute('INSERT Affaire A: A duration %(d)s', {'d': -12})[0][0]
   564         eid = self.execute('INSERT Affaire A: A duration %(d)s', {'d': -12})[0][0]
   565         rset = self.execute('Any ABS(D) WHERE X eid %(x)s, X duration D', {'x': eid})
   565         rset = self.execute('Any ABS(D) WHERE X eid %(x)s, X duration D', {'x': eid})
   566         self.assertEquals(rset.rows[0][0], 12)
   566         self.assertEqual(rset.rows[0][0], 12)
       
   567 
       
   568 ##     def test_select_simplified(self):
       
   569 ##         ueid = self.session.user.eid
       
   570 ##         rset = self.execute('Any L WHERE %s login L'%ueid)
       
   571 ##         self.assertEqual(rset.rows[0][0], 'admin')
       
   572 ##         rset = self.execute('Any L WHERE %(x)s login L', {'x':ueid})
       
   573 ##         self.assertEqual(rset.rows[0][0], 'admin')
   567 
   574 
   568     def test_select_searchable_text_1(self):
   575     def test_select_searchable_text_1(self):
   569         rset = self.execute(u"INSERT Personne X: X nom 'bidüle'")
   576         rset = self.execute(u"INSERT Personne X: X nom 'bidüle'")
   570         rset = self.execute(u"INSERT Societe X: X nom 'bidüle'")
   577         rset = self.execute(u"INSERT Societe X: X nom 'bidüle'")
   571         rset = self.execute("INSERT Societe X: X nom 'chouette'")
   578         rset = self.execute("INSERT Societe X: X nom 'chouette'")
   572         self.commit()
   579         self.commit()
   573         rset = self.execute('Any X where X has_text %(text)s', {'text': u'bidüle'})
   580         rset = self.execute('Any X where X has_text %(text)s', {'text': u'bidüle'})
   574         self.assertEquals(len(rset.rows), 2, rset.rows)
   581         self.assertEqual(len(rset.rows), 2, rset.rows)
   575         rset = self.execute(u'Any N where N has_text "bidüle"')
   582         rset = self.execute(u'Any N where N has_text "bidüle"')
   576         self.assertEquals(len(rset.rows), 2, rset.rows)
   583         self.assertEqual(len(rset.rows), 2, rset.rows)
   577         biduleeids = [r[0] for r in rset.rows]
   584         biduleeids = [r[0] for r in rset.rows]
   578         rset = self.execute(u'Any N where NOT N has_text "bidüle"')
   585         rset = self.execute(u'Any N where NOT N has_text "bidüle"')
   579         self.failIf([r[0] for r in rset.rows if r[0] in biduleeids])
   586         self.failIf([r[0] for r in rset.rows if r[0] in biduleeids])
   580         # duh?
   587         # duh?
   581         rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'})
   588         rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'})
   584         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   591         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   585         rset = self.execute("INSERT Personne X: X nom 'chouette'")
   592         rset = self.execute("INSERT Personne X: X nom 'chouette'")
   586         rset = self.execute("INSERT Societe X: X nom 'bidule'")
   593         rset = self.execute("INSERT Societe X: X nom 'bidule'")
   587         self.commit()
   594         self.commit()
   588         rset = self.execute('Personne N where N has_text "bidule"')
   595         rset = self.execute('Personne N where N has_text "bidule"')
   589         self.assertEquals(len(rset.rows), 1, rset.rows)
   596         self.assertEqual(len(rset.rows), 1, rset.rows)
   590 
   597 
   591     def test_select_searchable_text_3(self):
   598     def test_select_searchable_text_3(self):
   592         rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'M'")
   599         rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'M'")
   593         rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'F'")
   600         rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'F'")
   594         rset = self.execute("INSERT Societe X: X nom 'bidule'")
   601         rset = self.execute("INSERT Societe X: X nom 'bidule'")
   595         self.commit()
   602         self.commit()
   596         rset = self.execute('Any X where X has_text "bidule" and X sexe "M"')
   603         rset = self.execute('Any X where X has_text "bidule" and X sexe "M"')
   597         self.assertEquals(len(rset.rows), 1, rset.rows)
   604         self.assertEqual(len(rset.rows), 1, rset.rows)
   598 
   605 
   599     def test_select_multiple_searchable_text(self):
   606     def test_select_multiple_searchable_text(self):
   600         self.execute(u"INSERT Personne X: X nom 'bidüle'")
   607         self.execute(u"INSERT Personne X: X nom 'bidüle'")
   601         self.execute("INSERT Societe X: X nom 'chouette', S travaille X")
   608         self.execute("INSERT Societe X: X nom 'chouette', S travaille X")
   602         self.execute(u"INSERT Personne X: X nom 'bidüle'")
   609         self.execute(u"INSERT Personne X: X nom 'bidüle'")
   603         self.commit()
   610         self.commit()
   604         rset = self.execute('Personne X WHERE X has_text %(text)s, X travaille S, S has_text %(text2)s',
   611         rset = self.execute('Personne X WHERE X has_text %(text)s, X travaille S, S has_text %(text2)s',
   605                             {'text': u'bidüle',
   612                             {'text': u'bidüle',
   606                              'text2': u'chouette',}
   613                              'text2': u'chouette',}
   607                             )
   614                             )
   608         self.assertEquals(len(rset.rows), 1, rset.rows)
   615         self.assertEqual(len(rset.rows), 1, rset.rows)
   609 
   616 
   610     def test_select_no_descr(self):
   617     def test_select_no_descr(self):
   611         rset = self.execute('Any X WHERE X is CWGroup', build_descr=0)
   618         rset = self.execute('Any X WHERE X is CWGroup', build_descr=0)
   612         rset.rows.sort()
   619         rset.rows.sort()
   613         self.assertEquals(tuplify(rset.rows), [(1,), (2,), (3,), (4,)])
   620         self.assertEqual(tuplify(rset.rows), [(1,), (2,), (3,), (4,)])
   614         self.assertEquals(rset.description, ())
   621         self.assertEqual(rset.description, ())
   615 
   622 
   616     def test_select_limit_offset(self):
   623     def test_select_limit_offset(self):
   617         rset = self.execute('CWGroup X ORDERBY N LIMIT 2 WHERE X name N')
   624         rset = self.execute('CWGroup X ORDERBY N LIMIT 2 WHERE X name N')
   618         self.assertEquals(tuplify(rset.rows), [(1,), (2,)])
   625         self.assertEqual(tuplify(rset.rows), [(1,), (2,)])
   619         self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',)])
   626         self.assertEqual(rset.description, [('CWGroup',), ('CWGroup',)])
   620         rset = self.execute('CWGroup X ORDERBY N LIMIT 2 OFFSET 2 WHERE X name N')
   627         rset = self.execute('CWGroup X ORDERBY N LIMIT 2 OFFSET 2 WHERE X name N')
   621         self.assertEquals(tuplify(rset.rows), [(3,), (4,)])
   628         self.assertEqual(tuplify(rset.rows), [(3,), (4,)])
   622 
   629 
   623     def test_select_symmetric(self):
   630     def test_select_symmetric(self):
   624         self.execute("INSERT Personne X: X nom 'machin'")
   631         self.execute("INSERT Personne X: X nom 'machin'")
   625         self.execute("INSERT Personne X: X nom 'bidule'")
   632         self.execute("INSERT Personne X: X nom 'bidule'")
   626         self.execute("INSERT Personne X: X nom 'chouette'")
   633         self.execute("INSERT Personne X: X nom 'chouette'")
   627         self.execute("INSERT Personne X: X nom 'trucmuche'")
   634         self.execute("INSERT Personne X: X nom 'trucmuche'")
   628         self.execute("SET X connait Y WHERE X nom 'chouette', Y nom 'bidule'")
   635         self.execute("SET X connait Y WHERE X nom 'chouette', Y nom 'bidule'")
   629         self.execute("SET X connait Y WHERE X nom 'machin', Y nom 'chouette'")
   636         self.execute("SET X connait Y WHERE X nom 'machin', Y nom 'chouette'")
   630         rset = self.execute('Any P where P connait P2')
   637         rset = self.execute('Any P where P connait P2')
   631         self.assertEquals(len(rset.rows), 3, rset.rows)
   638         self.assertEqual(len(rset.rows), 3, rset.rows)
   632         rset = self.execute('Any P where NOT P connait P2')
   639         rset = self.execute('Any P where NOT P connait P2')
   633         self.assertEquals(len(rset.rows), 1, rset.rows) # trucmuche
   640         self.assertEqual(len(rset.rows), 1, rset.rows) # trucmuche
   634         rset = self.execute('Any P where P connait P2, P2 nom "bidule"')
   641         rset = self.execute('Any P where P connait P2, P2 nom "bidule"')
   635         self.assertEquals(len(rset.rows), 1, rset.rows)
   642         self.assertEqual(len(rset.rows), 1, rset.rows)
   636         rset = self.execute('Any P where P2 connait P, P2 nom "bidule"')
   643         rset = self.execute('Any P where P2 connait P, P2 nom "bidule"')
   637         self.assertEquals(len(rset.rows), 1, rset.rows)
   644         self.assertEqual(len(rset.rows), 1, rset.rows)
   638         rset = self.execute('Any P where P connait P2, P2 nom "chouette"')
   645         rset = self.execute('Any P where P connait P2, P2 nom "chouette"')
   639         self.assertEquals(len(rset.rows), 2, rset.rows)
   646         self.assertEqual(len(rset.rows), 2, rset.rows)
   640         rset = self.execute('Any P where P2 connait P, P2 nom "chouette"')
   647         rset = self.execute('Any P where P2 connait P, P2 nom "chouette"')
   641         self.assertEquals(len(rset.rows), 2, rset.rows)
   648         self.assertEqual(len(rset.rows), 2, rset.rows)
   642 
   649 
   643     def test_select_inline(self):
   650     def test_select_inline(self):
   644         self.execute("INSERT Personne X: X nom 'bidule'")
   651         self.execute("INSERT Personne X: X nom 'bidule'")
   645         self.execute("INSERT Note X: X type 'a'")
   652         self.execute("INSERT Note X: X type 'a'")
   646         self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'")
   653         self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'")
   647         rset = self.execute('Any N where N ecrit_par X, X nom "bidule"')
   654         rset = self.execute('Any N where N ecrit_par X, X nom "bidule"')
   648         self.assertEquals(len(rset.rows), 1, rset.rows)
   655         self.assertEqual(len(rset.rows), 1, rset.rows)
   649 
   656 
   650     def test_select_creation_date(self):
   657     def test_select_creation_date(self):
   651         self.execute("INSERT Personne X: X nom 'bidule'")
   658         self.execute("INSERT Personne X: X nom 'bidule'")
   652         rset = self.execute('Any D WHERE X nom "bidule", X creation_date D')
   659         rset = self.execute('Any D WHERE X nom "bidule", X creation_date D')
   653         self.assertEqual(len(rset.rows), 1)
   660         self.assertEqual(len(rset.rows), 1)
   713         self.assertEqual(len(rset.rows), 2)
   720         self.assertEqual(len(rset.rows), 2)
   714 
   721 
   715     def test_select_explicit_eid(self):
   722     def test_select_explicit_eid(self):
   716         rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid})
   723         rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid})
   717         self.failUnless(rset)
   724         self.failUnless(rset)
   718         self.assertEquals(rset.description[0][1], 'Int')
   725         self.assertEqual(rset.description[0][1], 'Int')
   719 
   726 
   720 #     def test_select_rewritten_optional(self):
   727 #     def test_select_rewritten_optional(self):
   721 #         eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
   728 #         eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
   722 #         rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)',
   729 #         rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)',
   723 #                             {'x': eid}, 'x')
   730 #                             {'x': eid}, 'x')
   724 #         self.assertEquals(rset.rows, [[eid]])
   731 #         self.assertEqual(rset.rows, [[eid]])
   725 
   732 
   726     def test_today_bug(self):
   733     def test_today_bug(self):
   727         self.execute("INSERT Tag X: X name 'bidule', X creation_date NOW")
   734         self.execute("INSERT Tag X: X name 'bidule', X creation_date NOW")
   728         self.execute("INSERT Tag Y: Y name 'toto'")
   735         self.execute("INSERT Tag Y: Y name 'toto'")
   729         rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D")
   736         rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D")
   740         self.assertEqual(len(rset.rows), 2)
   747         self.assertEqual(len(rset.rows), 2)
   741 
   748 
   742     def test_select_boolean(self):
   749     def test_select_boolean(self):
   743         rset = self.execute('Any N WHERE X is CWEType, X name N, X final %(val)s',
   750         rset = self.execute('Any N WHERE X is CWEType, X name N, X final %(val)s',
   744                             {'val': True})
   751                             {'val': True})
   745         self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes',
   752         self.assertEqual(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes',
   746                                                             'Date', 'Datetime',
   753                                                             'Date', 'Datetime',
   747                                                             'Decimal', 'Float',
   754                                                             'Decimal', 'Float',
   748                                                             'Int', 'Interval',
   755                                                             'Int', 'Interval',
   749                                                             'Password', 'String',
   756                                                             'Password', 'String',
   750                                                             'Time'])
   757                                                             'Time'])
   751         rset = self.execute('Any N WHERE X is CWEType, X name N, X final TRUE')
   758         rset = self.execute('Any N WHERE X is CWEType, X name N, X final TRUE')
   752         self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes',
   759         self.assertEqual(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes',
   753                                                             'Date', 'Datetime',
   760                                                             'Date', 'Datetime',
   754                                                             'Decimal', 'Float',
   761                                                             'Decimal', 'Float',
   755                                                             'Int', 'Interval',
   762                                                             'Int', 'Interval',
   756                                                             'Password', 'String',
   763                                                             'Password', 'String',
   757                                                             'Time'])
   764                                                             'Time'])
   758 
   765 
   759     def test_select_constant(self):
   766     def test_select_constant(self):
   760         rset = self.execute('Any X, "toto" ORDERBY X WHERE X is CWGroup')
   767         rset = self.execute('Any X, "toto" ORDERBY X WHERE X is CWGroup')
   761         self.assertEquals(rset.rows,
   768         self.assertEqual(rset.rows,
   762                           map(list, zip((1,2,3,4), ('toto','toto','toto','toto',))))
   769                           map(list, zip((1,2,3,4), ('toto','toto','toto','toto',))))
   763         self.assertIsInstance(rset[0][1], unicode)
   770         self.assertIsInstance(rset[0][1], unicode)
   764         self.assertEquals(rset.description,
   771         self.assertEqual(rset.description,
   765                           zip(('CWGroup', 'CWGroup', 'CWGroup', 'CWGroup'),
   772                           zip(('CWGroup', 'CWGroup', 'CWGroup', 'CWGroup'),
   766                               ('String', 'String', 'String', 'String',)))
   773                               ('String', 'String', 'String', 'String',)))
   767         rset = self.execute('Any X, %(value)s ORDERBY X WHERE X is CWGroup', {'value': 'toto'})
   774         rset = self.execute('Any X, %(value)s ORDERBY X WHERE X is CWGroup', {'value': 'toto'})
   768         self.assertEquals(rset.rows,
   775         self.assertEqual(rset.rows,
   769                           map(list, zip((1,2,3,4), ('toto','toto','toto','toto',))))
   776                           map(list, zip((1,2,3,4), ('toto','toto','toto','toto',))))
   770         self.assertIsInstance(rset[0][1], unicode)
   777         self.assertIsInstance(rset[0][1], unicode)
   771         self.assertEquals(rset.description,
   778         self.assertEqual(rset.description,
   772                           zip(('CWGroup', 'CWGroup', 'CWGroup', 'CWGroup'),
   779                           zip(('CWGroup', 'CWGroup', 'CWGroup', 'CWGroup'),
   773                               ('String', 'String', 'String', 'String',)))
   780                               ('String', 'String', 'String', 'String',)))
   774         rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN')
   781         rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN')
   775 
   782 
   776     def test_select_union(self):
   783     def test_select_union(self):
   777         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   784         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   778                             '((Any X,N WHERE X name N, X transition_of WF, WF workflow_of E, E name %(name)s)'
   785                             '((Any X,N WHERE X name N, X transition_of WF, WF workflow_of E, E name %(name)s)'
   779                             ' UNION '
   786                             ' UNION '
   780                             '(Any X,N WHERE X name N, X state_of WF, WF workflow_of E, E name %(name)s))',
   787                             '(Any X,N WHERE X name N, X state_of WF, WF workflow_of E, E name %(name)s))',
   781                             {'name': 'CWUser'})
   788                             {'name': 'CWUser'})
   782         self.assertEquals([x[1] for x in rset.rows],
   789         self.assertEqual([x[1] for x in rset.rows],
   783                           ['activate', 'activated', 'deactivate', 'deactivated'])
   790                           ['activate', 'activated', 'deactivate', 'deactivated'])
   784         self.assertEquals(rset.description,
   791         self.assertEqual(rset.description,
   785                           [('Transition', 'String'), ('State', 'String'),
   792                           [('Transition', 'String'), ('State', 'String'),
   786                            ('Transition', 'String'), ('State', 'String')])
   793                            ('Transition', 'String'), ('State', 'String')])
   787 
   794 
   788     def test_select_union_aggregat(self):
   795     def test_select_union_aggregat(self):
   789         # meaningless, the goal in to have group by done on different attribute
   796         # meaningless, the goal in to have group by done on different attribute
   799         self.execute('INSERT Transition X: X name "hop"')
   806         self.execute('INSERT Transition X: X name "hop"')
   800         rset = self.execute('Any N,NX ORDERBY 2 WITH N,NX BEING '
   807         rset = self.execute('Any N,NX ORDERBY 2 WITH N,NX BEING '
   801                             '((Any N,COUNT(X) GROUPBY N WHERE X name N, X is State HAVING COUNT(X)>1)'
   808                             '((Any N,COUNT(X) GROUPBY N WHERE X name N, X is State HAVING COUNT(X)>1)'
   802                             ' UNION '
   809                             ' UNION '
   803                             '(Any N,COUNT(X) GROUPBY N WHERE X name N, X is Transition HAVING COUNT(X)>1))')
   810                             '(Any N,COUNT(X) GROUPBY N WHERE X name N, X is Transition HAVING COUNT(X)>1))')
   804         self.assertEquals(rset.rows, [[u'hop', 2], [u'hop', 2]])
   811         self.assertEqual(rset.rows, [[u'hop', 2], [u'hop', 2]])
   805 
   812 
   806     def test_select_union_selection_with_diff_variables(self):
   813     def test_select_union_selection_with_diff_variables(self):
   807         rset = self.execute('(Any N WHERE X name N, X is State)'
   814         rset = self.execute('(Any N WHERE X name N, X is State)'
   808                             ' UNION '
   815                             ' UNION '
   809                             '(Any NN WHERE XX name NN, XX is Transition)')
   816                             '(Any NN WHERE XX name NN, XX is Transition)')
   810         self.assertEquals(sorted(r[0] for r in rset.rows),
   817         self.assertEqual(sorted(r[0] for r in rset.rows),
   811                           ['abort', 'activate', 'activated', 'ben non',
   818                           ['abort', 'activate', 'activated', 'ben non',
   812                            'deactivate', 'deactivated', 'done', 'en cours',
   819                            'deactivate', 'deactivated', 'done', 'en cours',
   813                            'end', 'finie', 'markasdone', 'pitetre', 'redoit',
   820                            'end', 'finie', 'markasdone', 'pitetre', 'redoit',
   814                            'start', 'todo'])
   821                            'start', 'todo'])
   815 
   822 
   818         eid2 = self.execute('CWUser X WHERE X login "admin"')[0][0]
   825         eid2 = self.execute('CWUser X WHERE X login "admin"')[0][0]
   819         rset = self.execute('(Any X WHERE X eid %(x)s)'
   826         rset = self.execute('(Any X WHERE X eid %(x)s)'
   820                             ' UNION '
   827                             ' UNION '
   821                             '(Any Y WHERE Y eid %(y)s)',
   828                             '(Any Y WHERE Y eid %(y)s)',
   822                             {'x': eid1, 'y': eid2})
   829                             {'x': eid1, 'y': eid2})
   823         self.assertEquals(rset.description[:], [('CWGroup',), ('CWUser',)])
   830         self.assertEqual(rset.description[:], [('CWGroup',), ('CWUser',)])
   824 
   831 
   825     def test_exists(self):
   832     def test_exists(self):
   826         geid = self.execute("INSERT CWGroup X: X name 'lulufanclub'")[0][0]
   833         geid = self.execute("INSERT CWGroup X: X name 'lulufanclub'")[0][0]
   827         self.execute("SET U in_group G WHERE G name 'lulufanclub'")
   834         self.execute("SET U in_group G WHERE G name 'lulufanclub'")
   828         peid = self.execute("INSERT Personne X: X prenom 'lulu', X nom 'petit'")[0][0]
   835         peid = self.execute("INSERT Personne X: X prenom 'lulu', X nom 'petit'")[0][0]
   829         rset = self.execute("Any X WHERE X prenom 'lulu',"
   836         rset = self.execute("Any X WHERE X prenom 'lulu',"
   830                             "EXISTS (U in_group G, G name 'lulufanclub' OR G name 'managers');")
   837                             "EXISTS (U in_group G, G name 'lulufanclub' OR G name 'managers');")
   831         self.assertEquals(rset.rows, [[peid]])
   838         self.assertEqual(rset.rows, [[peid]])
   832 
   839 
   833     def test_identity(self):
   840     def test_identity(self):
   834         eid = self.execute('Any X WHERE X identity Y, Y eid 1')[0][0]
   841         eid = self.execute('Any X WHERE X identity Y, Y eid 1')[0][0]
   835         self.assertEquals(eid, 1)
   842         self.assertEqual(eid, 1)
   836         eid = self.execute('Any X WHERE Y identity X, Y eid 1')[0][0]
   843         eid = self.execute('Any X WHERE Y identity X, Y eid 1')[0][0]
   837         self.assertEquals(eid, 1)
   844         self.assertEqual(eid, 1)
   838         login = self.execute('Any L WHERE X login "admin", X identity Y, Y login L')[0][0]
   845         login = self.execute('Any L WHERE X login "admin", X identity Y, Y login L')[0][0]
   839         self.assertEquals(login, 'admin')
   846         self.assertEqual(login, 'admin')
   840 
   847 
   841     def test_select_date_mathexp(self):
   848     def test_select_date_mathexp(self):
   842         rset = self.execute('Any X, TODAY - CD WHERE X is CWUser, X creation_date CD')
   849         rset = self.execute('Any X, TODAY - CD WHERE X is CWUser, X creation_date CD')
   843         self.failUnless(rset)
   850         self.failUnless(rset)
   844         self.failUnlessEqual(rset.description[0][1], 'Interval')
   851         self.failUnlessEqual(rset.description[0][1], 'Interval')
   850         # percent users by groups
   857         # percent users by groups
   851         self.execute('SET X in_group G WHERE G name "users"')
   858         self.execute('SET X in_group G WHERE G name "users"')
   852         rset = self.execute('Any GN, COUNT(X)*100/T GROUPBY GN ORDERBY 2,1'
   859         rset = self.execute('Any GN, COUNT(X)*100/T GROUPBY GN ORDERBY 2,1'
   853                             ' WHERE G name GN, X in_group G'
   860                             ' WHERE G name GN, X in_group G'
   854                             ' WITH T BEING (Any COUNT(U) WHERE U is CWUser)')
   861                             ' WITH T BEING (Any COUNT(U) WHERE U is CWUser)')
   855         self.assertEquals(rset.rows, [[u'guests', 50], [u'managers', 50], [u'users', 100]])
   862         self.assertEqual(rset.rows, [[u'guests', 50], [u'managers', 50], [u'users', 100]])
   856         self.assertEquals(rset.description, [('String', 'Int'), ('String', 'Int'), ('String', 'Int')])
   863         self.assertEqual(rset.description, [('String', 'Int'), ('String', 'Int'), ('String', 'Int')])
   857 
   864 
   858     def test_select_subquery_aggregat_2(self):
   865     def test_select_subquery_aggregat_2(self):
   859         expected = self.execute('Any X, 0, COUNT(T) GROUPBY X '
   866         expected = self.execute('Any X, 0, COUNT(T) GROUPBY X '
   860                                 'WHERE X is Workflow, T transition_of X').rows
   867                                 'WHERE X is Workflow, T transition_of X').rows
   861         rset = self.execute('''
   868         rset = self.execute('''
   862 Any P1,B,E WHERE P1 identity P2 WITH
   869 Any P1,B,E WHERE P1 identity P2 WITH
   863   P1,B BEING (Any P,COUNT(T) GROUPBY P WHERE P is Workflow, T is Transition,
   870   P1,B BEING (Any P,COUNT(T) GROUPBY P WHERE P is Workflow, T is Transition,
   864               T? transition_of P, T type "auto"),
   871               T? transition_of P, T type "auto"),
   865   P2,E BEING (Any P,COUNT(T) GROUPBY P WHERE P is Workflow, T is Transition,
   872   P2,E BEING (Any P,COUNT(T) GROUPBY P WHERE P is Workflow, T is Transition,
   866               T? transition_of P, T type "normal")''')
   873               T? transition_of P, T type "normal")''')
   867         self.assertEquals(sorted(rset.rows), sorted(expected))
   874         self.assertEqual(sorted(rset.rows), sorted(expected))
   868 
   875 
   869     def test_select_subquery_const(self):
   876     def test_select_subquery_const(self):
   870         rset = self.execute('Any X WITH X BEING ((Any NULL) UNION (Any "toto"))')
   877         rset = self.execute('Any X WITH X BEING ((Any NULL) UNION (Any "toto"))')
   871         self.assertEquals(rset.rows, [[None], ['toto']])
   878         self.assertEqual(rset.rows, [[None], ['toto']])
   872         self.assertEquals(rset.description, [(None,), ('String',)])
   879         self.assertEqual(rset.description, [(None,), ('String',)])
   873 
   880 
   874     # insertion queries tests #################################################
   881     # insertion queries tests #################################################
   875 
   882 
   876     def test_insert_is(self):
   883     def test_insert_is(self):
   877         eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0]
   884         eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0]
   878         etype, = self.execute("Any TN WHERE X is T, X eid %s, T name TN" % eid)[0]
   885         etype, = self.execute("Any TN WHERE X is T, X eid %s, T name TN" % eid)[0]
   879         self.assertEquals(etype, 'Personne')
   886         self.assertEqual(etype, 'Personne')
   880         self.execute("INSERT Personne X: X nom 'managers'")
   887         self.execute("INSERT Personne X: X nom 'managers'")
   881 
   888 
   882     def test_insert_1(self):
   889     def test_insert_1(self):
   883         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   890         rset = self.execute("INSERT Personne X: X nom 'bidule'")
   884         self.assertEquals(len(rset.rows), 1)
   891         self.assertEqual(len(rset.rows), 1)
   885         self.assertEquals(rset.description, [('Personne',)])
   892         self.assertEqual(rset.description, [('Personne',)])
   886         rset = self.execute('Personne X WHERE X nom "bidule"')
   893         rset = self.execute('Personne X WHERE X nom "bidule"')
   887         self.assert_(rset.rows)
   894         self.assert_(rset.rows)
   888         self.assertEquals(rset.description, [('Personne',)])
   895         self.assertEqual(rset.description, [('Personne',)])
   889 
   896 
   890     def test_insert_1_multiple(self):
   897     def test_insert_1_multiple(self):
   891         self.execute("INSERT Personne X: X nom 'bidule'")
   898         self.execute("INSERT Personne X: X nom 'bidule'")
   892         self.execute("INSERT Personne X: X nom 'chouette'")
   899         self.execute("INSERT Personne X: X nom 'chouette'")
   893         rset = self.execute("INSERT Societe Y: Y nom N, P travaille Y WHERE P nom N")
   900         rset = self.execute("INSERT Societe Y: Y nom N, P travaille Y WHERE P nom N")
   894         self.assertEquals(len(rset.rows), 2)
   901         self.assertEqual(len(rset.rows), 2)
   895         self.assertEquals(rset.description, [('Societe',), ('Societe',)])
   902         self.assertEqual(rset.description, [('Societe',), ('Societe',)])
   896 
   903 
   897     def test_insert_2(self):
   904     def test_insert_2(self):
   898         rset = self.execute("INSERT Personne X, Personne Y: X nom 'bidule', Y nom 'tutu'")
   905         rset = self.execute("INSERT Personne X, Personne Y: X nom 'bidule', Y nom 'tutu'")
   899         self.assertEquals(rset.description, [('Personne', 'Personne')])
   906         self.assertEqual(rset.description, [('Personne', 'Personne')])
   900         rset = self.execute('Personne X WHERE X nom "bidule" or X nom "tutu"')
   907         rset = self.execute('Personne X WHERE X nom "bidule" or X nom "tutu"')
   901         self.assert_(rset.rows)
   908         self.assert_(rset.rows)
   902         self.assertEquals(rset.description, [('Personne',), ('Personne',)])
   909         self.assertEqual(rset.description, [('Personne',), ('Personne',)])
   903 
   910 
   904     def test_insert_3(self):
   911     def test_insert_3(self):
   905         self.execute("INSERT Personne X: X nom Y WHERE U login 'admin', U login Y")
   912         self.execute("INSERT Personne X: X nom Y WHERE U login 'admin', U login Y")
   906         rset = self.execute('Personne X WHERE X nom "admin"')
   913         rset = self.execute('Personne X WHERE X nom "admin"')
   907         self.assert_(rset.rows)
   914         self.assert_(rset.rows)
   908         self.assertEquals(rset.description, [('Personne',)])
   915         self.assertEqual(rset.description, [('Personne',)])
   909 
   916 
   910     def test_insert_4(self):
   917     def test_insert_4(self):
   911         self.execute("INSERT Societe Y: Y nom 'toto'")
   918         self.execute("INSERT Societe Y: Y nom 'toto'")
   912         self.execute("INSERT Personne X: X nom 'bidule', X travaille Y WHERE Y nom 'toto'")
   919         self.execute("INSERT Personne X: X nom 'bidule', X travaille Y WHERE Y nom 'toto'")
   913         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   920         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   914         self.assert_(rset.rows)
   921         self.assert_(rset.rows)
   915         self.assertEquals(rset.description, [('Personne', 'Societe',)])
   922         self.assertEqual(rset.description, [('Personne', 'Societe',)])
   916 
   923 
   917     def test_insert_4bis(self):
   924     def test_insert_4bis(self):
   918         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   925         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   919         seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s",
   926         seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s",
   920                              {'x': str(peid)})[0][0]
   927                              {'x': str(peid)})[0][0]
   935     def test_insert_5(self):
   942     def test_insert_5(self):
   936         self.execute("INSERT Personne X: X nom 'bidule'")
   943         self.execute("INSERT Personne X: X nom 'bidule'")
   937         self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X nom 'bidule'")
   944         self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X nom 'bidule'")
   938         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   945         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   939         self.assert_(rset.rows)
   946         self.assert_(rset.rows)
   940         self.assertEquals(rset.description, [('Personne', 'Societe',)])
   947         self.assertEqual(rset.description, [('Personne', 'Societe',)])
   941 
   948 
   942     def test_insert_5bis(self):
   949     def test_insert_5bis(self):
   943         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   950         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
   944         self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s",
   951         self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s",
   945                      {'x': peid})
   952                      {'x': peid})
   946         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   953         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   947         self.assert_(rset.rows)
   954         self.assert_(rset.rows)
   948         self.assertEquals(rset.description, [('Personne', 'Societe',)])
   955         self.assertEqual(rset.description, [('Personne', 'Societe',)])
   949 
   956 
   950     def test_insert_6(self):
   957     def test_insert_6(self):
   951         self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto', X travaille Y")
   958         self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto', X travaille Y")
   952         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   959         rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y')
   953         self.assert_(rset.rows)
   960         self.assert_(rset.rows)
   954         self.assertEquals(rset.description, [('Personne', 'Societe',)])
   961         self.assertEqual(rset.description, [('Personne', 'Societe',)])
   955 
   962 
   956     def test_insert_7(self):
   963     def test_insert_7(self):
   957         self.execute("INSERT Personne X, Societe Y: X nom N, Y nom 'toto', X travaille Y WHERE U login 'admin', U login N")
   964         self.execute("INSERT Personne X, Societe Y: X nom N, Y nom 'toto', X travaille Y WHERE U login 'admin', U login N")
   958         rset = self.execute('Any X, Y WHERE X nom "admin", Y nom "toto", X travaille Y')
   965         rset = self.execute('Any X, Y WHERE X nom "admin", Y nom "toto", X travaille Y')
   959         self.assert_(rset.rows)
   966         self.assert_(rset.rows)
   960         self.assertEquals(rset.description, [('Personne', 'Societe',)])
   967         self.assertEqual(rset.description, [('Personne', 'Societe',)])
   961 
   968 
   962     def test_insert_7_2(self):
   969     def test_insert_7_2(self):
   963         self.execute("INSERT Personne X, Societe Y: X nom N, Y nom 'toto', X travaille Y WHERE U login N")
   970         self.execute("INSERT Personne X, Societe Y: X nom N, Y nom 'toto', X travaille Y WHERE U login N")
   964         rset = self.execute('Any X, Y WHERE Y nom "toto", X travaille Y')
   971         rset = self.execute('Any X, Y WHERE Y nom "toto", X travaille Y')
   965         self.assertEquals(len(rset), 2)
   972         self.assertEqual(len(rset), 2)
   966         self.assertEquals(rset.description, [('Personne', 'Societe',),
   973         self.assertEqual(rset.description, [('Personne', 'Societe',),
   967                                              ('Personne', 'Societe',)])
   974                                              ('Personne', 'Societe',)])
   968 
   975 
   969     def test_insert_8(self):
   976     def test_insert_8(self):
   970         self.execute("INSERT Societe Y, Personne X: Y nom N, X nom 'toto', X travaille Y WHERE U login 'admin', U login N")
   977         self.execute("INSERT Societe Y, Personne X: Y nom N, X nom 'toto', X travaille Y WHERE U login 'admin', U login N")
   971         rset = self.execute('Any X, Y WHERE X nom "toto", Y nom "admin", X travaille Y')
   978         rset = self.execute('Any X, Y WHERE X nom "toto", Y nom "admin", X travaille Y')
   972         self.assert_(rset.rows)
   979         self.assert_(rset.rows)
   973         self.assertEquals(rset.description, [('Personne', 'Societe',)])
   980         self.assertEqual(rset.description, [('Personne', 'Societe',)])
   974 
   981 
   975     def test_insert_9(self):
   982     def test_insert_9(self):
   976         self.execute("INSERT Societe X: X nom  'Lo'")
   983         self.execute("INSERT Societe X: X nom  'Lo'")
   977         self.execute("INSERT Societe X: X nom  'Gi'")
   984         self.execute("INSERT Societe X: X nom  'Gi'")
   978         self.execute("INSERT SubDivision X: X nom  'Lab'")
   985         self.execute("INSERT SubDivision X: X nom  'Lab'")
   979         rset = self.execute("INSERT Personne X: X nom N, X travaille Y, X travaille_subdivision Z WHERE Y is Societe, Z is SubDivision, Y nom N")
   986         rset = self.execute("INSERT Personne X: X nom N, X travaille Y, X travaille_subdivision Z WHERE Y is Societe, Z is SubDivision, Y nom N")
   980         self.assertEquals(len(rset), 2)
   987         self.assertEqual(len(rset), 2)
   981         self.assertEquals(rset.description, [('Personne',), ('Personne',)])
   988         self.assertEqual(rset.description, [('Personne',), ('Personne',)])
   982         # self.assertSetEquals(set(x.nom for x in rset.entities()),
   989         # self.assertSetEqual(set(x.nom for x in rset.entities()),
   983         #                      ['Lo', 'Gi'])
   990         #                      ['Lo', 'Gi'])
   984         # self.assertSetEquals(set(y.nom for x in rset.entities() for y in x.travaille),
   991         # self.assertSetEqual(set(y.nom for x in rset.entities() for y in x.travaille),
   985         #                      ['Lo', 'Gi'])
   992         #                      ['Lo', 'Gi'])
   986         # self.assertEquals([y.nom for x in rset.entities() for y in x.travaille_subdivision],
   993         # self.assertEqual([y.nom for x in rset.entities() for y in x.travaille_subdivision],
   987         #                      ['Lab', 'Lab'])
   994         #                      ['Lab', 'Lab'])
   988 
   995 
   989     def test_insert_query_error(self):
   996     def test_insert_query_error(self):
   990         self.assertRaises(Exception,
   997         self.assertRaises(Exception,
   991                           self.execute,
   998                           self.execute,
  1003 
  1010 
  1004     def test_insertion_description_with_where(self):
  1011     def test_insertion_description_with_where(self):
  1005         rset = self.execute('INSERT CWUser E, EmailAddress EM: E login "X", E upassword "X", '
  1012         rset = self.execute('INSERT CWUser E, EmailAddress EM: E login "X", E upassword "X", '
  1006                             'E primary_email EM, EM address "X", E in_group G '
  1013                             'E primary_email EM, EM address "X", E in_group G '
  1007                             'WHERE G name "managers"')
  1014                             'WHERE G name "managers"')
  1008         self.assertEquals(list(rset.description[0]), ['CWUser', 'EmailAddress'])
  1015         self.assertEqual(list(rset.description[0]), ['CWUser', 'EmailAddress'])
  1009 
  1016 
  1010     # deletion queries tests ##################################################
  1017     # deletion queries tests ##################################################
  1011 
  1018 
  1012     def test_delete_1(self):
  1019     def test_delete_1(self):
  1013         self.execute("INSERT Personne Y: Y nom 'toto'")
  1020         self.execute("INSERT Personne Y: Y nom 'toto'")
  1018         rset = self.execute('Personne X WHERE X nom "toto"')
  1025         rset = self.execute('Personne X WHERE X nom "toto"')
  1019         self.assertEqual(len(rset.rows), 0)
  1026         self.assertEqual(len(rset.rows), 0)
  1020 
  1027 
  1021     def test_delete_2(self):
  1028     def test_delete_2(self):
  1022         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")
  1029         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")
  1023         self.assertEquals(len(rset), 1)
  1030         self.assertEqual(len(rset), 1)
  1024         self.assertEquals(len(rset[0]), 3)
  1031         self.assertEqual(len(rset[0]), 3)
  1025         self.assertEquals(rset.description[0], ('Personne', 'Personne', 'Societe'))
  1032         self.assertEqual(rset.description[0], ('Personne', 'Personne', 'Societe'))
  1026         self.assertEquals(self.execute('Any N WHERE X nom N, X eid %s'% rset[0][0])[0][0], 'syt')
  1033         self.assertEqual(self.execute('Any N WHERE X nom N, X eid %s'% rset[0][0])[0][0], 'syt')
  1027         rset = self.execute('Personne X WHERE X travaille Y, Y nom "Logilab"')
  1034         rset = self.execute('Personne X WHERE X travaille Y, Y nom "Logilab"')
  1028         self.assertEqual(len(rset.rows), 2, rset.rows)
  1035         self.assertEqual(len(rset.rows), 2, rset.rows)
  1029         self.execute("DELETE X travaille Y WHERE X is Personne, Y nom 'Logilabo'")
  1036         self.execute("DELETE X travaille Y WHERE X is Personne, Y nom 'Logilabo'")
  1030         rset = self.execute('Personne X WHERE X travaille Y, Y nom "Logilab"')
  1037         rset = self.execute('Personne X WHERE X travaille Y, Y nom "Logilab"')
  1031         self.assertEqual(len(rset.rows), 2, rset.rows)
  1038         self.assertEqual(len(rset.rows), 2, rset.rows)
  1047     def test_delete_symmetric(self):
  1054     def test_delete_symmetric(self):
  1048         teid1 = self.execute("INSERT Folder T: T name 'toto'")[0][0]
  1055         teid1 = self.execute("INSERT Folder T: T name 'toto'")[0][0]
  1049         teid2 = self.execute("INSERT Folder T: T name 'tutu'")[0][0]
  1056         teid2 = self.execute("INSERT Folder T: T name 'tutu'")[0][0]
  1050         self.execute('SET X see_also Y WHERE X eid %s, Y eid %s' % (teid1, teid2))
  1057         self.execute('SET X see_also Y WHERE X eid %s, Y eid %s' % (teid1, teid2))
  1051         rset = self.execute('Any X,Y WHERE X see_also Y')
  1058         rset = self.execute('Any X,Y WHERE X see_also Y')
  1052         self.assertEquals(len(rset) , 2, rset.rows)
  1059         self.assertEqual(len(rset) , 2, rset.rows)
  1053         self.execute('DELETE X see_also Y WHERE X eid %s, Y eid %s' % (teid1, teid2))
  1060         self.execute('DELETE X see_also Y WHERE X eid %s, Y eid %s' % (teid1, teid2))
  1054         rset = self.execute('Any X,Y WHERE X see_also Y')
  1061         rset = self.execute('Any X,Y WHERE X see_also Y')
  1055         self.assertEquals(len(rset) , 0)
  1062         self.assertEqual(len(rset) , 0)
  1056         self.execute('SET X see_also Y WHERE X eid %s, Y eid %s' % (teid1, teid2))
  1063         self.execute('SET X see_also Y WHERE X eid %s, Y eid %s' % (teid1, teid2))
  1057         rset = self.execute('Any X,Y WHERE X see_also Y')
  1064         rset = self.execute('Any X,Y WHERE X see_also Y')
  1058         self.assertEquals(len(rset) , 2)
  1065         self.assertEqual(len(rset) , 2)
  1059         self.execute('DELETE X see_also Y WHERE X eid %s, Y eid %s' % (teid2, teid1))
  1066         self.execute('DELETE X see_also Y WHERE X eid %s, Y eid %s' % (teid2, teid1))
  1060         rset = self.execute('Any X,Y WHERE X see_also Y')
  1067         rset = self.execute('Any X,Y WHERE X see_also Y')
  1061         self.assertEquals(len(rset) , 0)
  1068         self.assertEqual(len(rset) , 0)
  1062 
  1069 
  1063     def test_nonregr_delete_cache(self):
  1070     def test_nonregr_delete_cache(self):
  1064         """test that relations are properly cleaned when an entity is deleted
  1071         """test that relations are properly cleaned when an entity is deleted
  1065         (using cachekey on sql generation returned always the same query for an eid,
  1072         (using cachekey on sql generation returned always the same query for an eid,
  1066         whatever the relation)
  1073         whatever the relation)
  1071         #'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y'
  1078         #'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y'
  1072         eeid, = self.o.execute(s, 'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y WHERE Y is EmailAddress')[0]
  1079         eeid, = self.o.execute(s, 'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y WHERE Y is EmailAddress')[0]
  1073         self.o.execute(s, "DELETE Email X")
  1080         self.o.execute(s, "DELETE Email X")
  1074         sqlc = s.pool['system']
  1081         sqlc = s.pool['system']
  1075         sqlc.execute('SELECT * FROM recipients_relation')
  1082         sqlc.execute('SELECT * FROM recipients_relation')
  1076         self.assertEquals(len(sqlc.fetchall()), 0)
  1083         self.assertEqual(len(sqlc.fetchall()), 0)
  1077         sqlc.execute('SELECT * FROM owned_by_relation WHERE eid_from=%s'%eeid)
  1084         sqlc.execute('SELECT * FROM owned_by_relation WHERE eid_from=%s'%eeid)
  1078         self.assertEquals(len(sqlc.fetchall()), 0)
  1085         self.assertEqual(len(sqlc.fetchall()), 0)
  1079 
  1086 
  1080     def test_nonregr_delete_cache2(self):
  1087     def test_nonregr_delete_cache2(self):
  1081         eid = self.execute("INSERT Folder T: T name 'toto'")[0][0]
  1088         eid = self.execute("INSERT Folder T: T name 'toto'")[0][0]
  1082         self.commit()
  1089         self.commit()
  1083         # fill the cache
  1090         # fill the cache
  1086         self.execute("Folder X WHERE X eid %(x)s", {'x': eid})
  1093         self.execute("Folder X WHERE X eid %(x)s", {'x': eid})
  1087         self.execute("Folder X WHERE X eid %s" % eid)
  1094         self.execute("Folder X WHERE X eid %s" % eid)
  1088         self.execute("DELETE Folder T WHERE T eid %s" % eid)
  1095         self.execute("DELETE Folder T WHERE T eid %s" % eid)
  1089         self.commit()
  1096         self.commit()
  1090         rset = self.execute("Any X WHERE X eid %(x)s", {'x': eid})
  1097         rset = self.execute("Any X WHERE X eid %(x)s", {'x': eid})
  1091         self.assertEquals(rset.rows, [])
  1098         self.assertEqual(rset.rows, [])
  1092         rset = self.execute("Any X WHERE X eid %s" % eid)
  1099         rset = self.execute("Any X WHERE X eid %s" % eid)
  1093         self.assertEquals(rset.rows, [])
  1100         self.assertEqual(rset.rows, [])
  1094         rset = self.execute("Folder X WHERE X eid %(x)s", {'x': eid})
  1101         rset = self.execute("Folder X WHERE X eid %(x)s", {'x': eid})
  1095         self.assertEquals(rset.rows, [])
  1102         self.assertEqual(rset.rows, [])
  1096         rset = self.execute("Folder X WHERE X eid %s" %eid)
  1103         rset = self.execute("Folder X WHERE X eid %s" %eid)
  1097         self.assertEquals(rset.rows, [])
  1104         self.assertEqual(rset.rows, [])
  1098 
  1105 
  1099     # update queries tests ####################################################
  1106     # update queries tests ####################################################
  1100 
  1107 
  1101     def test_update_1(self):
  1108     def test_update_1(self):
  1102         peid = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0]
  1109         peid = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0]
  1108         self.assertEqual(tuplify(rset.rows), [('tutu', 'original')])
  1115         self.assertEqual(tuplify(rset.rows), [('tutu', 'original')])
  1109 
  1116 
  1110     def test_update_2(self):
  1117     def test_update_2(self):
  1111         peid, seid = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")[0]
  1118         peid, seid = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")[0]
  1112         rset = self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'")
  1119         rset = self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'")
  1113         self.assertEquals(tuplify(rset.rows), [(peid, seid)])
  1120         self.assertEqual(tuplify(rset.rows), [(peid, seid)])
  1114         rset = self.execute('Any X, Y WHERE X travaille Y')
  1121         rset = self.execute('Any X, Y WHERE X travaille Y')
  1115         self.assertEqual(len(rset.rows), 1)
  1122         self.assertEqual(len(rset.rows), 1)
  1116 
  1123 
  1117     def test_update_2bis(self):
  1124     def test_update_2bis(self):
  1118         rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
  1125         rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")
  1136 
  1143 
  1137     def test_update_multiple1(self):
  1144     def test_update_multiple1(self):
  1138         peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1145         peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1139         peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0]
  1146         peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0]
  1140         self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'")
  1147         self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'")
  1141         self.assertEquals(self.execute('Any X WHERE X nom "toto"').rows, [[peid1]])
  1148         self.assertEqual(self.execute('Any X WHERE X nom "toto"').rows, [[peid1]])
  1142         self.assertEquals(self.execute('Any X WHERE X nom "tutu"').rows, [[peid2]])
  1149         self.assertEqual(self.execute('Any X WHERE X nom "tutu"').rows, [[peid2]])
  1143 
  1150 
  1144     def test_update_multiple2(self):
  1151     def test_update_multiple2(self):
  1145         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0]
  1152         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0]
  1146         peid1 = self.execute("INSERT Personne Y: Y nom 'turlu'")[0][0]
  1153         peid1 = self.execute("INSERT Personne Y: Y nom 'turlu'")[0][0]
  1147         peid2 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1154         peid2 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0]
  1160                 break
  1167                 break
  1161         self.execute('SET X ordernum Y+1 WHERE X from_entity SE, SE name "Personne", X ordernum Y, X ordernum >= %(order)s',
  1168         self.execute('SET X ordernum Y+1 WHERE X from_entity SE, SE name "Personne", X ordernum Y, X ordernum >= %(order)s',
  1162                      {'order': orders[splitidx]})
  1169                      {'order': orders[splitidx]})
  1163         orders2 = [r[0] for r in self.execute('Any O ORDERBY O WHERE ST name "Personne", X from_entity ST, X ordernum O')]
  1170         orders2 = [r[0] for r in self.execute('Any O ORDERBY O WHERE ST name "Personne", X from_entity ST, X ordernum O')]
  1164         orders = orders[:splitidx] + [o+1 for o in orders[splitidx:]]
  1171         orders = orders[:splitidx] + [o+1 for o in orders[splitidx:]]
  1165         self.assertEquals(orders2, orders)
  1172         self.assertEqual(orders2, orders)
  1166 
  1173 
  1167     def test_update_string_concat(self):
  1174     def test_update_string_concat(self):
  1168         beid = self.execute("INSERT Bookmark Y: Y title 'toto', Y path '/view'")[0][0]
  1175         beid = self.execute("INSERT Bookmark Y: Y title 'toto', Y path '/view'")[0][0]
  1169         self.execute('SET X title XN + %(suffix)s WHERE X is Bookmark, X title XN', {'suffix': u'-moved'})
  1176         self.execute('SET X title XN + %(suffix)s WHERE X is Bookmark, X title XN', {'suffix': u'-moved'})
  1170         newname = self.execute('Any XN WHERE X eid %(x)s, X title XN', {'x': beid})[0][0]
  1177         newname = self.execute('Any XN WHERE X eid %(x)s, X title XN', {'x': beid})[0][0]
  1171         self.assertEquals(newname, 'toto-moved')
  1178         self.assertEqual(newname, 'toto-moved')
  1172 
  1179 
  1173     def test_update_query_error(self):
  1180     def test_update_query_error(self):
  1174         self.execute("INSERT Personne Y: Y nom 'toto'")
  1181         self.execute("INSERT Personne Y: Y nom 'toto'")
  1175         self.assertRaises(Exception, self.execute, "SET X nom 'toto', X is Personne")
  1182         self.assertRaises(Exception, self.execute, "SET X nom 'toto', X is Personne")
  1176         self.assertRaises(QueryError, self.execute, "SET X nom 'toto', X has_text 'tutu' WHERE X is Personne")
  1183         self.assertRaises(QueryError, self.execute, "SET X nom 'toto', X has_text 'tutu' WHERE X is Personne")
  1177         self.assertRaises(QueryError, self.execute, "SET X login 'tutu', X eid %s" % cnx.user(self.session).eid)
  1184         self.assertRaises(QueryError, self.execute, "SET X login 'tutu', X eid %s" % cnx.user(self.session).eid)
  1178 
  1185 
  1179 
       
  1180     # upassword encryption tests #################################################
  1186     # upassword encryption tests #################################################
  1181 
  1187 
  1182     def test_insert_upassword(self):
  1188     def test_insert_upassword(self):
  1183         rset = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")
  1189         rset = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")
  1184         self.assertEquals(len(rset.rows), 1)
  1190         self.assertEqual(len(rset.rows), 1)
  1185         self.assertEquals(rset.description, [('CWUser',)])
  1191         self.assertEqual(rset.description, [('CWUser',)])
  1186         self.assertRaises(Unauthorized,
  1192         self.assertRaises(Unauthorized,
  1187                           self.execute, "Any P WHERE X is CWUser, X login 'bob', X upassword P")
  1193                           self.execute, "Any P WHERE X is CWUser, X login 'bob', X upassword P")
  1188         cursor = self.pool['system']
  1194         cursor = self.pool['system']
  1189         cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'"
  1195         cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'"
  1190                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1196                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1191         passwd = str(cursor.fetchone()[0])
  1197         passwd = str(cursor.fetchone()[0])
  1192         self.assertEquals(passwd, crypt_password('toto', passwd[:2]))
  1198         self.assertEqual(passwd, crypt_password('toto', passwd[:2]))
  1193         rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s",
  1199         rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s",
  1194                             {'pwd': Binary(passwd)})
  1200                             {'pwd': Binary(passwd)})
  1195         self.assertEquals(len(rset.rows), 1)
  1201         self.assertEqual(len(rset.rows), 1)
  1196         self.assertEquals(rset.description, [('CWUser',)])
  1202         self.assertEqual(rset.description, [('CWUser',)])
  1197 
  1203 
  1198     def test_update_upassword(self):
  1204     def test_update_upassword(self):
  1199         cursor = self.pool['system']
  1205         cursor = self.pool['system']
  1200         rset = self.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'})
  1206         rset = self.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'})
  1201         self.assertEquals(rset.description[0][0], 'CWUser')
  1207         self.assertEqual(rset.description[0][0], 'CWUser')
  1202         rset = self.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'",
  1208         rset = self.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'",
  1203                             {'pwd': 'tutu'})
  1209                             {'pwd': 'tutu'})
  1204         cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'"
  1210         cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'"
  1205                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1211                        % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX))
  1206         passwd = str(cursor.fetchone()[0])
  1212         passwd = str(cursor.fetchone()[0])
  1207         self.assertEquals(passwd, crypt_password('tutu', passwd[:2]))
  1213         self.assertEqual(passwd, crypt_password('tutu', passwd[:2]))
  1208         rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s",
  1214         rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s",
  1209                             {'pwd': Binary(passwd)})
  1215                             {'pwd': Binary(passwd)})
  1210         self.assertEquals(len(rset.rows), 1)
  1216         self.assertEqual(len(rset.rows), 1)
  1211         self.assertEquals(rset.description, [('CWUser',)])
  1217         self.assertEqual(rset.description, [('CWUser',)])
  1212 
  1218 
  1213     # non regression tests ####################################################
  1219     # non regression tests ####################################################
  1214 
  1220 
  1215     def test_nonregr_1(self):
  1221     def test_nonregr_1(self):
  1216         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
  1222         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
  1217         self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'")
  1223         self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'")
  1218         rset = self.execute('Any X WHERE T tags X')
  1224         rset = self.execute('Any X WHERE T tags X')
  1219         self.assertEquals(len(rset.rows), 1, rset.rows)
  1225         self.assertEqual(len(rset.rows), 1, rset.rows)
  1220         rset = self.execute('Any T WHERE T tags X, X is State')
  1226         rset = self.execute('Any T WHERE T tags X, X is State')
  1221         self.assertEquals(rset.rows, [[teid]])
  1227         self.assertEqual(rset.rows, [[teid]])
  1222         rset = self.execute('Any T WHERE T tags X')
  1228         rset = self.execute('Any T WHERE T tags X')
  1223         self.assertEquals(rset.rows, [[teid]])
  1229         self.assertEqual(rset.rows, [[teid]])
  1224 
  1230 
  1225     def test_nonregr_2(self):
  1231     def test_nonregr_2(self):
  1226         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
  1232         teid = self.execute("INSERT Tag X: X name 'tag'")[0][0]
  1227         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
  1233         geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]
  1228         self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
  1234         self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",
  1229                        {'g': geid, 't': teid})
  1235                        {'g': geid, 't': teid})
  1230         rset = self.execute('Any X WHERE E eid %(x)s, E tags X',
  1236         rset = self.execute('Any X WHERE E eid %(x)s, E tags X',
  1231                               {'x': teid})
  1237                               {'x': teid})
  1232         self.assertEquals(rset.rows, [[geid]])
  1238         self.assertEqual(rset.rows, [[geid]])
  1233 
  1239 
  1234     def test_nonregr_3(self):
  1240     def test_nonregr_3(self):
  1235         """bad sql generated on the second query (destination_state is not
  1241         """bad sql generated on the second query (destination_state is not
  1236         detected as an inlined relation)
  1242         detected as an inlined relation)
  1237         """
  1243         """
  1238         rset = self.execute('Any S,ES,T WHERE S state_of WF, WF workflow_of ET, ET name "CWUser",'
  1244         rset = self.execute('Any S,ES,T WHERE S state_of WF, WF workflow_of ET, ET name "CWUser",'
  1239                              'ES allowed_transition T, T destination_state S')
  1245                              'ES allowed_transition T, T destination_state S')
  1240         self.assertEquals(len(rset.rows), 2)
  1246         self.assertEqual(len(rset.rows), 2)
  1241 
  1247 
  1242     def test_nonregr_4(self):
  1248     def test_nonregr_4(self):
  1243         # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3
  1249         # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3
  1244         # union queries and that make for instance a 266Ko sql query which is refused
  1250         # union queries and that make for instance a 266Ko sql query which is refused
  1245         # by the server (or client lib)
  1251         # by the server (or client lib)
  1246         rset = self.execute('Any ER,SE,OE WHERE SE name "Comment", ER name "comments", OE name "Comment",'
  1252         rset = self.execute('Any ER,SE,OE WHERE SE name "Comment", ER name "comments", OE name "Comment",'
  1247                             'ER is CWRType, SE is CWEType, OE is CWEType')
  1253                             'ER is CWRType, SE is CWEType, OE is CWEType')
  1248         self.assertEquals(len(rset), 1)
  1254         self.assertEqual(len(rset), 1)
  1249 
  1255 
  1250     def test_nonregr_5(self):
  1256     def test_nonregr_5(self):
  1251         # jpl #15505: equivalent queries returning different result sets
  1257         # jpl #15505: equivalent queries returning different result sets
  1252         teid1 = self.execute("INSERT Folder X: X name 'hop'")[0][0]
  1258         teid1 = self.execute("INSERT Folder X: X name 'hop'")[0][0]
  1253         teid2 = self.execute("INSERT Folder X: X name 'hip'")[0][0]
  1259         teid2 = self.execute("INSERT Folder X: X name 'hip'")[0][0]
  1263                              'N filed_under T, W concerne N,'
  1269                              'N filed_under T, W concerne N,'
  1264                              'W is Affaire, W filed_under A, A eid %s' % (teid1, teid2))
  1270                              'W is Affaire, W filed_under A, A eid %s' % (teid1, teid2))
  1265         rset4 = self.execute('Any N,U WHERE N todo_by U, T eid %s,'
  1271         rset4 = self.execute('Any N,U WHERE N todo_by U, T eid %s,'
  1266                              'N filed_under T, W concerne N,'
  1272                              'N filed_under T, W concerne N,'
  1267                              'W filed_under A, A eid %s' % (teid1, teid2))
  1273                              'W filed_under A, A eid %s' % (teid1, teid2))
  1268         self.assertEquals(rset1.rows, rset2.rows)
  1274         self.assertEqual(rset1.rows, rset2.rows)
  1269         self.assertEquals(rset1.rows, rset3.rows)
  1275         self.assertEqual(rset1.rows, rset3.rows)
  1270         self.assertEquals(rset1.rows, rset4.rows)
  1276         self.assertEqual(rset1.rows, rset4.rows)
  1271 
  1277 
  1272     def test_nonregr_6(self):
  1278     def test_nonregr_6(self):
  1273         self.execute('Any N,COUNT(S) GROUPBY N ORDERBY COUNT(N) WHERE S name N, S is State')
  1279         self.execute('Any N,COUNT(S) GROUPBY N ORDERBY COUNT(N) WHERE S name N, S is State')
  1274 
  1280 
  1275     def test_sqlite_encoding(self):
  1281     def test_sqlite_encoding(self):
  1283                         'modification_date': '2003/03/12 11:00',
  1289                         'modification_date': '2003/03/12 11:00',
  1284                         'creation_date': '2000/07/03 11:00'})
  1290                         'creation_date': '2000/07/03 11:00'})
  1285         rset = self.execute('Any lower(N) ORDERBY LOWER(N) WHERE X is Tag, X name N,'
  1291         rset = self.execute('Any lower(N) ORDERBY LOWER(N) WHERE X is Tag, X name N,'
  1286                             'X owned_by U, U eid %(x)s',
  1292                             'X owned_by U, U eid %(x)s',
  1287                             {'x':self.session.user.eid})
  1293                             {'x':self.session.user.eid})
  1288         self.assertEquals(rset.rows, [[u'\xe9name0']])
  1294         self.assertEqual(rset.rows, [[u'\xe9name0']])
  1289 
  1295 
  1290 
  1296 
  1291     def test_nonregr_description(self):
  1297     def test_nonregr_description(self):
  1292         """check that a correct description is built in case where infered
  1298         """check that a correct description is built in case where infered
  1293         solutions may be "fusionned" into one by the querier while all solutions
  1299         solutions may be "fusionned" into one by the querier while all solutions
  1297         self.execute("INSERT Societe Y: Y nom 'toto'")
  1303         self.execute("INSERT Societe Y: Y nom 'toto'")
  1298         beid = self.execute("INSERT Basket B: B name 'mybasket'")[0][0]
  1304         beid = self.execute("INSERT Basket B: B name 'mybasket'")[0][0]
  1299         self.execute("SET X in_basket B WHERE X is Personne")
  1305         self.execute("SET X in_basket B WHERE X is Personne")
  1300         self.execute("SET X in_basket B WHERE X is Societe")
  1306         self.execute("SET X in_basket B WHERE X is Societe")
  1301         rset = self.execute('Any X WHERE X in_basket B, B eid %s' % beid)
  1307         rset = self.execute('Any X WHERE X in_basket B, B eid %s' % beid)
  1302         self.assertEquals(len(rset), 2)
  1308         self.assertEqual(len(rset), 2)
  1303         self.assertEquals(rset.description, [('Personne',), ('Societe',)])
  1309         self.assertEqual(rset.description, [('Personne',), ('Societe',)])
  1304 
  1310 
  1305 
  1311 
  1306     def test_nonregr_cache_1(self):
  1312     def test_nonregr_cache_1(self):
  1307         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
  1313         peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
  1308         beid = self.execute("INSERT Basket X: X name 'tag'")[0][0]
  1314         beid = self.execute("INSERT Basket X: X name 'tag'")[0][0]
  1309         self.execute("SET X in_basket Y WHERE X is Personne, Y eid %(y)s",
  1315         self.execute("SET X in_basket Y WHERE X is Personne, Y eid %(y)s",
  1310                        {'y': beid})
  1316                        {'y': beid})
  1311         rset = self.execute("Any X WHERE X in_basket B, B eid %(x)s",
  1317         rset = self.execute("Any X WHERE X in_basket B, B eid %(x)s",
  1312                        {'x': beid})
  1318                        {'x': beid})
  1313         self.assertEquals(rset.rows, [[peid]])
  1319         self.assertEqual(rset.rows, [[peid]])
  1314         rset = self.execute("Any X WHERE X in_basket B, B eid %(x)s",
  1320         rset = self.execute("Any X WHERE X in_basket B, B eid %(x)s",
  1315                        {'x': beid})
  1321                        {'x': beid})
  1316         self.assertEquals(rset.rows, [[peid]])
  1322         self.assertEqual(rset.rows, [[peid]])
  1317 
  1323 
  1318     def test_nonregr_has_text_cache(self):
  1324     def test_nonregr_has_text_cache(self):
  1319         eid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
  1325         eid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]
  1320         eid2 = self.execute("INSERT Personne X: X nom 'tag'")[0][0]
  1326         eid2 = self.execute("INSERT Personne X: X nom 'tag'")[0][0]
  1321         self.commit()
  1327         self.commit()
  1322         rset = self.execute("Any X WHERE X has_text %(text)s", {'text': 'bidule'})
  1328         rset = self.execute("Any X WHERE X has_text %(text)s", {'text': 'bidule'})
  1323         self.assertEquals(rset.rows, [[eid1]])
  1329         self.assertEqual(rset.rows, [[eid1]])
  1324         rset = self.execute("Any X WHERE X has_text %(text)s", {'text': 'tag'})
  1330         rset = self.execute("Any X WHERE X has_text %(text)s", {'text': 'tag'})
  1325         self.assertEquals(rset.rows, [[eid2]])
  1331         self.assertEqual(rset.rows, [[eid2]])
  1326 
  1332 
  1327     def test_nonregr_sortterm_management(self):
  1333     def test_nonregr_sortterm_management(self):
  1328         """Error: Variable has no attribute 'sql' in rql2sql.py (visit_variable)
  1334         """Error: Variable has no attribute 'sql' in rql2sql.py (visit_variable)
  1329 
  1335 
  1330         cause: old variable ref inserted into a fresh rqlst copy
  1336         cause: old variable ref inserted into a fresh rqlst copy
  1345         self.execute("SET X filed_under Y WHERE X eid %s, Y eid %s"
  1351         self.execute("SET X filed_under Y WHERE X eid %s, Y eid %s"
  1346                        % (neid2, teid2))
  1352                        % (neid2, teid2))
  1347         self.execute("SET X todo_by Y WHERE X is Note, Y eid %s" % ueid)
  1353         self.execute("SET X todo_by Y WHERE X is Note, Y eid %s" % ueid)
  1348         rset = self.execute('Any N WHERE N todo_by U, N is Note, U eid %s, N filed_under T, T eid %s'
  1354         rset = self.execute('Any N WHERE N todo_by U, N is Note, U eid %s, N filed_under T, T eid %s'
  1349                              % (ueid, teid1))
  1355                              % (ueid, teid1))
  1350         self.assertEquals(len(rset), 1)
  1356         self.assertEqual(len(rset), 1)
  1351 
  1357 
  1352     def test_nonregr_XXX(self):
  1358     def test_nonregr_XXX(self):
  1353         teid = self.execute('Transition S WHERE S name "deactivate"')[0][0]
  1359         teid = self.execute('Transition S WHERE S name "deactivate"')[0][0]
  1354         rset = self.execute('Any O WHERE O is State, '
  1360         rset = self.execute('Any O WHERE O is State, '
  1355                              'S eid %(x)s, S transition_of ET, O state_of ET', {'x': teid})
  1361                              'S eid %(x)s, S transition_of ET, O state_of ET', {'x': teid})
  1356         self.assertEquals(len(rset), 2)
  1362         self.assertEqual(len(rset), 2)
  1357         rset = self.execute('Any O WHERE O is State, NOT S destination_state O, '
  1363         rset = self.execute('Any O WHERE O is State, NOT S destination_state O, '
  1358                              'S eid %(x)s, S transition_of ET, O state_of ET', {'x': teid})
  1364                              'S eid %(x)s, S transition_of ET, O state_of ET', {'x': teid})
  1359         self.assertEquals(len(rset), 1)
  1365         self.assertEqual(len(rset), 1)
  1360 
  1366 
  1361 
  1367 
  1362     def test_nonregr_set_datetime(self):
  1368     def test_nonregr_set_datetime(self):
  1363         # huum, psycopg specific
  1369         # huum, psycopg specific
  1364         self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()})
  1370         self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()})
  1371 
  1377 
  1372     def test_nonregr_u_owned_by_u(self):
  1378     def test_nonregr_u_owned_by_u(self):
  1373         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G "
  1379         ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G "
  1374                              "WHERE G name 'users'")[0][0]
  1380                              "WHERE G name 'users'")[0][0]
  1375         rset = self.execute("CWUser U")
  1381         rset = self.execute("CWUser U")
  1376         self.assertEquals(len(rset), 3) # bob + admin + anon
  1382         self.assertEqual(len(rset), 3) # bob + admin + anon
  1377         rset = self.execute("Any U WHERE NOT U owned_by U")
  1383         rset = self.execute("Any U WHERE NOT U owned_by U")
  1378         self.assertEquals(len(rset), 0) # even admin created at repo initialization time should belong to itself
  1384         self.assertEqual(len(rset), 0) # even admin created at repo initialization time should belong to itself
  1379 
  1385 
  1380     def test_nonreg_update_index(self):
  1386     def test_nonreg_update_index(self):
  1381         # this is the kind of queries generated by "cubicweb-ctl db-check -ry"
  1387         # this is the kind of queries generated by "cubicweb-ctl db-check -ry"
  1382         self.execute("SET X description D WHERE X is State, X description D")
  1388         self.execute("SET X description D WHERE X is State, X description D")
  1383 
  1389 
  1388     def test_nonregr_orderby(self):
  1394     def test_nonregr_orderby(self):
  1389         seid = self.execute('Any X WHERE X name "activated"')[0][0]
  1395         seid = self.execute('Any X WHERE X name "activated"')[0][0]
  1390         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)
  1396         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)
  1391 
  1397 
  1392     def test_nonregr_solution_cache(self):
  1398     def test_nonregr_solution_cache(self):
  1393         self.skip('XXX should be fixed or documented') # (doesn't occur if cache key is provided.)
  1399         self.skipTest('XXX should be fixed or documented') # (doesn't occur if cache key is provided.)
  1394         rset = self.execute('Any X WHERE X is CWUser, X eid %(x)s', {'x':self.ueid})
  1400         rset = self.execute('Any X WHERE X is CWUser, X eid %(x)s', {'x':self.ueid})
  1395         self.assertEquals(len(rset), 1)
  1401         self.assertEqual(len(rset), 1)
  1396         rset = self.execute('Any X WHERE X is CWUser, X eid %(x)s', {'x':12345})
  1402         rset = self.execute('Any X WHERE X is CWUser, X eid %(x)s', {'x':12345})
  1397         self.assertEquals(len(rset), 0)
  1403         self.assertEqual(len(rset), 0)
  1398 
  1404 
  1399     def test_nonregr_final_norestr(self):
  1405     def test_nonregr_final_norestr(self):
  1400         self.assertRaises(BadRQLQuery, self.execute, 'Date X')
  1406         self.assertRaises(BadRQLQuery, self.execute, 'Date X')
  1401 
  1407 
  1402 
  1408