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"') |