|
1 from cubicweb import ConnectionError |
|
2 from cubicweb.dbapi import ProgrammingError |
|
3 from cubicweb.devtools.apptest import EnvBasedTC |
|
4 |
|
5 |
|
6 class DBAPITC(EnvBasedTC): |
|
7 @property |
|
8 def cnx(self): |
|
9 return self.login('anon') |
|
10 |
|
11 def test_public_repo_api(self): |
|
12 cnx = self.cnx |
|
13 self.assertEquals(cnx.get_schema(), self.env.repo.schema) |
|
14 self.assertEquals(cnx.source_defs(), {'system': {'adapter': 'native', 'uri': 'system'}}) |
|
15 self.restore_connection() # proper way to close cnx |
|
16 self.assertRaises(ProgrammingError, cnx.get_schema) |
|
17 self.assertRaises(ProgrammingError, cnx.source_defs) |
|
18 |
|
19 def test_db_api(self): |
|
20 cnx = self.cnx |
|
21 self.assertEquals(cnx.rollback(), None) |
|
22 self.assertEquals(cnx.commit(), None) |
|
23 self.restore_connection() # proper way to close cnx |
|
24 #self.assertEquals(cnx.close(), None) |
|
25 self.assertRaises(ProgrammingError, cnx.rollback) |
|
26 self.assertRaises(ProgrammingError, cnx.commit) |
|
27 self.assertRaises(ProgrammingError, cnx.close) |
|
28 |
|
29 def test_api(self): |
|
30 cnx = self.cnx |
|
31 self.assertEquals(cnx.user(None).login, 'anon') |
|
32 self.assertEquals(cnx.describe(1), (u'EGroup', u'system', None)) |
|
33 self.restore_connection() # proper way to close cnx |
|
34 self.assertRaises(ConnectionError, cnx.user, None) |
|
35 self.assertRaises(ConnectionError, cnx.describe, 1) |
|
36 |
|
37 def test_session_data_api(self): |
|
38 cnx = self.cnx |
|
39 self.assertEquals(cnx.get_session_data('data'), None) |
|
40 self.assertEquals(cnx.session_data(), {}) |
|
41 cnx.set_session_data('data', 4) |
|
42 self.assertEquals(cnx.get_session_data('data'), 4) |
|
43 self.assertEquals(cnx.session_data(), {'data': 4}) |
|
44 cnx.del_session_data('data') |
|
45 cnx.del_session_data('whatever') |
|
46 self.assertEquals(cnx.get_session_data('data'), None) |
|
47 self.assertEquals(cnx.session_data(), {}) |
|
48 cnx.session_data()['data'] = 4 |
|
49 self.assertEquals(cnx.get_session_data('data'), 4) |
|
50 self.assertEquals(cnx.session_data(), {'data': 4}) |
|
51 |
|
52 def test_shared_data_api(self): |
|
53 cnx = self.cnx |
|
54 self.assertEquals(cnx.get_shared_data('data'), None) |
|
55 cnx.set_shared_data('data', 4) |
|
56 self.assertEquals(cnx.get_shared_data('data'), 4) |
|
57 cnx.get_shared_data('data', pop=True) |
|
58 cnx.get_shared_data('whatever', pop=True) |
|
59 self.assertEquals(cnx.get_shared_data('data'), None) |
|
60 cnx.set_shared_data('data', 4) |
|
61 self.assertEquals(cnx.get_shared_data('data'), 4) |
|
62 self.restore_connection() # proper way to close cnx |
|
63 self.assertRaises(ConnectionError, cnx.check) |
|
64 self.assertRaises(ConnectionError, cnx.set_shared_data, 'data', 0) |
|
65 self.assertRaises(ConnectionError, cnx.get_shared_data, 'data') |
|
66 |
|
67 |
|
68 # class DBAPICursorTC(EnvBasedTC): |
|
69 |
|
70 # @property |
|
71 # def cursor(self): |
|
72 # return self.env.cnx.cursor() |
|
73 |
|
74 # def test_api(self): |
|
75 # cu = self.cursor |
|
76 # self.assertEquals(cu.describe(1), (u'EGroup', u'system', None)) |
|
77 # #cu.close() |
|
78 # #self.assertRaises(ConnectionError, cu.describe, 1) |
|
79 |
|
80 if __name__ == '__main__': |
|
81 from logilab.common.testlib import unittest_main |
|
82 unittest_main() |