»Core Development>Code coverage>Lib/sqlite3/test/transactions.py

Python code coverage for Lib/sqlite3/test/transactions.py

1n/a#-*- coding: iso-8859-1 -*-
2n/a# pysqlite2/test/transactions.py: tests transactions
4n/a# Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de>
6n/a# This file is part of pysqlite.
8n/a# This software is provided 'as-is', without any express or implied
9n/a# warranty. In no event will the authors be held liable for any damages
10n/a# arising from the use of this software.
12n/a# Permission is granted to anyone to use this software for any purpose,
13n/a# including commercial applications, and to alter it and redistribute it
14n/a# freely, subject to the following restrictions:
16n/a# 1. The origin of this software must not be misrepresented; you must not
17n/a# claim that you wrote the original software. If you use this software
18n/a# in a product, an acknowledgment in the product documentation would be
19n/a# appreciated but is not required.
20n/a# 2. Altered source versions must be plainly marked as such, and must not be
21n/a# misrepresented as being the original software.
22n/a# 3. This notice may not be removed or altered from any source distribution.
24n/aimport os, unittest
25n/aimport sqlite3 as sqlite
27n/adef get_db_path():
28n/a return "sqlite_testdb"
30n/aclass TransactionTests(unittest.TestCase):
31n/a def setUp(self):
32n/a try:
33n/a os.remove(get_db_path())
34n/a except OSError:
35n/a pass
37n/a self.con1 = sqlite.connect(get_db_path(), timeout=0.1)
38n/a self.cur1 = self.con1.cursor()
40n/a self.con2 = sqlite.connect(get_db_path(), timeout=0.1)
41n/a self.cur2 = self.con2.cursor()
43n/a def tearDown(self):
44n/a self.cur1.close()
45n/a self.con1.close()
47n/a self.cur2.close()
48n/a self.con2.close()
50n/a try:
51n/a os.unlink(get_db_path())
52n/a except OSError:
53n/a pass
55n/a def CheckDMLDoesNotAutoCommitBefore(self):
56n/a self.cur1.execute("create table test(i)")
57n/a self.cur1.execute("insert into test(i) values (5)")
58n/a self.cur1.execute("create table test2(j)")
59n/a self.cur2.execute("select i from test")
60n/a res = self.cur2.fetchall()
61n/a self.assertEqual(len(res), 0)
63n/a def CheckInsertStartsTransaction(self):
64n/a self.cur1.execute("create table test(i)")
65n/a self.cur1.execute("insert into test(i) values (5)")
66n/a self.cur2.execute("select i from test")
67n/a res = self.cur2.fetchall()
68n/a self.assertEqual(len(res), 0)
70n/a def CheckUpdateStartsTransaction(self):
71n/a self.cur1.execute("create table test(i)")
72n/a self.cur1.execute("insert into test(i) values (5)")
73n/a self.con1.commit()
74n/a self.cur1.execute("update test set i=6")
75n/a self.cur2.execute("select i from test")
76n/a res = self.cur2.fetchone()[0]
77n/a self.assertEqual(res, 5)
79n/a def CheckDeleteStartsTransaction(self):
80n/a self.cur1.execute("create table test(i)")
81n/a self.cur1.execute("insert into test(i) values (5)")
82n/a self.con1.commit()
83n/a self.cur1.execute("delete from test")
84n/a self.cur2.execute("select i from test")
85n/a res = self.cur2.fetchall()
86n/a self.assertEqual(len(res), 1)
88n/a def CheckReplaceStartsTransaction(self):
89n/a self.cur1.execute("create table test(i)")
90n/a self.cur1.execute("insert into test(i) values (5)")
91n/a self.con1.commit()
92n/a self.cur1.execute("replace into test(i) values (6)")
93n/a self.cur2.execute("select i from test")
94n/a res = self.cur2.fetchall()
95n/a self.assertEqual(len(res), 1)
96n/a self.assertEqual(res[0][0], 5)
98n/a def CheckToggleAutoCommit(self):
99n/a self.cur1.execute("create table test(i)")
100n/a self.cur1.execute("insert into test(i) values (5)")
101n/a self.con1.isolation_level = None
102n/a self.assertEqual(self.con1.isolation_level, None)
103n/a self.cur2.execute("select i from test")
104n/a res = self.cur2.fetchall()
105n/a self.assertEqual(len(res), 1)
107n/a self.con1.isolation_level = "DEFERRED"
108n/a self.assertEqual(self.con1.isolation_level , "DEFERRED")
109n/a self.cur1.execute("insert into test(i) values (5)")
110n/a self.cur2.execute("select i from test")
111n/a res = self.cur2.fetchall()
112n/a self.assertEqual(len(res), 1)
114n/a @unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 2),
115n/a 'test hangs on sqlite versions older than 3.2.2')
116n/a def CheckRaiseTimeout(self):
117n/a self.cur1.execute("create table test(i)")
118n/a self.cur1.execute("insert into test(i) values (5)")
119n/a with self.assertRaises(sqlite.OperationalError):
120n/a self.cur2.execute("insert into test(i) values (5)")
122n/a @unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 2),
123n/a 'test hangs on sqlite versions older than 3.2.2')
124n/a def CheckLocking(self):
125n/a """
126n/a This tests the improved concurrency with pysqlite 2.3.4. You needed
127n/a to roll back con2 before you could commit con1.
128n/a """
129n/a self.cur1.execute("create table test(i)")
130n/a self.cur1.execute("insert into test(i) values (5)")
131n/a with self.assertRaises(sqlite.OperationalError):
132n/a self.cur2.execute("insert into test(i) values (5)")
133n/a # NO self.con2.rollback() HERE!!!
134n/a self.con1.commit()
136n/a def CheckRollbackCursorConsistency(self):
137n/a """
138n/a Checks if cursors on the connection are set into a "reset" state
139n/a when a rollback is done on the connection.
140n/a """
141n/a con = sqlite.connect(":memory:")
142n/a cur = con.cursor()
143n/a cur.execute("create table test(x)")
144n/a cur.execute("insert into test(x) values (5)")
145n/a cur.execute("select 1 union select 2 union select 3")
147n/a con.rollback()
148n/a with self.assertRaises(sqlite.InterfaceError):
149n/a cur.fetchall()
151n/aclass SpecialCommandTests(unittest.TestCase):
152n/a def setUp(self):
153n/a self.con = sqlite.connect(":memory:")
154n/a self.cur = self.con.cursor()
156n/a def CheckDropTable(self):
157n/a self.cur.execute("create table test(i)")
158n/a self.cur.execute("insert into test(i) values (5)")
159n/a self.cur.execute("drop table test")
161n/a def CheckPragma(self):
162n/a self.cur.execute("create table test(i)")
163n/a self.cur.execute("insert into test(i) values (5)")
164n/a self.cur.execute("pragma count_changes=1")
166n/a def tearDown(self):
167n/a self.cur.close()
168n/a self.con.close()
170n/aclass TransactionalDDL(unittest.TestCase):
171n/a def setUp(self):
172n/a self.con = sqlite.connect(":memory:")
174n/a def CheckDdlDoesNotAutostartTransaction(self):
175n/a # For backwards compatibility reasons, DDL statements should not
176n/a # implicitly start a transaction.
177n/a self.con.execute("create table test(i)")
178n/a self.con.rollback()
179n/a result = self.con.execute("select * from test").fetchall()
180n/a self.assertEqual(result, [])
182n/a def CheckTransactionalDDL(self):
183n/a # You can achieve transactional DDL by issuing a BEGIN
184n/a # statement manually.
185n/a self.con.execute("begin")
186n/a self.con.execute("create table test(i)")
187n/a self.con.rollback()
188n/a with self.assertRaises(sqlite.OperationalError):
189n/a self.con.execute("select * from test")
191n/a def tearDown(self):
192n/a self.con.close()
194n/adef suite():
195n/a default_suite = unittest.makeSuite(TransactionTests, "Check")
196n/a special_command_suite = unittest.makeSuite(SpecialCommandTests, "Check")
197n/a ddl_suite = unittest.makeSuite(TransactionalDDL, "Check")
198n/a return unittest.TestSuite((default_suite, special_command_suite, ddl_suite))
200n/adef test():
201n/a runner = unittest.TextTestRunner()
202n/a runner.run(suite())
204n/aif __name__ == "__main__":
205n/a test()