43 from cubicweb import cwconfig, dbapi, devtools, web, server, repoapi |
43 from cubicweb import cwconfig, dbapi, devtools, web, server, repoapi |
44 from cubicweb.utils import json |
44 from cubicweb.utils import json |
45 from cubicweb.sobjects import notification |
45 from cubicweb.sobjects import notification |
46 from cubicweb.web import Redirect, application |
46 from cubicweb.web import Redirect, application |
47 from cubicweb.server.hook import SendMailOp |
47 from cubicweb.server.hook import SendMailOp |
|
48 from cubicweb.server.session import Session |
48 from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS |
49 from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS |
49 from cubicweb.devtools import fake, htmlparser, DEFAULT_EMPTY_DB_ID |
50 from cubicweb.devtools import fake, htmlparser, DEFAULT_EMPTY_DB_ID |
50 from cubicweb.utils import json |
51 from cubicweb.utils import json |
51 |
52 |
52 # low-level utilities ########################################################## |
53 # low-level utilities ########################################################## |
152 def __exit__(self, exctype, exc, tb): |
153 def __exit__(self, exctype, exc, tb): |
153 try: |
154 try: |
154 return self.cnx.__exit__(exctype, exc, tb) |
155 return self.cnx.__exit__(exctype, exc, tb) |
155 finally: |
156 finally: |
156 self.testcase.restore_connection() |
157 self.testcase.restore_connection() |
|
158 |
|
159 # Repoaccess utility ###############################################3########### |
|
160 |
|
161 class RepoAccess(object): |
|
162 """An helper to easily create object to access the repo as a specific user |
|
163 |
|
164 Each RepoAccess have it own session. |
|
165 |
|
166 A repo access can create three type of object: |
|
167 |
|
168 .. automethod:: cubicweb.testlib.RepoAccess.repo_cnx |
|
169 .. automethod:: cubicweb.testlib.RepoAccess.client_cnx |
|
170 .. automethod:: cubicweb.testlib.RepoAccess.web_request |
|
171 |
|
172 The RepoAccess need to be closed to destroy the associated Session. |
|
173 TestCase usually take care of this aspect for the user. |
|
174 |
|
175 .. automethod:: cubicweb.testlib.RepoAccess.close |
|
176 """ |
|
177 |
|
178 def __init__(self, repo, login, requestcls): |
|
179 self._repo = repo |
|
180 self._login = login |
|
181 self.requestcls = requestcls |
|
182 # opening session |
|
183 # |
|
184 # XXX this very hackish code should be cleaned and move on repo. |
|
185 with repo.internal_cnx() as cnx: |
|
186 rset = cnx.execute('CWUser U WHERE U login %(u)s', {'u': login}) |
|
187 user = rset.get_entity(0, 0) |
|
188 user.groups |
|
189 user.properties |
|
190 self._session = Session(user, repo) |
|
191 repo._sessions[self._session.id] = self._session |
|
192 self._session.user._cw = self._session |
|
193 |
|
194 @ contextmanager |
|
195 def repo_cnx(self): |
|
196 """Context manager returning a server side connection for the user""" |
|
197 with self._session.new_cnx() as cnx: |
|
198 yield cnx |
|
199 |
|
200 @ contextmanager |
|
201 def client_cnx(self): |
|
202 """Context manager returning a client side connection for the user""" |
|
203 with repoapi.ClientConnection(self._session) as cnx: |
|
204 yield cnx |
|
205 |
|
206 @ contextmanager |
|
207 def web_request(self, url=None, headers={}, **kwargs): |
|
208 """Context manager returning a web request pre-linked to a client cnx |
|
209 |
|
210 To commit and rollback use:: |
|
211 |
|
212 req.cnx.commit() |
|
213 req.cnx.rolback() |
|
214 """ |
|
215 req = self.requestcls(self._repo.vreg, url=url, headers=headers, form=kwargs) |
|
216 clt_cnx = repoapi.ClientConnection(self._session) |
|
217 req.set_cnx(clt_cnx) |
|
218 with clt_cnx: |
|
219 yield req |
|
220 |
|
221 def close(self): |
|
222 """Close the session associated to the RepoAccess""" |
|
223 if self._session is not None: |
|
224 self._repo.close(self._session.id) |
|
225 self._session = None |
|
226 |
|
227 |
157 |
228 |
158 # base class for cubicweb tests requiring a full cw environments ############### |
229 # base class for cubicweb tests requiring a full cw environments ############### |
159 |
230 |
160 class CubicWebTC(TestCase): |
231 class CubicWebTC(TestCase): |
161 """abstract class for test using an apptest environment |
232 """abstract class for test using an apptest environment |
184 self._admin_session = None |
255 self._admin_session = None |
185 self._admin_clt_cnx = None |
256 self._admin_clt_cnx = None |
186 self._current_session = None |
257 self._current_session = None |
187 self._current_clt_cnx = None |
258 self._current_clt_cnx = None |
188 self.repo = None |
259 self.repo = None |
|
260 self._open_access = set() |
189 super(CubicWebTC, self).__init__(*args, **kwargs) |
261 super(CubicWebTC, self).__init__(*args, **kwargs) |
190 |
262 |
191 # repository connection handling ########################################### |
263 # repository connection handling ########################################### |
|
264 def new_access(self, login): |
|
265 """provide a new RepoAccess object for a given user |
|
266 |
|
267 The access is automatically closed at the end of the test.""" |
|
268 access = RepoAccess(self.repo, login, self.requestcls) |
|
269 self._open_access.add(access) |
|
270 return access |
|
271 |
|
272 def _close_access(self): |
|
273 while self._open_access: |
|
274 self._open_access.pop().close() |
192 |
275 |
193 def set_cnx(self, cnx): |
276 def set_cnx(self, cnx): |
194 # XXX we want to deprecate this |
277 # XXX we want to deprecate this |
195 assert getattr(cnx, '_session', None) is not None |
278 assert getattr(cnx, '_session', None) is not None |
196 if cnx is self._admin_clt_cnx: |
279 if cnx is self._admin_clt_cnx: |