← Back to branch summary

~ubuntu-branches/ubuntu/trusty/python3.4/trusty-proposed

« back to all changes in this revision

Viewing changes to Lib/sqlite3/test/transactions.py

  • Committer: Package Import Robot
  • Author(s): Matthias Klose
  • Date: 2013-11-25 09:44:27 UTC
  • Revision ID: package-import@ubuntu.com-20131125094427-lzxj8ap5w01lmo7f
Tags: upstream-3.4~b1
Import upstream version 3.4~b1

Show diffs side-by-side

added added

removed removed

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