Lines Matching refs:self
36 def setUp(self):
37 self.con1 = sqlite.connect(TESTFN, timeout=TIMEOUT)
38 self.cur1 = self.con1.cursor()
40 self.con2 = sqlite.connect(TESTFN, timeout=TIMEOUT)
41 self.cur2 = self.con2.cursor()
43 def tearDown(self):
45 self.cur1.close()
46 self.con1.close()
48 self.cur2.close()
49 self.con2.close()
54 def test_dml_does_not_auto_commit_before(self):
55 self.cur1.execute("create table test(i)")
56 self.cur1.execute("insert into test(i) values (5)")
57 self.cur1.execute("create table test2(j)")
58 self.cur2.execute("select i from test")
59 res = self.cur2.fetchall()
60 self.assertEqual(len(res), 0)
62 def test_insert_starts_transaction(self):
63 self.cur1.execute("create table test(i)")
64 self.cur1.execute("insert into test(i) values (5)")
65 self.cur2.execute("select i from test")
66 res = self.cur2.fetchall()
67 self.assertEqual(len(res), 0)
69 def test_update_starts_transaction(self):
70 self.cur1.execute("create table test(i)")
71 self.cur1.execute("insert into test(i) values (5)")
72 self.con1.commit()
73 self.cur1.execute("update test set i=6")
74 self.cur2.execute("select i from test")
75 res = self.cur2.fetchone()[0]
76 self.assertEqual(res, 5)
78 def test_delete_starts_transaction(self):
79 self.cur1.execute("create table test(i)")
80 self.cur1.execute("insert into test(i) values (5)")
81 self.con1.commit()
82 self.cur1.execute("delete from test")
83 self.cur2.execute("select i from test")
84 res = self.cur2.fetchall()
85 self.assertEqual(len(res), 1)
87 def test_replace_starts_transaction(self):
88 self.cur1.execute("create table test(i)")
89 self.cur1.execute("insert into test(i) values (5)")
90 self.con1.commit()
91 self.cur1.execute("replace into test(i) values (6)")
92 self.cur2.execute("select i from test")
93 res = self.cur2.fetchall()
94 self.assertEqual(len(res), 1)
95 self.assertEqual(res[0][0], 5)
97 def test_toggle_auto_commit(self):
98 self.cur1.execute("create table test(i)")
99 self.cur1.execute("insert into test(i) values (5)")
100 self.con1.isolation_level = None
101 self.assertEqual(self.con1.isolation_level, None)
102 self.cur2.execute("select i from test")
103 res = self.cur2.fetchall()
104 self.assertEqual(len(res), 1)
106 self.con1.isolation_level = "DEFERRED"
107 self.assertEqual(self.con1.isolation_level , "DEFERRED")
108 self.cur1.execute("insert into test(i) values (5)")
109 self.cur2.execute("select i from test")
110 res = self.cur2.fetchall()
111 self.assertEqual(len(res), 1)
113 def test_raise_timeout(self):
114 self.cur1.execute("create table test(i)")
115 self.cur1.execute("insert into test(i) values (5)")
116 with self.assertRaises(sqlite.OperationalError):
117 self.cur2.execute("insert into test(i) values (5)")
119 def test_locking(self):
124 self.cur1.execute("create table test(i)")
125 self.cur1.execute("insert into test(i) values (5)")
126 with self.assertRaises(sqlite.OperationalError):
127 self.cur2.execute("insert into test(i) values (5)")
128 # NO self.con2.rollback() HERE!!!
129 self.con1.commit()
131 def test_rollback_cursor_consistency(self):
140 self.assertEqual(cur.fetchall(), [(1,), (2,), (3,)])
142 def test_multiple_cursors_and_iternext(self):
153 self.con1.execute("create table t(t)")
154 sql(self.con1, "insert into t values (?), (?), (?)", "u1", "u2", "u3")
155 self.con1.commit()
158 count = sql(self.con2, "select count(*) from t").fetchone()[0]
159 self.assertEqual(count, 3)
160 changes = sql(self.con2, "delete from t").rowcount
161 self.assertEqual(changes, 3)
162 self.con2.commit()
165 sql(self.con1, "insert into t values (?)", "u4")
166 sql(self.con1, "insert into t values (?)", "u5")
169 count = sql(self.con2, "select count(*) from t").fetchone()[0]
170 self.assertEqual(count, 0)
173 count = sql(self.con1, "select count(*) from t").fetchone()[0]
174 self.assertEqual(count, 2)
177 self.con1.commit()
178 count = sql(self.con2, "select count(*) from t").fetchone()[0]
179 self.assertEqual(count, 2)
185 def setUp(self):
186 self.con = sqlite.connect(":memory:")
187 self.cur1 = self.con.cursor()
188 self.cur2 = self.con.cursor()
189 with self.con:
190 self.con.execute("create table t(c)");
191 self.con.executemany("insert into t values(?)", [(0,), (1,), (2,)])
192 self.cur1.execute("begin transaction")
194 self.cur1.execute(select)
195 self.con.rollback()
196 self.res = self.cur2.execute(select) # Reusing stmt from cache
198 def tearDown(self):
199 self.con.close()
201 def _check_rows(self):
202 for i, row in enumerate(self.res):
203 self.assertEqual(row[0], i)
205 def test_no_duplicate_rows_after_rollback_del_cursor(self):
206 del self.cur1
207 self._check_rows()
209 def test_no_duplicate_rows_after_rollback_close_cursor(self):
210 self.cur1.close()
211 self._check_rows()
213 def test_no_duplicate_rows_after_rollback_new_query(self):
214 self.cur1.execute("select c from t where c = 1")
215 self._check_rows()
220 def setUp(self):
221 self.con = sqlite.connect(":memory:")
222 self.cur = self.con.cursor()
224 def test_drop_table(self):
225 self.cur.execute("create table test(i)")
226 self.cur.execute("insert into test(i) values (5)")
227 self.cur.execute("drop table test")
229 def test_pragma(self):
230 self.cur.execute("create table test(i)")
231 self.cur.execute("insert into test(i) values (5)")
232 self.cur.execute("pragma count_changes=1")
234 def tearDown(self):
235 self.cur.close()
236 self.con.close()
240 def setUp(self):
241 self.con = sqlite.connect(":memory:")
243 def test_ddl_does_not_autostart_transaction(self):
246 self.con.execute("create table test(i)")
247 self.con.rollback()
248 result = self.con.execute("select * from test").fetchall()
249 self.assertEqual(result, [])
251 def test_immediate_transactional_ddl(self):
254 self.con.execute("begin immediate")
255 self.con.execute("create table test(i)")
256 self.con.rollback()
257 with self.assertRaises(sqlite.OperationalError):
258 self.con.execute("select * from test")
260 def test_transactional_ddl(self):
263 self.con.execute("begin")
264 self.con.execute("create table test(i)")
265 self.con.rollback()
266 with self.assertRaises(sqlite.OperationalError):
267 self.con.execute("select * from test")
269 def tearDown(self):
270 self.con.close()
277 def setUp(self):
278 self.traced = []
280 def _run_test(self, cx):
281 cx.execute(self.CREATE)
282 cx.set_trace_callback(lambda stmt: self.traced.append(stmt))
284 cx.execute(self.INSERT)
286 def test_isolation_level_default(self):
288 self._run_test(cx)
289 self.assertEqual(self.traced, ["BEGIN ", self.INSERT, "COMMIT"])
291 def test_isolation_level_begin(self):
293 self._run_test(cx)
294 self.assertEqual(self.traced, ["BEGIN ", self.INSERT, "COMMIT"])
296 def test_isolation_level_deferred(self):
298 self._run_test(cx)
299 self.assertEqual(self.traced, ["BEGIN DEFERRED", self.INSERT, "COMMIT"])
301 def test_isolation_level_immediate(self):
303 self._run_test(cx)
304 self.assertEqual(self.traced,
305 ["BEGIN IMMEDIATE", self.INSERT, "COMMIT"])
307 def test_isolation_level_exclusive(self):
309 self._run_test(cx)
310 self.assertEqual(self.traced,
311 ["BEGIN EXCLUSIVE", self.INSERT, "COMMIT"])
313 def test_isolation_level_none(self):
315 self._run_test(cx)
316 self.assertEqual(self.traced, [self.INSERT])
322 def setUp(self):
323 self.cx = sqlite.connect(":memory:")
324 self.cx.execute("create table t(t)")
325 self.traced = []
326 self.cx.set_trace_callback(lambda stmt: self.traced.append(stmt))
328 def tearDown(self):
329 self.cx.close()
331 def test_isolation_level_default(self):
332 with self.cx:
333 self.cx.execute(self.QUERY)
334 self.assertEqual(self.traced, ["BEGIN ", self.QUERY, "COMMIT"])
336 def test_isolation_level_begin(self):
337 self.cx.isolation_level = ""
338 with self.cx:
339 self.cx.execute(self.QUERY)
340 self.assertEqual(self.traced, ["BEGIN ", self.QUERY, "COMMIT"])
342 def test_isolation_level_deferrred(self):
343 self.cx.isolation_level = "DEFERRED"
344 with self.cx:
345 self.cx.execute(self.QUERY)
346 self.assertEqual(self.traced, ["BEGIN DEFERRED", self.QUERY, "COMMIT"])
348 def test_isolation_level_immediate(self):
349 self.cx.isolation_level = "IMMEDIATE"
350 with self.cx:
351 self.cx.execute(self.QUERY)
352 self.assertEqual(self.traced,
353 ["BEGIN IMMEDIATE", self.QUERY, "COMMIT"])
355 def test_isolation_level_exclusive(self):
356 self.cx.isolation_level = "EXCLUSIVE"
357 with self.cx:
358 self.cx.execute(self.QUERY)
359 self.assertEqual(self.traced,
360 ["BEGIN EXCLUSIVE", self.QUERY, "COMMIT"])
362 def test_isolation_level_none(self):
363 self.cx.isolation_level = None
364 with self.cx:
365 self.cx.execute(self.QUERY)
366 self.assertEqual(self.traced, [self.QUERY])