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