24 """TestCase for cubicweb.rset.attr_desc_iterator""" |
24 """TestCase for cubicweb.rset.attr_desc_iterator""" |
25 |
25 |
26 def test_relations_description(self): |
26 def test_relations_description(self): |
27 """tests relations_description() function""" |
27 """tests relations_description() function""" |
28 queries = { |
28 queries = { |
29 'Any U,L,M where U is EUser, U login L, U mail M' : [(1, 'login', 'subject'), (2, 'mail', 'subject')], |
29 'Any U,L,M where U is CWUser, U login L, U mail M' : [(1, 'login', 'subject'), (2, 'mail', 'subject')], |
30 'Any U,L,M where U is EUser, L is Foo, U mail M' : [(2, 'mail', 'subject')], |
30 'Any U,L,M where U is CWUser, L is Foo, U mail M' : [(2, 'mail', 'subject')], |
31 'Any C,P where C is Company, C employs P' : [(1, 'employs', 'subject')], |
31 'Any C,P where C is Company, C employs P' : [(1, 'employs', 'subject')], |
32 'Any C,P where C is Company, P employed_by P' : [], |
32 'Any C,P where C is Company, P employed_by P' : [], |
33 'Any C where C is Company, C employs P' : [], |
33 'Any C where C is Company, C employs P' : [], |
34 } |
34 } |
35 for rql, relations in queries.items(): |
35 for rql, relations in queries.items(): |
37 self.assertEquals((rql, result), (rql, relations)) |
37 self.assertEquals((rql, result), (rql, relations)) |
38 |
38 |
39 def test_relations_description_indexed(self): |
39 def test_relations_description_indexed(self): |
40 """tests relations_description() function""" |
40 """tests relations_description() function""" |
41 queries = { |
41 queries = { |
42 'Any C,U,P,L,M where C is Company, C employs P, U is EUser, U login L, U mail M' : |
42 'Any C,U,P,L,M where C is Company, C employs P, U is CWUser, U login L, U mail M' : |
43 {0: [(2,'employs', 'subject')], 1: [(3,'login', 'subject'), (4,'mail', 'subject')]}, |
43 {0: [(2,'employs', 'subject')], 1: [(3,'login', 'subject'), (4,'mail', 'subject')]}, |
44 } |
44 } |
45 for rql, results in queries.items(): |
45 for rql, results in queries.items(): |
46 for var_index, relations in results.items(): |
46 for var_index, relations in results.items(): |
47 result = list(attr_desc_iterator(parse(rql).children[0], var_index)) |
47 result = list(attr_desc_iterator(parse(rql).children[0], var_index)) |
52 class ResultSetTC(EnvBasedTC): |
52 class ResultSetTC(EnvBasedTC): |
53 |
53 |
54 def setUp(self): |
54 def setUp(self): |
55 super(ResultSetTC, self).setUp() |
55 super(ResultSetTC, self).setUp() |
56 self.rset = ResultSet([[12, 'adim'], [13, 'syt']], |
56 self.rset = ResultSet([[12, 'adim'], [13, 'syt']], |
57 'Any U,L where U is EUser, U login L', |
57 'Any U,L where U is CWUser, U login L', |
58 description=[['EUser', 'String'], ['Bar', 'String']]) |
58 description=[['CWUser', 'String'], ['Bar', 'String']]) |
59 self.rset.vreg = self.vreg |
59 self.rset.vreg = self.vreg |
60 |
60 |
61 def compare_urls(self, url1, url2): |
61 def compare_urls(self, url1, url2): |
62 info1 = urlsplit(url1) |
62 info1 = urlsplit(url1) |
63 info2 = urlsplit(url2) |
63 info2 = urlsplit(url2) |
81 self.compare_urls(req.build_url('view', _restpath=''), baseurl) |
81 self.compare_urls(req.build_url('view', _restpath=''), baseurl) |
82 |
82 |
83 |
83 |
84 def test_resultset_build(self): |
84 def test_resultset_build(self): |
85 """test basic build of a ResultSet""" |
85 """test basic build of a ResultSet""" |
86 rs = ResultSet([1,2,3], 'EGroup X', description=['EGroup', 'EGroup', 'EGroup']) |
86 rs = ResultSet([1,2,3], 'CWGroup X', description=['CWGroup', 'CWGroup', 'CWGroup']) |
87 self.assertEquals(rs.rowcount, 3) |
87 self.assertEquals(rs.rowcount, 3) |
88 self.assertEquals(rs.rows, [1,2,3]) |
88 self.assertEquals(rs.rows, [1,2,3]) |
89 self.assertEquals(rs.description, ['EGroup', 'EGroup', 'EGroup']) |
89 self.assertEquals(rs.description, ['CWGroup', 'CWGroup', 'CWGroup']) |
90 |
90 |
91 |
91 |
92 def test_resultset_limit(self): |
92 def test_resultset_limit(self): |
93 rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']], |
93 rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']], |
94 'Any U,L where U is EUser, U login L', |
94 'Any U,L where U is CWUser, U login L', |
95 description=[['EUser', 'String']] * 3) |
95 description=[['CWUser', 'String']] * 3) |
96 rs.req = self.request() |
96 rs.req = self.request() |
97 rs.vreg = self.env.vreg |
97 rs.vreg = self.env.vreg |
98 |
98 |
99 self.assertEquals(rs.limit(2).rows, [[12000, 'adim'], [13000, 'syt']]) |
99 self.assertEquals(rs.limit(2).rows, [[12000, 'adim'], [13000, 'syt']]) |
100 rs2 = rs.limit(2, offset=1) |
100 rs2 = rs.limit(2, offset=1) |
104 self.assertEquals(rs.limit(2, offset=3).rows, []) |
104 self.assertEquals(rs.limit(2, offset=3).rows, []) |
105 |
105 |
106 |
106 |
107 def test_resultset_filter(self): |
107 def test_resultset_filter(self): |
108 rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']], |
108 rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']], |
109 'Any U,L where U is EUser, U login L', |
109 'Any U,L where U is CWUser, U login L', |
110 description=[['EUser', 'String']] * 3) |
110 description=[['CWUser', 'String']] * 3) |
111 rs.req = self.request() |
111 rs.req = self.request() |
112 rs.vreg = self.env.vreg |
112 rs.vreg = self.env.vreg |
113 def test_filter(entity): |
113 def test_filter(entity): |
114 return entity.login != 'nico' |
114 return entity.login != 'nico' |
115 |
115 |
117 self.assertEquals(len(rs2), 2) |
117 self.assertEquals(len(rs2), 2) |
118 self.assertEquals([login for _, login in rs2], ['adim', 'syt']) |
118 self.assertEquals([login for _, login in rs2], ['adim', 'syt']) |
119 |
119 |
120 def test_resultset_transform(self): |
120 def test_resultset_transform(self): |
121 rs = ResultSet([[12, 'adim'], [13, 'syt'], [14, 'nico']], |
121 rs = ResultSet([[12, 'adim'], [13, 'syt'], [14, 'nico']], |
122 'Any U,L where U is EUser, U login L', |
122 'Any U,L where U is CWUser, U login L', |
123 description=[['EUser', 'String']] * 3) |
123 description=[['CWUser', 'String']] * 3) |
124 rs.req = self.request() |
124 rs.req = self.request() |
125 def test_transform(row, desc): |
125 def test_transform(row, desc): |
126 return row[1:], desc[1:] |
126 return row[1:], desc[1:] |
127 rs2 = rs.transformed_rset(test_transform) |
127 rs2 = rs.transformed_rset(test_transform) |
128 |
128 |
129 self.assertEquals(len(rs2), 3) |
129 self.assertEquals(len(rs2), 3) |
130 self.assertEquals(list(rs2), [['adim'],['syt'],['nico']]) |
130 self.assertEquals(list(rs2), [['adim'],['syt'],['nico']]) |
131 |
131 |
132 def test_resultset_sort(self): |
132 def test_resultset_sort(self): |
133 rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']], |
133 rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']], |
134 'Any U,L where U is EUser, U login L', |
134 'Any U,L where U is CWUser, U login L', |
135 description=[['EUser', 'String']] * 3) |
135 description=[['CWUser', 'String']] * 3) |
136 rs.req = self.request() |
136 rs.req = self.request() |
137 rs.vreg = self.env.vreg |
137 rs.vreg = self.env.vreg |
138 |
138 |
139 rs2 = rs.sorted_rset(lambda e:e['login']) |
139 rs2 = rs.sorted_rset(lambda e:e['login']) |
140 self.assertEquals(len(rs2), 3) |
140 self.assertEquals(len(rs2), 3) |
158 rs = ResultSet([[12000, 'adim', u'Adim chez les pinguins'], |
158 rs = ResultSet([[12000, 'adim', u'Adim chez les pinguins'], |
159 [12000, 'adim', u'Jardiner facile'], |
159 [12000, 'adim', u'Jardiner facile'], |
160 [13000, 'syt', u'Le carrelage en 42 leçons'], |
160 [13000, 'syt', u'Le carrelage en 42 leçons'], |
161 [14000, 'nico', u'La tarte tatin en 15 minutes'], |
161 [14000, 'nico', u'La tarte tatin en 15 minutes'], |
162 [14000, 'nico', u"L'épluchage du castor commun"]], |
162 [14000, 'nico', u"L'épluchage du castor commun"]], |
163 'Any U, L, T WHERE U is EUser, U login L,'\ |
163 'Any U, L, T WHERE U is CWUser, U login L,'\ |
164 'D created_by U, D title T', |
164 'D created_by U, D title T', |
165 description=[['EUser', 'String', 'String']] * 5) |
165 description=[['CWUser', 'String', 'String']] * 5) |
166 rs.req = self.request() |
166 rs.req = self.request() |
167 rs.vreg = self.env.vreg |
167 rs.vreg = self.env.vreg |
168 |
168 |
169 rsets = rs.split_rset(lambda e:e['login']) |
169 rsets = rs.split_rset(lambda e:e['login']) |
170 self.assertEquals(len(rsets), 3) |
170 self.assertEquals(len(rsets), 3) |
204 rqlst1 = self.rset.syntax_tree() |
204 rqlst1 = self.rset.syntax_tree() |
205 rqlst2 = self.rset.syntax_tree() |
205 rqlst2 = self.rset.syntax_tree() |
206 self.assert_(rqlst1 is rqlst2) |
206 self.assert_(rqlst1 is rqlst2) |
207 |
207 |
208 def test_get_entity_simple(self): |
208 def test_get_entity_simple(self): |
209 self.add_entity('EUser', login=u'adim', upassword='adim', |
209 self.add_entity('CWUser', login=u'adim', upassword='adim', |
210 surname=u'di mascio', firstname=u'adrien') |
210 surname=u'di mascio', firstname=u'adrien') |
211 e = self.entity('Any X,T WHERE X login "adim", X surname T') |
211 e = self.entity('Any X,T WHERE X login "adim", X surname T') |
212 self.assertEquals(e['surname'], 'di mascio') |
212 self.assertEquals(e['surname'], 'di mascio') |
213 self.assertRaises(KeyError, e.__getitem__, 'firstname') |
213 self.assertRaises(KeyError, e.__getitem__, 'firstname') |
214 self.assertRaises(KeyError, e.__getitem__, 'creation_date') |
214 self.assertRaises(KeyError, e.__getitem__, 'creation_date') |
249 |
249 |
250 def test_get_entity_relation_cache_compt(self): |
250 def test_get_entity_relation_cache_compt(self): |
251 rset = self.execute('Any X,S WHERE X in_state S, X login "anon"') |
251 rset = self.execute('Any X,S WHERE X in_state S, X login "anon"') |
252 e = rset.get_entity(0, 0) |
252 e = rset.get_entity(0, 0) |
253 seid = self.execute('State X WHERE X name "activated"')[0][0] |
253 seid = self.execute('State X WHERE X name "activated"')[0][0] |
254 # for_user / in_group are prefetched in EUser __init__, in_state should |
254 # for_user / in_group are prefetched in CWUser __init__, in_state should |
255 # be filed from our query rset |
255 # be filed from our query rset |
256 self.assertEquals(pprelcachedict(e._related_cache), |
256 self.assertEquals(pprelcachedict(e._related_cache), |
257 [('in_state_subject', [seid])]) |
257 [('in_state_subject', [seid])]) |
258 |
258 |
259 def test_get_entity_advanced_prefilled_cache(self): |
259 def test_get_entity_advanced_prefilled_cache(self): |
274 self.assertEquals(s['name'], 'activated') |
274 self.assertEquals(s['name'], 'activated') |
275 self.assertRaises(KeyError, s.__getitem__, 'description') |
275 self.assertRaises(KeyError, s.__getitem__, 'description') |
276 |
276 |
277 |
277 |
278 def test_get_entity_cache_with_left_outer_join(self): |
278 def test_get_entity_cache_with_left_outer_join(self): |
279 eid = self.execute('INSERT EUser E: E login "joe", E upassword "joe", E in_group G ' |
279 eid = self.execute('INSERT CWUser E: E login "joe", E upassword "joe", E in_group G ' |
280 'WHERE G name "users"')[0][0] |
280 'WHERE G name "users"')[0][0] |
281 rset = self.execute('Any X,E WHERE X eid %(x)s, X primary_email E?', {'x': eid}) |
281 rset = self.execute('Any X,E WHERE X eid %(x)s, X primary_email E?', {'x': eid}) |
282 e = rset.get_entity(0, 0) |
282 e = rset.get_entity(0, 0) |
283 # if any of the assertion below fails with a KeyError, the relation is not cached |
283 # if any of the assertion below fails with a KeyError, the relation is not cached |
284 # related entities should be an empty list |
284 # related entities should be an empty list |
292 def test_get_entity_union(self): |
292 def test_get_entity_union(self): |
293 e = self.add_entity('Bookmark', title=u'manger', path=u'path') |
293 e = self.add_entity('Bookmark', title=u'manger', path=u'path') |
294 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
294 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
295 '((Any X,N WHERE X is Bookmark, X title N)' |
295 '((Any X,N WHERE X is Bookmark, X title N)' |
296 ' UNION ' |
296 ' UNION ' |
297 ' (Any X,N WHERE X is EGroup, X name N))') |
297 ' (Any X,N WHERE X is CWGroup, X name N))') |
298 expected = (('EGroup', 'guests'), ('EGroup', 'managers'), |
298 expected = (('CWGroup', 'guests'), ('CWGroup', 'managers'), |
299 ('Bookmark', 'manger'), ('EGroup', 'owners'), |
299 ('Bookmark', 'manger'), ('CWGroup', 'owners'), |
300 ('EGroup', 'users')) |
300 ('CWGroup', 'users')) |
301 for entity in rset.entities(): # test get_entity for each row actually |
301 for entity in rset.entities(): # test get_entity for each row actually |
302 etype, n = expected[entity.row] |
302 etype, n = expected[entity.row] |
303 self.assertEquals(entity.id, etype) |
303 self.assertEquals(entity.id, etype) |
304 attr = etype == 'Bookmark' and 'title' or 'name' |
304 attr = etype == 'Bookmark' and 'title' or 'name' |
305 self.assertEquals(entity[attr], n) |
305 self.assertEquals(entity[attr], n) |
312 self.assertEquals(rtype, None) |
312 self.assertEquals(rtype, None) |
313 |
313 |
314 def test_related_entity_union_subquery(self): |
314 def test_related_entity_union_subquery(self): |
315 e = self.add_entity('Bookmark', title=u'aaaa', path=u'path') |
315 e = self.add_entity('Bookmark', title=u'aaaa', path=u'path') |
316 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
316 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
317 '((Any X,N WHERE X is EGroup, X name N)' |
317 '((Any X,N WHERE X is CWGroup, X name N)' |
318 ' UNION ' |
318 ' UNION ' |
319 ' (Any X,N WHERE X is Bookmark, X title N))') |
319 ' (Any X,N WHERE X is Bookmark, X title N))') |
320 entity, rtype = rset.related_entity(0, 1) |
320 entity, rtype = rset.related_entity(0, 1) |
321 self.assertEquals(entity.eid, e.eid) |
321 self.assertEquals(entity.eid, e.eid) |
322 self.assertEquals(rtype, 'title') |
322 self.assertEquals(rtype, 'title') |
323 entity, rtype = rset.related_entity(1, 1) |
323 entity, rtype = rset.related_entity(1, 1) |
324 self.assertEquals(entity.id, 'EGroup') |
324 self.assertEquals(entity.id, 'CWGroup') |
325 self.assertEquals(rtype, 'name') |
325 self.assertEquals(rtype, 'name') |
326 rset = self.execute('Any X,N ORDERBY N WHERE X is Bookmark WITH X,N BEING ' |
326 rset = self.execute('Any X,N ORDERBY N WHERE X is Bookmark WITH X,N BEING ' |
327 '((Any X,N WHERE X is EGroup, X name N)' |
327 '((Any X,N WHERE X is CWGroup, X name N)' |
328 ' UNION ' |
328 ' UNION ' |
329 ' (Any X,N WHERE X is Bookmark, X title N))') |
329 ' (Any X,N WHERE X is Bookmark, X title N))') |
330 entity, rtype = rset.related_entity(0, 1) |
330 entity, rtype = rset.related_entity(0, 1) |
331 self.assertEquals(entity.eid, e.eid) |
331 self.assertEquals(entity.eid, e.eid) |
332 self.assertEquals(rtype, 'title') |
332 self.assertEquals(rtype, 'title') |
334 def test_entities(self): |
334 def test_entities(self): |
335 rset = self.execute('Any U,G WHERE U in_group G') |
335 rset = self.execute('Any U,G WHERE U in_group G') |
336 # make sure we have at least one element |
336 # make sure we have at least one element |
337 self.failUnless(rset) |
337 self.failUnless(rset) |
338 self.assertEquals(set(e.e_schema.type for e in rset.entities(0)), |
338 self.assertEquals(set(e.e_schema.type for e in rset.entities(0)), |
339 set(['EUser',])) |
339 set(['CWUser',])) |
340 self.assertEquals(set(e.e_schema.type for e in rset.entities(1)), |
340 self.assertEquals(set(e.e_schema.type for e in rset.entities(1)), |
341 set(['EGroup',])) |
341 set(['CWGroup',])) |
342 |
342 |
343 def test_printable_rql(self): |
343 def test_printable_rql(self): |
344 rset = self.execute(u'EEType X WHERE X final FALSE, X meta FALSE') |
344 rset = self.execute(u'CWEType X WHERE X final FALSE, X meta FALSE') |
345 self.assertEquals(rset.printable_rql(), |
345 self.assertEquals(rset.printable_rql(), |
346 'Any X WHERE X final FALSE, X meta FALSE, X is EEType') |
346 'Any X WHERE X final FALSE, X meta FALSE, X is CWEType') |
347 |
347 |
348 |
348 |
349 def test_searched_text(self): |
349 def test_searched_text(self): |
350 rset = self.execute(u'Any X WHERE X has_text "foobar"') |
350 rset = self.execute(u'Any X WHERE X has_text "foobar"') |
351 self.assertEquals(rset.searched_text(), 'foobar') |
351 self.assertEquals(rset.searched_text(), 'foobar') |