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 |
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() |