14 # FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more |
14 # FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more |
15 # details. |
15 # details. |
16 # |
16 # |
17 # You should have received a copy of the GNU Lesser General Public License along |
17 # You should have received a copy of the GNU Lesser General Public License along |
18 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
18 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
19 """unit tests for module cubicweb.server.repository |
19 """unit tests for module cubicweb.server.repository""" |
20 |
|
21 """ |
|
22 from __future__ import with_statement |
|
23 |
20 |
24 from __future__ import with_statement |
21 from __future__ import with_statement |
25 |
22 |
26 import os |
23 import os |
27 import sys |
24 import sys |
213 self.assertEquals(len(rset), 0) |
210 self.assertEquals(len(rset), 0) |
214 |
211 |
215 def test_transaction_interleaved(self): |
212 def test_transaction_interleaved(self): |
216 self.skip('implement me') |
213 self.skip('implement me') |
217 |
214 |
218 def test_close_wait_processing_request(self): |
215 def test_close_kill_processing_request(self): |
219 repo = self.repo |
216 repo = self.repo |
220 cnxid = repo.connect(self.admlogin, password=self.admpassword) |
217 cnxid = repo.connect(self.admlogin, password=self.admpassword) |
221 repo.execute(cnxid, 'INSERT CWUser X: X login "toto", X upassword "tutu", X in_group G WHERE G name "users"') |
218 repo.execute(cnxid, 'INSERT CWUser X: X login "toto", X upassword "tutu", X in_group G WHERE G name "users"') |
222 repo.commit(cnxid) |
219 repo.commit(cnxid) |
223 # close has to be in the thread due to sqlite limitations |
220 # close has to be in the thread due to sqlite limitations |
224 def close_in_a_few_moment(): |
221 def close_in_a_few_moment(): |
225 time.sleep(0.1) |
222 time.sleep(0.1) |
226 repo.close(cnxid) |
223 repo.close(cnxid) |
227 t = threading.Thread(target=close_in_a_few_moment) |
224 t = threading.Thread(target=close_in_a_few_moment) |
228 t.start() |
225 t.start() |
229 try: |
226 def run_transaction(): |
230 repo.execute(cnxid, 'DELETE CWUser X WHERE X login "toto"') |
227 repo.execute(cnxid, 'DELETE CWUser X WHERE X login "toto"') |
231 repo.commit(cnxid) |
228 repo.commit(cnxid) |
|
229 try: |
|
230 ex = self.assertRaises(Exception, run_transaction) |
|
231 self.assertEquals(str(ex), 'try to access pool on a closed session') |
232 finally: |
232 finally: |
233 t.join() |
233 t.join() |
234 |
234 |
235 def test_initial_schema(self): |
235 def test_initial_schema(self): |
236 schema = self.repo.schema |
236 schema = self.repo.schema |