test/unittest_rset.py
brancholdstable
changeset 6665 90f2f20367bc
parent 6340 470d8e828fda
child 6366 1806148d6ce8
equal deleted inserted replaced
6018:f4d1d5d9ccbb 6665:90f2f20367bc
    50             'Any C,P where C is Company, P employed_by P' : [],
    50             'Any C,P where C is Company, P employed_by P' : [],
    51             'Any C where C is Company, C employs P' : [],
    51             'Any C where C is Company, C employs P' : [],
    52             }
    52             }
    53         for rql, relations in queries.items():
    53         for rql, relations in queries.items():
    54             result = list(attr_desc_iterator(parse(rql).children[0]))
    54             result = list(attr_desc_iterator(parse(rql).children[0]))
    55             self.assertEquals((rql, result), (rql, relations))
    55             self.assertEqual((rql, result), (rql, relations))
    56 
    56 
    57     def test_relations_description_indexed(self):
    57     def test_relations_description_indexed(self):
    58         """tests relations_description() function"""
    58         """tests relations_description() function"""
    59         queries = {
    59         queries = {
    60             'Any C,U,P,L,M where C is Company, C employs P, U is CWUser, U login L, U mail M' :
    60             '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')]},
    61             {0: [(2,'employs', 'subject')], 1: [(3,'login', 'subject'), (4,'mail', 'subject')]},
    62             }
    62             }
    63         for rql, results in queries.items():
    63         for rql, results in queries.items():
    64             for var_index, relations in results.items():
    64             for var_index, relations in results.items():
    65                 result = list(attr_desc_iterator(parse(rql).children[0], var_index))
    65                 result = list(attr_desc_iterator(parse(rql).children[0], var_index))
    66                 self.assertEquals(result, relations)
    66                 self.assertEqual(result, relations)
    67 
    67 
    68 
    68 
    69 
    69 
    70 class ResultSetTC(CubicWebTC):
    70 class ResultSetTC(CubicWebTC):
    71 
    71 
    77         self.rset.req = mock_object(vreg=self.vreg)
    77         self.rset.req = mock_object(vreg=self.vreg)
    78 
    78 
    79     def compare_urls(self, url1, url2):
    79     def compare_urls(self, url1, url2):
    80         info1 = urlsplit(url1)
    80         info1 = urlsplit(url1)
    81         info2 = urlsplit(url2)
    81         info2 = urlsplit(url2)
    82         self.assertEquals(info1[:3], info2[:3])
    82         self.assertEqual(info1[:3], info2[:3])
    83         if info1[3] != info2[3]:
    83         if info1[3] != info2[3]:
    84             params1 = dict(pair.split('=') for pair in info1[3].split('&'))
    84             params1 = dict(pair.split('=') for pair in info1[3].split('&'))
    85             params2 = dict(pair.split('=') for pair in info1[3].split('&'))
    85             params2 = dict(pair.split('=') for pair in info1[3].split('&'))
    86             self.assertDictEquals(params1, params2)
    86             self.assertDictEqual(params1, params2)
    87 
    87 
    88     def test_pickle(self):
    88     def test_pickle(self):
    89         del self.rset.req
    89         del self.rset.req
    90         self.assertEquals(len(pickle.dumps(self.rset)), 392)
    90         self.assertEqual(len(pickle.dumps(self.rset)), 392)
    91 
    91 
    92     def test_build_url(self):
    92     def test_build_url(self):
    93         req = self.request()
    93         req = self.request()
    94         baseurl = req.base_url()
    94         baseurl = req.base_url()
    95         self.compare_urls(req.build_url('view', vid='foo', rql='yo'),
    95         self.compare_urls(req.build_url('view', vid='foo', rql='yo'),
   103 
   103 
   104 
   104 
   105     def test_resultset_build(self):
   105     def test_resultset_build(self):
   106         """test basic build of a ResultSet"""
   106         """test basic build of a ResultSet"""
   107         rs = ResultSet([1,2,3], 'CWGroup X', description=['CWGroup', 'CWGroup', 'CWGroup'])
   107         rs = ResultSet([1,2,3], 'CWGroup X', description=['CWGroup', 'CWGroup', 'CWGroup'])
   108         self.assertEquals(rs.rowcount, 3)
   108         self.assertEqual(rs.rowcount, 3)
   109         self.assertEquals(rs.rows, [1,2,3])
   109         self.assertEqual(rs.rows, [1,2,3])
   110         self.assertEquals(rs.description, ['CWGroup', 'CWGroup', 'CWGroup'])
   110         self.assertEqual(rs.description, ['CWGroup', 'CWGroup', 'CWGroup'])
   111 
   111 
   112 
   112 
   113     def test_resultset_limit(self):
   113     def test_resultset_limit(self):
   114         rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
   114         rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
   115                        'Any U,L where U is CWUser, U login L',
   115                        'Any U,L where U is CWUser, U login L',
   116                        description=[['CWUser', 'String']] * 3)
   116                        description=[['CWUser', 'String']] * 3)
   117         rs.req = self.request()
   117         rs.req = self.request()
   118         rs.vreg = self.vreg
   118         rs.vreg = self.vreg
   119 
   119 
   120         self.assertEquals(rs.limit(2).rows, [[12000, 'adim'], [13000, 'syt']])
   120         self.assertEqual(rs.limit(2).rows, [[12000, 'adim'], [13000, 'syt']])
   121         rs2 = rs.limit(2, offset=1)
   121         rs2 = rs.limit(2, offset=1)
   122         self.assertEquals(rs2.rows, [[13000, 'syt'], [14000, 'nico']])
   122         self.assertEqual(rs2.rows, [[13000, 'syt'], [14000, 'nico']])
   123         self.assertEquals(rs2.get_entity(0, 0).cw_row, 0)
   123         self.assertEqual(rs2.get_entity(0, 0).cw_row, 0)
   124         self.assertEquals(rs.limit(2, offset=2).rows, [[14000, 'nico']])
   124         self.assertEqual(rs.limit(2, offset=2).rows, [[14000, 'nico']])
   125         self.assertEquals(rs.limit(2, offset=3).rows, [])
   125         self.assertEqual(rs.limit(2, offset=3).rows, [])
   126 
   126 
   127 
   127 
   128     def test_resultset_filter(self):
   128     def test_resultset_filter(self):
   129         rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
   129         rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
   130                        'Any U,L where U is CWUser, U login L',
   130                        'Any U,L where U is CWUser, U login L',
   133         rs.vreg = self.vreg
   133         rs.vreg = self.vreg
   134         def test_filter(entity):
   134         def test_filter(entity):
   135             return entity.login != 'nico'
   135             return entity.login != 'nico'
   136 
   136 
   137         rs2 = rs.filtered_rset(test_filter)
   137         rs2 = rs.filtered_rset(test_filter)
   138         self.assertEquals(len(rs2), 2)
   138         self.assertEqual(len(rs2), 2)
   139         self.assertEquals([login for _, login in rs2], ['adim', 'syt'])
   139         self.assertEqual([login for _, login in rs2], ['adim', 'syt'])
   140 
   140 
   141     def test_resultset_transform(self):
   141     def test_resultset_transform(self):
   142         rs = ResultSet([[12, 'adim'], [13, 'syt'], [14, 'nico']],
   142         rs = ResultSet([[12, 'adim'], [13, 'syt'], [14, 'nico']],
   143                        'Any U,L where U is CWUser, U login L',
   143                        'Any U,L where U is CWUser, U login L',
   144                        description=[['CWUser', 'String']] * 3)
   144                        description=[['CWUser', 'String']] * 3)
   145         rs.req = self.request()
   145         rs.req = self.request()
   146         def test_transform(row, desc):
   146         def test_transform(row, desc):
   147             return row[1:], desc[1:]
   147             return row[1:], desc[1:]
   148         rs2 = rs.transformed_rset(test_transform)
   148         rs2 = rs.transformed_rset(test_transform)
   149 
   149 
   150         self.assertEquals(len(rs2), 3)
   150         self.assertEqual(len(rs2), 3)
   151         self.assertEquals(list(rs2), [['adim'],['syt'],['nico']])
   151         self.assertEqual(list(rs2), [['adim'],['syt'],['nico']])
   152 
   152 
   153     def test_resultset_sort(self):
   153     def test_resultset_sort(self):
   154         rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
   154         rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
   155                        'Any U,L where U is CWUser, U login L',
   155                        'Any U,L where U is CWUser, U login L',
   156                        description=[['CWUser', 'String']] * 3)
   156                        description=[['CWUser', 'String']] * 3)
   157         rs.req = self.request()
   157         rs.req = self.request()
   158         rs.vreg = self.vreg
   158         rs.vreg = self.vreg
   159 
   159 
   160         rs2 = rs.sorted_rset(lambda e:e['login'])
   160         rs2 = rs.sorted_rset(lambda e:e['login'])
   161         self.assertEquals(len(rs2), 3)
   161         self.assertEqual(len(rs2), 3)
   162         self.assertEquals([login for _, login in rs2], ['adim', 'nico', 'syt'])
   162         self.assertEqual([login for _, login in rs2], ['adim', 'nico', 'syt'])
   163         # make sure rs is unchanged
   163         # make sure rs is unchanged
   164         self.assertEquals([login for _, login in rs], ['adim', 'syt', 'nico'])
   164         self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])
   165 
   165 
   166         rs2 = rs.sorted_rset(lambda e:e['login'], reverse=True)
   166         rs2 = rs.sorted_rset(lambda e:e['login'], reverse=True)
   167         self.assertEquals(len(rs2), 3)
   167         self.assertEqual(len(rs2), 3)
   168         self.assertEquals([login for _, login in rs2], ['syt', 'nico', 'adim'])
   168         self.assertEqual([login for _, login in rs2], ['syt', 'nico', 'adim'])
   169         # make sure rs is unchanged
   169         # make sure rs is unchanged
   170         self.assertEquals([login for _, login in rs], ['adim', 'syt', 'nico'])
   170         self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])
   171 
   171 
   172         rs3 = rs.sorted_rset(lambda row: row[1], col=-1)
   172         rs3 = rs.sorted_rset(lambda row: row[1], col=-1)
   173         self.assertEquals(len(rs3), 3)
   173         self.assertEqual(len(rs3), 3)
   174         self.assertEquals([login for _, login in rs3], ['adim', 'nico', 'syt'])
   174         self.assertEqual([login for _, login in rs3], ['adim', 'nico', 'syt'])
   175         # make sure rs is unchanged
   175         # make sure rs is unchanged
   176         self.assertEquals([login for _, login in rs], ['adim', 'syt', 'nico'])
   176         self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])
   177 
   177 
   178     def test_resultset_split(self):
   178     def test_resultset_split(self):
   179         rs = ResultSet([[12000, 'adim', u'Adim chez les pinguins'],
   179         rs = ResultSet([[12000, 'adim', u'Adim chez les pinguins'],
   180                         [12000, 'adim', u'Jardiner facile'],
   180                         [12000, 'adim', u'Jardiner facile'],
   181                         [13000, 'syt',  u'Le carrelage en 42 leçons'],
   181                         [13000, 'syt',  u'Le carrelage en 42 leçons'],
   186                        description=[['CWUser', 'String', 'String']] * 5)
   186                        description=[['CWUser', 'String', 'String']] * 5)
   187         rs.req = self.request()
   187         rs.req = self.request()
   188         rs.vreg = self.vreg
   188         rs.vreg = self.vreg
   189 
   189 
   190         rsets = rs.split_rset(lambda e:e['login'])
   190         rsets = rs.split_rset(lambda e:e['login'])
   191         self.assertEquals(len(rsets), 3)
   191         self.assertEqual(len(rsets), 3)
   192         self.assertEquals([login for _, login,_ in rsets[0]], ['adim', 'adim'])
   192         self.assertEqual([login for _, login,_ in rsets[0]], ['adim', 'adim'])
   193         self.assertEquals([login for _, login,_ in rsets[1]], ['syt'])
   193         self.assertEqual([login for _, login,_ in rsets[1]], ['syt'])
   194         self.assertEquals([login for _, login,_ in rsets[2]], ['nico', 'nico'])
   194         self.assertEqual([login for _, login,_ in rsets[2]], ['nico', 'nico'])
   195         # make sure rs is unchanged
   195         # make sure rs is unchanged
   196         self.assertEquals([login for _, login,_ in rs], ['adim', 'adim', 'syt', 'nico', 'nico'])
   196         self.assertEqual([login for _, login,_ in rs], ['adim', 'adim', 'syt', 'nico', 'nico'])
   197 
   197 
   198         rsets = rs.split_rset(lambda e:e['login'], return_dict=True)
   198         rsets = rs.split_rset(lambda e:e['login'], return_dict=True)
   199         self.assertEquals(len(rsets), 3)
   199         self.assertEqual(len(rsets), 3)
   200         self.assertEquals([login for _, login,_ in rsets['nico']], ['nico', 'nico'])
   200         self.assertEqual([login for _, login,_ in rsets['nico']], ['nico', 'nico'])
   201         self.assertEquals([login for _, login,_ in rsets['adim']], ['adim', 'adim'])
   201         self.assertEqual([login for _, login,_ in rsets['adim']], ['adim', 'adim'])
   202         self.assertEquals([login for _, login,_ in rsets['syt']], ['syt'])
   202         self.assertEqual([login for _, login,_ in rsets['syt']], ['syt'])
   203         # make sure rs is unchanged
   203         # make sure rs is unchanged
   204         self.assertEquals([login for _, login,_ in rs], ['adim', 'adim', 'syt', 'nico', 'nico'])
   204         self.assertEqual([login for _, login,_ in rs], ['adim', 'adim', 'syt', 'nico', 'nico'])
   205 
   205 
   206         rsets = rs.split_rset(lambda s: s.count('d'), col=2)
   206         rsets = rs.split_rset(lambda s: s.count('d'), col=2)
   207         self.assertEquals(len(rsets), 2)
   207         self.assertEqual(len(rsets), 2)
   208         self.assertEquals([title for _, _, title in rsets[0]],
   208         self.assertEqual([title for _, _, title in rsets[0]],
   209                           [u"Adim chez les pinguins",
   209                           [u"Adim chez les pinguins",
   210                            u"Jardiner facile",
   210                            u"Jardiner facile",
   211                            u"L'épluchage du castor commun",])
   211                            u"L'épluchage du castor commun",])
   212         self.assertEquals([title for _, _, title in rsets[1]],
   212         self.assertEqual([title for _, _, title in rsets[1]],
   213                           [u"Le carrelage en 42 leçons",
   213                           [u"Le carrelage en 42 leçons",
   214                            u"La tarte tatin en 15 minutes",])
   214                            u"La tarte tatin en 15 minutes",])
   215         # make sure rs is unchanged
   215         # make sure rs is unchanged
   216         self.assertEquals([title for _, _, title in rs],
   216         self.assertEqual([title for _, _, title in rs],
   217                           [u'Adim chez les pinguins',
   217                           [u'Adim chez les pinguins',
   218                            u'Jardiner facile',
   218                            u'Jardiner facile',
   219                            u'Le carrelage en 42 leçons',
   219                            u'Le carrelage en 42 leçons',
   220                            u'La tarte tatin en 15 minutes',
   220                            u'La tarte tatin en 15 minutes',
   221                            u"L'épluchage du castor commun"])
   221                            u"L'épluchage du castor commun"])
   226         rqlst2 = self.rset.syntax_tree()
   226         rqlst2 = self.rset.syntax_tree()
   227         self.assert_(rqlst1 is rqlst2)
   227         self.assert_(rqlst1 is rqlst2)
   228 
   228 
   229     def test_get_entity_simple(self):
   229     def test_get_entity_simple(self):
   230         self.request().create_entity('CWUser', login=u'adim', upassword='adim',
   230         self.request().create_entity('CWUser', login=u'adim', upassword='adim',
   231                         surname=u'di mascio', firstname=u'adrien')
   231                                      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)
   232         e = self.execute('Any X,T WHERE X login "adim", X surname T').get_entity(0, 0)
   233         self.assertEquals(e['surname'], 'di mascio')
   233         self.assertEqual(e['surname'], 'di mascio')
   234         self.assertRaises(KeyError, e.__getitem__, 'firstname')
   234         self.assertRaises(KeyError, e.__getitem__, 'firstname')
   235         self.assertRaises(KeyError, e.__getitem__, 'creation_date')
   235         self.assertRaises(KeyError, e.__getitem__, 'creation_date')
   236         self.assertEquals(pprelcachedict(e._related_cache), [])
   236         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   237         e.complete()
   237         e.complete()
   238         self.assertEquals(e['firstname'], 'adrien')
   238         self.assertEqual(e['firstname'], 'adrien')
   239         self.assertEquals(pprelcachedict(e._related_cache), [])
   239         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   240 
   240 
   241     def test_get_entity_advanced(self):
   241     def test_get_entity_advanced(self):
   242         self.request().create_entity('Bookmark', title=u'zou', path=u'/view')
   242         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"')
   243         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')
   244         rset = self.execute('Any X,Y,XT,YN WHERE X bookmarked_by Y, X title XT, Y login YN')
   245 
   245 
   246         e = rset.get_entity(0, 0)
   246         e = rset.get_entity(0, 0)
   247         self.assertEquals(e.cw_row, 0)
   247         self.assertEqual(e.cw_row, 0)
   248         self.assertEquals(e.cw_col, 0)
   248         self.assertEqual(e.cw_col, 0)
   249         self.assertEquals(e['title'], 'zou')
   249         self.assertEqual(e['title'], 'zou')
   250         self.assertRaises(KeyError, e.__getitem__, 'path')
   250         self.assertRaises(KeyError, e.__getitem__, 'path')
   251         self.assertEquals(e.view('text'), 'zou')
   251         self.assertEqual(e.view('text'), 'zou')
   252         self.assertEquals(pprelcachedict(e._related_cache), [])
   252         self.assertEqual(pprelcachedict(e._cw_related_cache), [])
   253 
   253 
   254         e = rset.get_entity(0, 1)
   254         e = rset.get_entity(0, 1)
   255         self.assertEquals(e.cw_row, 0)
   255         self.assertEqual(e.cw_row, 0)
   256         self.assertEquals(e.cw_col, 1)
   256         self.assertEqual(e.cw_col, 1)
   257         self.assertEquals(e['login'], 'anon')
   257         self.assertEqual(e['login'], 'anon')
   258         self.assertRaises(KeyError, e.__getitem__, 'firstname')
   258         self.assertRaises(KeyError, e.__getitem__, 'firstname')
   259         self.assertEquals(pprelcachedict(e._related_cache),
   259         self.assertEqual(pprelcachedict(e._cw_related_cache),
   260                           [])
   260                           [])
   261         e.complete()
   261         e.complete()
   262         self.assertEquals(e['firstname'], None)
   262         self.assertEqual(e['firstname'], None)
   263         self.assertEquals(e.view('text'), 'anon')
   263         self.assertEqual(e.view('text'), 'anon')
   264         self.assertEquals(pprelcachedict(e._related_cache),
   264         self.assertEqual(pprelcachedict(e._cw_related_cache),
   265                           [])
   265                           [])
   266 
   266 
   267         self.assertRaises(NotAnEntity, rset.get_entity, 0, 2)
   267         self.assertRaises(NotAnEntity, rset.get_entity, 0, 2)
   268         self.assertRaises(NotAnEntity, rset.get_entity, 0, 3)
   268         self.assertRaises(NotAnEntity, rset.get_entity, 0, 3)
   269 
   269 
   271         rset = self.execute('Any X,S WHERE X in_state S, X login "anon"')
   271         rset = self.execute('Any X,S WHERE X in_state S, X login "anon"')
   272         e = rset.get_entity(0, 0)
   272         e = rset.get_entity(0, 0)
   273         seid = self.execute('State X WHERE X name "activated"')[0][0]
   273         seid = self.execute('State X WHERE X name "activated"')[0][0]
   274         # for_user / in_group are prefetched in CWUser __init__, in_state should
   274         # for_user / in_group are prefetched in CWUser __init__, in_state should
   275         # be filed from our query rset
   275         # be filed from our query rset
   276         self.assertEquals(pprelcachedict(e._related_cache),
   276         self.assertEqual(pprelcachedict(e._cw_related_cache),
   277                           [('in_state_subject', [seid])])
   277                           [('in_state_subject', [seid])])
   278 
   278 
   279     def test_get_entity_advanced_prefilled_cache(self):
   279     def test_get_entity_advanced_prefilled_cache(self):
   280         e = self.request().create_entity('Bookmark', title=u'zou', path=u'path')
   280         e = self.request().create_entity('Bookmark', title=u'zou', path=u'path')
   281         self.commit()
   281         self.commit()
   282         rset = self.execute('Any X,U,S,XT,UL,SN WHERE X created_by U, U in_state S, '
   282         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)
   283                             'X title XT, S name SN, U login UL, X eid %s' % e.eid)
   284         e = rset.get_entity(0, 0)
   284         e = rset.get_entity(0, 0)
   285         self.assertEquals(e['title'], 'zou')
   285         self.assertEqual(e['title'], 'zou')
   286         self.assertEquals(pprelcachedict(e._related_cache),
   286         self.assertEqual(pprelcachedict(e._cw_related_cache),
   287                           [('created_by_subject', [5])])
   287                           [('created_by_subject', [5])])
   288         # first level of recursion
   288         # first level of recursion
   289         u = e.created_by[0]
   289         u = e.created_by[0]
   290         self.assertEquals(u['login'], 'admin')
   290         self.assertEqual(u['login'], 'admin')
   291         self.assertRaises(KeyError, u.__getitem__, 'firstname')
   291         self.assertRaises(KeyError, u.__getitem__, 'firstname')
   292         # second level of recursion
   292         # second level of recursion
   293         s = u.in_state[0]
   293         s = u.in_state[0]
   294         self.assertEquals(s['name'], 'activated')
   294         self.assertEqual(s['name'], 'activated')
   295         self.assertRaises(KeyError, s.__getitem__, 'description')
   295         self.assertRaises(KeyError, s.__getitem__, 'description')
   296 
   296 
   297 
   297 
   298     def test_get_entity_cache_with_left_outer_join(self):
   298     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 '
   299         eid = self.execute('INSERT CWUser E: E login "joe", E upassword "joe", E in_group G '
   300                            'WHERE G name "users"')[0][0]
   300                            'WHERE G name "users"')[0][0]
   301         rset = self.execute('Any X,E WHERE X eid %(x)s, X primary_email E?', {'x': eid})
   301         rset = self.execute('Any X,E WHERE X eid %(x)s, X primary_email E?', {'x': eid})
   302         e = rset.get_entity(0, 0)
   302         e = rset.get_entity(0, 0)
   303         # if any of the assertion below fails with a KeyError, the relation is not cached
   303         # if any of the assertion below fails with a KeyError, the relation is not cached
   304         # related entities should be an empty list
   304         # related entities should be an empty list
   305         self.assertEquals(e.related_cache('primary_email', 'subject', True), ())
   305         self.assertEqual(e._cw_relation_cache('primary_email', 'subject', True), ())
   306         # related rset should be an empty rset
   306         # related rset should be an empty rset
   307         cached = e.related_cache('primary_email', 'subject', False)
   307         cached = e._cw_relation_cache('primary_email', 'subject', False)
   308         self.assertIsInstance(cached, ResultSet)
   308         self.assertIsInstance(cached, ResultSet)
   309         self.assertEquals(cached.rowcount, 0)
   309         self.assertEqual(cached.rowcount, 0)
   310 
   310 
   311 
   311 
   312     def test_get_entity_union(self):
   312     def test_get_entity_union(self):
   313         e = self.request().create_entity('Bookmark', title=u'manger', path=u'path')
   313         e = self.request().create_entity('Bookmark', title=u'manger', path=u'path')
   314         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   314         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   318         expected = (('CWGroup', 'guests'), ('CWGroup', 'managers'),
   318         expected = (('CWGroup', 'guests'), ('CWGroup', 'managers'),
   319                     ('Bookmark', 'manger'), ('CWGroup', 'owners'),
   319                     ('Bookmark', 'manger'), ('CWGroup', 'owners'),
   320                     ('CWGroup', 'users'))
   320                     ('CWGroup', 'users'))
   321         for entity in rset.entities(): # test get_entity for each row actually
   321         for entity in rset.entities(): # test get_entity for each row actually
   322             etype, n = expected[entity.cw_row]
   322             etype, n = expected[entity.cw_row]
   323             self.assertEquals(entity.__regid__, etype)
   323             self.assertEqual(entity.__regid__, etype)
   324             attr = etype == 'Bookmark' and 'title' or 'name'
   324             attr = etype == 'Bookmark' and 'title' or 'name'
   325             self.assertEquals(entity[attr], n)
   325             self.assertEqual(entity[attr], n)
   326 
   326 
   327     def test_related_entity_optional(self):
   327     def test_related_entity_optional(self):
   328         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
   328         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')
   329         rset = self.execute('Any B,U,L WHERE B bookmarked_by U?, U login L')
   330         entity, rtype = rset.related_entity(0, 2)
   330         entity, rtype = rset.related_entity(0, 2)
   331         self.assertEquals(entity, None)
   331         self.assertEqual(entity, None)
   332         self.assertEquals(rtype, None)
   332         self.assertEqual(rtype, None)
   333 
   333 
   334     def test_related_entity_union_subquery(self):
   334     def test_related_entity_union_subquery(self):
   335         e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
   335         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 '
   336         rset = self.execute('Any X,N ORDERBY N WITH X,N BEING '
   337                             '((Any X,N WHERE X is CWGroup, X name N)'
   337                             '((Any X,N WHERE X is CWGroup, X name N)'
   338                             ' UNION '
   338                             ' UNION '
   339                             ' (Any X,N WHERE X is Bookmark, X title N))')
   339                             ' (Any X,N WHERE X is Bookmark, X title N))')
   340         entity, rtype = rset.related_entity(0, 1)
   340         entity, rtype = rset.related_entity(0, 1)
   341         self.assertEquals(entity.eid, e.eid)
   341         self.assertEqual(entity.eid, e.eid)
   342         self.assertEquals(rtype, 'title')
   342         self.assertEqual(rtype, 'title')
   343         entity, rtype = rset.related_entity(1, 1)
   343         entity, rtype = rset.related_entity(1, 1)
   344         self.assertEquals(entity.__regid__, 'CWGroup')
   344         self.assertEqual(entity.__regid__, 'CWGroup')
   345         self.assertEquals(rtype, 'name')
   345         self.assertEqual(rtype, 'name')
   346         #
   346         #
   347         rset = self.execute('Any X,N ORDERBY N WHERE X is Bookmark WITH X,N BEING '
   347         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)'
   348                             '((Any X,N WHERE X is CWGroup, X name N)'
   349                             ' UNION '
   349                             ' UNION '
   350                             ' (Any X,N WHERE X is Bookmark, X title N))')
   350                             ' (Any X,N WHERE X is Bookmark, X title N))')
   351         entity, rtype = rset.related_entity(0, 1)
   351         entity, rtype = rset.related_entity(0, 1)
   352         self.assertEquals(entity.eid, e.eid)
   352         self.assertEqual(entity.eid, e.eid)
   353         self.assertEquals(rtype, 'title')
   353         self.assertEqual(rtype, 'title')
   354         #
   354         #
   355         rset = self.execute('Any X,N ORDERBY N WITH N,X BEING '
   355         rset = self.execute('Any X,N ORDERBY N WITH N,X BEING '
   356                             '((Any N,X WHERE X is CWGroup, X name N)'
   356                             '((Any N,X WHERE X is CWGroup, X name N)'
   357                             ' UNION '
   357                             ' UNION '
   358                             ' (Any N,X WHERE X is Bookmark, X title N))')
   358                             ' (Any N,X WHERE X is Bookmark, X title N))')
   359         entity, rtype = rset.related_entity(0, 1)
   359         entity, rtype = rset.related_entity(0, 1)
   360         self.assertEquals(entity.eid, e.eid)
   360         self.assertEqual(entity.eid, e.eid)
   361         self.assertEquals(rtype, 'title')
   361         self.assertEqual(rtype, 'title')
   362 
   362 
   363     def test_related_entity_trap_subquery(self):
   363     def test_related_entity_trap_subquery(self):
   364         req = self.request()
   364         req = self.request()
   365         req.create_entity('Bookmark', title=u'test bookmark', path=u'')
   365         req.create_entity('Bookmark', title=u'test bookmark', path=u'')
   366         self.execute('SET B bookmarked_by U WHERE U login "admin"')
   366         self.execute('SET B bookmarked_by U WHERE U login "admin"')
   367         rset = self.execute('Any B,T,L WHERE B bookmarked_by U, U login L '
   367         rset = self.execute('Any B,T,L WHERE B bookmarked_by U, U login L '
   368                             'WITH B,T BEING (Any B,T WHERE B is Bookmark, B title T)')
   368                             'WITH B,T BEING (Any B,T WHERE B is Bookmark, B title T)')
   369         rset.related_entity(0, 2)
   369         rset.related_entity(0, 2)
   370 
   370 
       
   371     def test_related_entity_subquery_outerjoin(self):
       
   372         rset = self.execute('Any X,S,L WHERE X in_state S '
       
   373                             'WITH X, L BEING (Any X,MAX(L) GROUPBY X '
       
   374                             'WHERE X is CWUser, T? wf_info_for X, T creation_date L)')
       
   375         self.assertEqual(len(rset), 2)
       
   376         rset.related_entity(0, 1)
       
   377         rset.related_entity(0, 2)
       
   378 
   371     def test_entities(self):
   379     def test_entities(self):
   372         rset = self.execute('Any U,G WHERE U in_group G')
   380         rset = self.execute('Any U,G WHERE U in_group G')
   373         # make sure we have at least one element
   381         # make sure we have at least one element
   374         self.failUnless(rset)
   382         self.failUnless(rset)
   375         self.assertEquals(set(e.e_schema.type for e in rset.entities(0)),
   383         self.assertEqual(set(e.e_schema.type for e in rset.entities(0)),
   376                           set(['CWUser',]))
   384                           set(['CWUser',]))
   377         self.assertEquals(set(e.e_schema.type for e in rset.entities(1)),
   385         self.assertEqual(set(e.e_schema.type for e in rset.entities(1)),
   378                           set(['CWGroup',]))
   386                           set(['CWGroup',]))
   379 
   387 
   380     def test_printable_rql(self):
   388     def test_printable_rql(self):
   381         rset = self.execute(u'CWEType X WHERE X final FALSE')
   389         rset = self.execute(u'CWEType X WHERE X final FALSE')
   382         self.assertEquals(rset.printable_rql(),
   390         self.assertEqual(rset.printable_rql(),
   383                           'Any X WHERE X final FALSE, X is CWEType')
   391                           'Any X WHERE X final FALSE, X is CWEType')
   384 
   392 
   385     def test_searched_text(self):
   393     def test_searched_text(self):
   386         rset = self.execute(u'Any X WHERE X has_text "foobar"')
   394         rset = self.execute(u'Any X WHERE X has_text "foobar"')
   387         self.assertEquals(rset.searched_text(), 'foobar')
   395         self.assertEqual(rset.searched_text(), 'foobar')
   388         rset = self.execute(u'Any X WHERE X has_text %(text)s', {'text' : 'foo'})
   396         rset = self.execute(u'Any X WHERE X has_text %(text)s', {'text' : 'foo'})
   389         self.assertEquals(rset.searched_text(), 'foo')
   397         self.assertEqual(rset.searched_text(), 'foo')
   390 
   398 
   391     def test_union_limited_rql(self):
   399     def test_union_limited_rql(self):
   392         rset = self.execute('(Any X,N WHERE X is Bookmark, X title N)'
   400         rset = self.execute('(Any X,N WHERE X is Bookmark, X title N)'
   393                             ' UNION '
   401                             ' UNION '
   394                             '(Any X,N WHERE X is CWGroup, X name N)')
   402                             '(Any X,N WHERE X is CWGroup, X name N)')
   395         rset.limit(2, 10, inplace=True)
   403         rset.limit(2, 10, inplace=True)
   396         self.assertEquals(rset.limited_rql(),
   404         self.assertEqual(rset.limited_rql(),
   397                           'Any A,B LIMIT 2 OFFSET 10 '
   405                           'Any A,B LIMIT 2 OFFSET 10 '
   398                           'WITH A,B BEING ('
   406                           'WITH A,B BEING ('
   399                           '(Any X,N WHERE X is Bookmark, X title N) '
   407                           '(Any X,N WHERE X is Bookmark, X title N) '
   400                           'UNION '
   408                           'UNION '
   401                           '(Any X,N WHERE X is CWGroup, X name N)'
   409                           '(Any X,N WHERE X is CWGroup, X name N)'
   402                           ')')
   410                           ')')
   403 
   411 
   404     def test_count_users_by_date(self):
   412     def test_count_users_by_date(self):
   405         rset = self.execute('Any D, COUNT(U) GROUPBY D WHERE U is CWUser, U creation_date D')
   413         rset = self.execute('Any D, COUNT(U) GROUPBY D WHERE U is CWUser, U creation_date D')
   406         self.assertEquals(rset.related_entity(0,0), (None, None))
   414         self.assertEqual(rset.related_entity(0,0), (None, None))
       
   415 
       
   416     def test_str(self):
       
   417         rset = self.execute('(Any X,N WHERE X is CWGroup, X name N)')
       
   418         self.assertIsInstance(str(rset), basestring)
       
   419         self.assertEqual(len(str(rset).splitlines()), 1)
       
   420 
       
   421     def test_repr(self):
       
   422         rset = self.execute('(Any X,N WHERE X is CWGroup, X name N)')
       
   423         self.assertIsInstance(repr(rset), basestring)
       
   424         self.assertTrue(len(repr(rset).splitlines()) > 1)
       
   425 
       
   426         rset = self.execute('(Any X WHERE X is CWGroup, X name "managers")')
       
   427         self.assertIsInstance(str(rset), basestring)
       
   428         self.assertEqual(len(str(rset).splitlines()), 1)
   407 
   429 
   408 if __name__ == '__main__':
   430 if __name__ == '__main__':
   409     unittest_main()
   431     unittest_main()