test/unittest_rset.py
brancholdstable
changeset 7074 e4580e5f0703
parent 6915 99eb71b311e4
child 7298 a448e470c150
equal deleted inserted replaced
6749:48f468f33704 7074:e4580e5f0703
    14 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
    14 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
    15 # details.
    15 # details.
    16 #
    16 #
    17 # You should have received a copy of the GNU Lesser General Public License along
    17 # You should have received a copy of the GNU Lesser General Public License along
    18 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
    18 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
    19 """unit tests for module cubicweb.utils
    19 """unit tests for module cubicweb.utils"""
    20 
       
    21 """
       
    22 
    20 
    23 from urlparse import urlsplit
    21 from urlparse import urlsplit
    24 import pickle
    22 import pickle
    25 
    23 
    26 from rql import parse
    24 from rql import parse
    49             'Any C,P where C is Company, C employs P' : [(1, 'employs', 'subject')],
    47             'Any C,P where C is Company, C employs P' : [(1, 'employs', 'subject')],
    50             'Any C,P where C is Company, P employed_by P' : [],
    48             'Any C,P where C is Company, P employed_by P' : [],
    51             'Any C where C is Company, C employs P' : [],
    49             'Any C where C is Company, C employs P' : [],
    52             }
    50             }
    53         for rql, relations in queries.items():
    51         for rql, relations in queries.items():
    54             result = list(attr_desc_iterator(parse(rql).children[0]))
    52             result = list(attr_desc_iterator(parse(rql).children[0], 0, 0))
    55             self.assertEqual((rql, result), (rql, relations))
    53             self.assertEqual((rql, result), (rql, relations))
    56 
    54 
    57     def test_relations_description_indexed(self):
    55     def test_relations_description_indexed(self):
    58         """tests relations_description() function"""
    56         """tests relations_description() function"""
    59         queries = {
    57         queries = {
    60             'Any C,U,P,L,M where C is Company, C employs P, U is CWUser, U login L, U mail M' :
    58             'Any C,U,P,L,M where C is Company, C employs P, U is CWUser, U login L, U mail M' :
    61             {0: [(2,'employs', 'subject')], 1: [(3,'login', 'subject'), (4,'mail', 'subject')]},
    59             {0: [(2,'employs', 'subject')], 1: [(3,'login', 'subject'), (4,'mail', 'subject')]},
    62             }
    60             }
    63         for rql, results in queries.items():
    61         for rql, results in queries.items():
    64             for var_index, relations in results.items():
    62             for idx, relations in results.items():
    65                 result = list(attr_desc_iterator(parse(rql).children[0], var_index))
    63                 result = list(attr_desc_iterator(parse(rql).children[0], idx, idx))
    66                 self.assertEqual(result, relations)
    64                 self.assertEqual(result, relations)
    67 
    65 
    68 
    66 
    69 
    67 
    70 class ResultSetTC(CubicWebTC):
    68 class ResultSetTC(CubicWebTC):
   155                        'Any U,L where U is CWUser, U login L',
   153                        'Any U,L where U is CWUser, U login L',
   156                        description=[['CWUser', 'String']] * 3)
   154                        description=[['CWUser', 'String']] * 3)
   157         rs.req = self.request()
   155         rs.req = self.request()
   158         rs.vreg = self.vreg
   156         rs.vreg = self.vreg
   159 
   157 
   160         rs2 = rs.sorted_rset(lambda e:e['login'])
   158         rs2 = rs.sorted_rset(lambda e:e.cw_attr_cache['login'])
   161         self.assertEqual(len(rs2), 3)
   159         self.assertEqual(len(rs2), 3)
   162         self.assertEqual([login for _, login in rs2], ['adim', 'nico', 'syt'])
   160         self.assertEqual([login for _, login in rs2], ['adim', 'nico', 'syt'])
   163         # make sure rs is unchanged
   161         # make sure rs is unchanged
   164         self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])
   162         self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])
   165 
   163 
   166         rs2 = rs.sorted_rset(lambda e:e['login'], reverse=True)
   164         rs2 = rs.sorted_rset(lambda e:e.cw_attr_cache['login'], reverse=True)
   167         self.assertEqual(len(rs2), 3)
   165         self.assertEqual(len(rs2), 3)
   168         self.assertEqual([login for _, login in rs2], ['syt', 'nico', 'adim'])
   166         self.assertEqual([login for _, login in rs2], ['syt', 'nico', 'adim'])
   169         # make sure rs is unchanged
   167         # make sure rs is unchanged
   170         self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])
   168         self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])
   171 
   169 
   184                        'Any U, L, T WHERE U is CWUser, U login L,'\
   182                        'Any U, L, T WHERE U is CWUser, U login L,'\
   185                        'D created_by U, D title T',
   183                        'D created_by U, D title T',
   186                        description=[['CWUser', 'String', 'String']] * 5)
   184                        description=[['CWUser', 'String', 'String']] * 5)
   187         rs.req = self.request()
   185         rs.req = self.request()
   188         rs.vreg = self.vreg
   186         rs.vreg = self.vreg
   189 
   187         rsets = rs.split_rset(lambda e:e.cw_attr_cache['login'])
   190         rsets = rs.split_rset(lambda e:e['login'])
       
   191         self.assertEqual(len(rsets), 3)
   188         self.assertEqual(len(rsets), 3)
   192         self.assertEqual([login for _, login,_ in rsets[0]], ['adim', 'adim'])
   189         self.assertEqual([login for _, login,_ in rsets[0]], ['adim', 'adim'])
   193         self.assertEqual([login for _, login,_ in rsets[1]], ['syt'])
   190         self.assertEqual([login for _, login,_ in rsets[1]], ['syt'])
   194         self.assertEqual([login for _, login,_ in rsets[2]], ['nico', 'nico'])
   191         self.assertEqual([login for _, login,_ in rsets[2]], ['nico', 'nico'])
   195         # make sure rs is unchanged
   192         # make sure rs is unchanged
   196         self.assertEqual([login for _, login,_ in rs], ['adim', 'adim', 'syt', 'nico', 'nico'])
   193         self.assertEqual([login for _, login,_ in rs], ['adim', 'adim', 'syt', 'nico', 'nico'])
   197 
   194 
   198         rsets = rs.split_rset(lambda e:e['login'], return_dict=True)
   195         rsets = rs.split_rset(lambda e:e.cw_attr_cache['login'], return_dict=True)
   199         self.assertEqual(len(rsets), 3)
   196         self.assertEqual(len(rsets), 3)
   200         self.assertEqual([login for _, login,_ in rsets['nico']], ['nico', 'nico'])
   197         self.assertEqual([login for _, login,_ in rsets['nico']], ['nico', 'nico'])
   201         self.assertEqual([login for _, login,_ in rsets['adim']], ['adim', 'adim'])
   198         self.assertEqual([login for _, login,_ in rsets['adim']], ['adim', 'adim'])
   202         self.assertEqual([login for _, login,_ in rsets['syt']], ['syt'])
   199         self.assertEqual([login for _, login,_ in rsets['syt']], ['syt'])
   203         # make sure rs is unchanged
   200         # make sure rs is unchanged
   228 
   225 
   229     def test_get_entity_simple(self):
   226     def test_get_entity_simple(self):
   230         self.request().create_entity('CWUser', login=u'adim', upassword='adim',
   227         self.request().create_entity('CWUser', login=u'adim', upassword='adim',
   231                                      surname=u'di mascio', firstname=u'adrien')
   228                                      surname=u'di mascio', firstname=u'adrien')
   232         e = self.execute('Any X,T WHERE X login "adim", X surname T').get_entity(0, 0)
   229         e = self.execute('Any X,T WHERE X login "adim", X surname T').get_entity(0, 0)
   233         self.assertEqual(e['surname'], 'di mascio')
   230         self.assertEqual(e.cw_attr_cache['surname'], 'di mascio')
   234         self.assertRaises(KeyError, e.__getitem__, 'firstname')
   231         self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'firstname')
   235         self.assertRaises(KeyError, e.__getitem__, 'creation_date')
   232         self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'creation_date')
   236         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   233         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   237         e.complete()
   234         e.complete()
   238         self.assertEqual(e['firstname'], 'adrien')
   235         self.assertEqual(e.cw_attr_cache['firstname'], 'adrien')
   239         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   236         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   240 
   237 
   241     def test_get_entity_advanced(self):
   238     def test_get_entity_advanced(self):
   242         self.request().create_entity('Bookmark', title=u'zou', path=u'/view')
   239         self.request().create_entity('Bookmark', title=u'zou', path=u'/view')
   243         self.execute('SET X bookmarked_by Y WHERE X is Bookmark, Y login "anon"')
   240         self.execute('SET X bookmarked_by Y WHERE X is Bookmark, Y login "anon"')
   244         rset = self.execute('Any X,Y,XT,YN WHERE X bookmarked_by Y, X title XT, Y login YN')
   241         rset = self.execute('Any X,Y,XT,YN WHERE X bookmarked_by Y, X title XT, Y login YN')
   245 
   242 
   246         e = rset.get_entity(0, 0)
   243         e = rset.get_entity(0, 0)
   247         self.assertEqual(e.cw_row, 0)
   244         self.assertEqual(e.cw_row, 0)
   248         self.assertEqual(e.cw_col, 0)
   245         self.assertEqual(e.cw_col, 0)
   249         self.assertEqual(e['title'], 'zou')
   246         self.assertEqual(e.cw_attr_cache['title'], 'zou')
   250         self.assertRaises(KeyError, e.__getitem__, 'path')
   247         self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'path')
   251         self.assertEqual(e.view('text'), 'zou')
   248         self.assertEqual(e.view('text'), 'zou')
   252         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   249         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   253 
   250 
   254         e = rset.get_entity(0, 1)
   251         e = rset.get_entity(0, 1)
   255         self.assertEqual(e.cw_row, 0)
   252         self.assertEqual(e.cw_row, 0)
   256         self.assertEqual(e.cw_col, 1)
   253         self.assertEqual(e.cw_col, 1)
   257         self.assertEqual(e['login'], 'anon')
   254         self.assertEqual(e.cw_attr_cache['login'], 'anon')
   258         self.assertRaises(KeyError, e.__getitem__, 'firstname')
   255         self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'firstname')
   259         self.assertEqual(pprelcachedict(e._cw_related_cache),
   256         self.assertEqual(pprelcachedict(e._cw_related_cache),
   260                           [])
   257                           [])
   261         e.complete()
   258         e.complete()
   262         self.assertEqual(e['firstname'], None)
   259         self.assertEqual(e.cw_attr_cache['firstname'], None)
   263         self.assertEqual(e.view('text'), 'anon')
   260         self.assertEqual(e.view('text'), 'anon')
   264         self.assertEqual(pprelcachedict(e._cw_related_cache),
   261         self.assertEqual(pprelcachedict(e._cw_related_cache),
   265                           [])
   262                           [])
   266 
   263 
   267         self.assertRaises(NotAnEntity, rset.get_entity, 0, 2)
   264         self.assertRaises(NotAnEntity, rset.get_entity, 0, 2)
   280         e = self.request().create_entity('Bookmark', title=u'zou', path=u'path')
   277         e = self.request().create_entity('Bookmark', title=u'zou', path=u'path')
   281         self.commit()
   278         self.commit()
   282         rset = self.execute('Any X,U,S,XT,UL,SN WHERE X created_by U, U in_state S, '
   279         rset = self.execute('Any X,U,S,XT,UL,SN WHERE X created_by U, U in_state S, '
   283                             'X title XT, S name SN, U login UL, X eid %s' % e.eid)
   280                             'X title XT, S name SN, U login UL, X eid %s' % e.eid)
   284         e = rset.get_entity(0, 0)
   281         e = rset.get_entity(0, 0)
   285         self.assertEqual(e['title'], 'zou')
   282         self.assertEqual(e.cw_attr_cache['title'], 'zou')
   286         self.assertEqual(pprelcachedict(e._cw_related_cache),
   283         self.assertEqual(pprelcachedict(e._cw_related_cache),
   287                           [('created_by_subject', [5])])
   284                           [('created_by_subject', [self.user().eid])])
   288         # first level of recursion
   285         # first level of recursion
   289         u = e.created_by[0]
   286         u = e.created_by[0]
   290         self.assertEqual(u['login'], 'admin')
   287         self.assertEqual(u.cw_attr_cache['login'], 'admin')
   291         self.assertRaises(KeyError, u.__getitem__, 'firstname')
   288         self.assertRaises(KeyError, u.cw_attr_cache.__getitem__, 'firstname')
   292         # second level of recursion
   289         # second level of recursion
   293         s = u.in_state[0]
   290         s = u.in_state[0]
   294         self.assertEqual(s['name'], 'activated')
   291         self.assertEqual(s.cw_attr_cache['name'], 'activated')
   295         self.assertRaises(KeyError, s.__getitem__, 'description')
   292         self.assertRaises(KeyError, s.cw_attr_cache.__getitem__, 'description')
   296 
   293 
   297 
   294 
   298     def test_get_entity_cache_with_left_outer_join(self):
   295     def test_get_entity_cache_with_left_outer_join(self):
   299         eid = self.execute('INSERT CWUser E: E login "joe", E upassword "joe", E in_group G '
   296         eid = self.execute('INSERT CWUser E: E login "joe", E upassword "joe", E in_group G '
   300                            'WHERE G name "users"')[0][0]
   297                            'WHERE G name "users"')[0][0]
   320                     ('CWGroup', 'users'))
   317                     ('CWGroup', 'users'))
   321         for entity in rset.entities(): # test get_entity for each row actually
   318         for entity in rset.entities(): # test get_entity for each row actually
   322             etype, n = expected[entity.cw_row]
   319             etype, n = expected[entity.cw_row]
   323             self.assertEqual(entity.__regid__, etype)
   320             self.assertEqual(entity.__regid__, etype)
   324             attr = etype == 'Bookmark' and 'title' or 'name'
   321             attr = etype == 'Bookmark' and 'title' or 'name'
   325             self.assertEqual(entity[attr], n)
   322             self.assertEqual(entity.cw_attr_cache[attr], n)
   326 
   323 
   327     def test_related_entity_optional(self):
   324     def test_related_entity_optional(self):
   328         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
   325         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
   329         rset = self.execute('Any B,U,L WHERE B bookmarked_by U?, U login L')
   326         rset = self.execute('Any B,U,L WHERE B bookmarked_by U?, U login L')
   330         entity, rtype = rset.related_entity(0, 2)
   327         entity, rtype = rset.related_entity(0, 2)
   331         self.assertEqual(entity, None)
   328         self.assertEqual(entity, None)
   332         self.assertEqual(rtype, None)
   329         self.assertEqual(rtype, None)
   333 
   330 
   334     def test_related_entity_union_subquery(self):
   331     def test_related_entity_union_subquery_1(self):
   335         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
   332         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
   336         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   333         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   337                             '((Any X,N WHERE X is CWGroup, X name N)'
   334                             '((Any X,N WHERE X is CWGroup, X name N)'
   338                             ' UNION '
   335                             ' UNION '
   339                             ' (Any X,N WHERE X is Bookmark, X title N))')
   336                             ' (Any X,N WHERE X is Bookmark, X title N))')
   340         entity, rtype = rset.related_entity(0, 1)
   337         entity, rtype = rset.related_entity(0, 1)
   341         self.assertEqual(entity.eid, e.eid)
   338         self.assertEqual(entity.eid, e.eid)
   342         self.assertEqual(rtype, 'title')
   339         self.assertEqual(rtype, 'title')
       
   340         self.assertEqual(entity.title, 'aaaa')
   343         entity, rtype = rset.related_entity(1, 1)
   341         entity, rtype = rset.related_entity(1, 1)
   344         self.assertEqual(entity.__regid__, 'CWGroup')
   342         self.assertEqual(entity.__regid__, 'CWGroup')
   345         self.assertEqual(rtype, 'name')
   343         self.assertEqual(rtype, 'name')
   346         #
   344         self.assertEqual(entity.name, 'guests')
       
   345 
       
   346     def test_related_entity_union_subquery_2(self):
       
   347         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
   347         rset = self.execute('Any X,N ORDERBY N WHERE X is Bookmark WITH X,N BEING '
   348         rset = self.execute('Any X,N ORDERBY N WHERE X is Bookmark WITH X,N BEING '
   348                             '((Any X,N WHERE X is CWGroup, X name N)'
   349                             '((Any X,N WHERE X is CWGroup, X name N)'
   349                             ' UNION '
   350                             ' UNION '
   350                             ' (Any X,N WHERE X is Bookmark, X title N))')
   351                             ' (Any X,N WHERE X is Bookmark, X title N))')
   351         entity, rtype = rset.related_entity(0, 1)
   352         entity, rtype = rset.related_entity(0, 1)
   352         self.assertEqual(entity.eid, e.eid)
   353         self.assertEqual(entity.eid, e.eid)
   353         self.assertEqual(rtype, 'title')
   354         self.assertEqual(rtype, 'title')
   354         #
   355         self.assertEqual(entity.title, 'aaaa')
       
   356 
       
   357     def test_related_entity_union_subquery_3(self):
       
   358         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
   355         rset = self.execute('Any X,N ORDERBY N WITH N,X BEING '
   359         rset = self.execute('Any X,N ORDERBY N WITH N,X BEING '
   356                             '((Any N,X WHERE X is CWGroup, X name N)'
   360                             '((Any N,X WHERE X is CWGroup, X name N)'
   357                             ' UNION '
   361                             ' UNION '
   358                             ' (Any N,X WHERE X is Bookmark, X title N))')
   362                             ' (Any N,X WHERE X is Bookmark, X title N))')
   359         entity, rtype = rset.related_entity(0, 1)
   363         entity, rtype = rset.related_entity(0, 1)
   360         self.assertEqual(entity.eid, e.eid)
   364         self.assertEqual(entity.eid, e.eid)
   361         self.assertEqual(rtype, 'title')
   365         self.assertEqual(rtype, 'title')
       
   366         self.assertEqual(entity.title, 'aaaa')
       
   367 
       
   368     def test_related_entity_union_subquery_4(self):
       
   369         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
       
   370         rset = self.execute('Any X,X, N ORDERBY N WITH X,N BEING '
       
   371                             '((Any X,N WHERE X is CWGroup, X name N)'
       
   372                             ' UNION '
       
   373                             ' (Any X,N WHERE X is Bookmark, X title N))')
       
   374         entity, rtype = rset.related_entity(0, 2)
       
   375         self.assertEqual(entity.eid, e.eid)
       
   376         self.assertEqual(rtype, 'title')
       
   377         self.assertEqual(entity.title, 'aaaa')
   362 
   378 
   363     def test_related_entity_trap_subquery(self):
   379     def test_related_entity_trap_subquery(self):
   364         req = self.request()
   380         req = self.request()
   365         req.create_entity('Bookmark', title=u'test bookmark', path=u'')
   381         req.create_entity('Bookmark', title=u'test bookmark', path=u'')
   366         self.execute('SET B bookmarked_by U WHERE U login "admin"')
   382         self.execute('SET B bookmarked_by U WHERE U login "admin"')
   382         self.failUnless(rset)
   398         self.failUnless(rset)
   383         self.assertEqual(set(e.e_schema.type for e in rset.entities(0)),
   399         self.assertEqual(set(e.e_schema.type for e in rset.entities(0)),
   384                           set(['CWUser',]))
   400                           set(['CWUser',]))
   385         self.assertEqual(set(e.e_schema.type for e in rset.entities(1)),
   401         self.assertEqual(set(e.e_schema.type for e in rset.entities(1)),
   386                           set(['CWGroup',]))
   402                           set(['CWGroup',]))
       
   403 
       
   404     def test_iter_rows_with_entities(self):
       
   405         rset = self.execute('Any U,UN,G,GN WHERE U in_group G, U login UN, G name GN')
       
   406         # make sure we have at least one element
       
   407         self.failUnless(rset)
       
   408         out = list(rset.iter_rows_with_entities())[0]
       
   409         self.assertEqual( out[0].login, out[1] )
       
   410         self.assertEqual( out[2].name, out[3] )
   387 
   411 
   388     def test_printable_rql(self):
   412     def test_printable_rql(self):
   389         rset = self.execute(u'CWEType X WHERE X final FALSE')
   413         rset = self.execute(u'CWEType X WHERE X final FALSE')
   390         self.assertEqual(rset.printable_rql(),
   414         self.assertEqual(rset.printable_rql(),
   391                           'Any X WHERE X final FALSE, X is CWEType')
   415                           'Any X WHERE X final FALSE, X is CWEType')