37 # a[0] may be a dict or a key/value tuple |
37 # a[0] may be a dict or a key/value tuple |
38 return (sorted(dict(a[0]).items()), [e.expression for e in a[1]]) |
38 return (sorted(dict(a[0]).items()), [e.expression for e in a[1]]) |
39 |
39 |
40 |
40 |
41 def check_plan(self, rql, expected, kwargs=None): |
41 def check_plan(self, rql, expected, kwargs=None): |
42 with self._access.cnx() as cnx: |
42 with self.admin_access.cnx() as cnx: |
43 plan = self._prepare_plan(cnx, rql, kwargs) |
43 plan = self._prepare_plan(cnx, rql, kwargs) |
44 self.planner.build_plan(plan) |
44 self.planner.build_plan(plan) |
45 try: |
45 try: |
46 self.assertEqual(len(plan.steps), len(expected), |
46 self.assertEqual(len(plan.steps), len(expected), |
47 'expected %s steps, got %s' % (len(expected), len(plan.steps))) |
47 'expected %s steps, got %s' % (len(expected), len(plan.steps))) |
192 class BaseQuerierTC(TestCase): |
192 class BaseQuerierTC(TestCase): |
193 repo = None # set this in concrete class |
193 repo = None # set this in concrete class |
194 |
194 |
195 def setUp(self): |
195 def setUp(self): |
196 self.o = self.repo.querier |
196 self.o = self.repo.querier |
197 self._access = RepoAccess(self.repo, 'admin', FakeRequest) |
197 self.admin_access = RepoAccess(self.repo, 'admin', FakeRequest) |
198 self.ueid = self._access._user.eid |
198 self.ueid = self.admin_access._user.eid |
199 assert self.ueid != -1 |
199 assert self.ueid != -1 |
200 self.repo._type_cache = {} # clear cache |
200 self.repo._type_cache = {} # clear cache |
201 self.maxeid = self.get_max_eid() |
201 self.maxeid = self.get_max_eid() |
202 do_monkey_patch() |
202 do_monkey_patch() |
203 self._dumb_sessions = [] |
203 self._dumb_sessions = [] |
204 |
204 |
205 def get_max_eid(self): |
205 def get_max_eid(self): |
206 with self._access.cnx() as cnx: |
206 with self.admin_access.cnx() as cnx: |
207 return cnx.execute('Any MAX(X)')[0][0] |
207 return cnx.execute('Any MAX(X)')[0][0] |
208 |
208 |
209 def cleanup(self): |
209 def cleanup(self): |
210 with self._access.cnx() as cnx: |
210 with self.admin_access.cnx() as cnx: |
211 cnx.execute('DELETE Any X WHERE X eid > %s' % self.maxeid) |
211 cnx.execute('DELETE Any X WHERE X eid > %s' % self.maxeid) |
212 cnx.commit() |
212 cnx.commit() |
213 |
213 |
214 def tearDown(self): |
214 def tearDown(self): |
215 undo_monkey_patch() |
215 undo_monkey_patch() |
216 self.cleanup() |
216 self.cleanup() |
217 assert self._access._user.eid != -1 |
217 assert self.admin_access._user.eid != -1 |
218 |
218 |
219 def set_debug(self, debug): |
219 def set_debug(self, debug): |
220 set_debug(debug) |
220 set_debug(debug) |
221 def debugged(self, debug): |
221 def debugged(self, debug): |
222 return debugged(debug) |
222 return debugged(debug) |
247 |
247 |
248 @contextmanager |
248 @contextmanager |
249 def user_groups_session(self, *groups): |
249 def user_groups_session(self, *groups): |
250 """lightweight session using the current user with hi-jacked groups""" |
250 """lightweight session using the current user with hi-jacked groups""" |
251 # use cnx.user.eid to get correct owned_by relation, unless explicit eid |
251 # use cnx.user.eid to get correct owned_by relation, unless explicit eid |
252 with self._access.cnx() as cnx: |
252 with self.admin_access.cnx() as cnx: |
253 user_eid = cnx.user.eid |
253 user_eid = cnx.user.eid |
254 cnx.user._cw.data[user_session_cache_key(user_eid, 'groups')] = set(groups) |
254 cnx.user._cw.data[user_session_cache_key(user_eid, 'groups')] = set(groups) |
255 yield cnx |
255 yield cnx |
256 |
256 |
257 def qexecute(self, rql, args=None, build_descr=True): |
257 def qexecute(self, rql, args=None, build_descr=True): |
258 with self._access.cnx() as cnx: |
258 with self.admin_access.cnx() as cnx: |
259 try: |
259 try: |
260 return self.o.execute(cnx, rql, args, build_descr) |
260 return self.o.execute(cnx, rql, args, build_descr) |
261 finally: |
261 finally: |
262 if rql.startswith(('INSERT', 'DELETE', 'SET')): |
262 if rql.startswith(('INSERT', 'DELETE', 'SET')): |
263 cnx.commit() |
263 cnx.commit() |