Python sqlite3ºÍµ¥Ôª²âÊÔ

 
import os
import unittest # °üº¬µ¥Ôª²âÊÔÄ£¿é
import sqlite3 as sqlite # °üº¬sqlite3Ä£¿é
def get_db_path():
return "sqlite_testdb"
class TransactionTests(unittest.TestCase): # µ¥Ôª²âÊÔµÚÒ»²½:¡¡ÓÉTestCaseÅÉÉúÀà
def setUp(self): # µ¥Ôª²âÊÔ»·¾³ÅäÖÃ
try:
os.remove(get_db_path())
except:
pass
self.con1 = sqlite.connect(get_db_path(), timeout=0.1) # Á¬½ÓÊý¾Ý¿â
self.cur1 = self.con1.cursor() # »ñÈ¡Óαê
self.con2 = sqlite.connect(get_db_path(), timeout=0.1)
self.cur2 = self.con2.cursor()
def tearDown(self): # µ¥Ôª²âÊÔ»·¾³Çå³ý
self.cur1.close() # ¹Ø±ÕÓαê
self.con1.close() # ¹Ø±ÕÁ¬½Ó
self.cur2.close()
self.con2.close()
os.unlink(get_db_path())
def CheckDMLdoesAutoCommitBefore(self):
self.cur1.execute("create table test(i)") # Ö´ÐÐSQL²éѯ
self.cur1.execute("insert into test(i) values (5)")
self.cur1.execute("create table test2(j)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
self.failUnlessEqual(len(res), 1) # ²âÊÔ
def CheckInsertStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
self.failUnlessEqual(len(res), 0)
def CheckUpdateStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.con1.commit()
self.cur1.execute("update test set i=6")
self.cur2.execute("select i from test")
res = self.cur2.fetchone()[0]
self.failUnlessEqual(res, 5)
def CheckDeleteStartsTransaction(self):
self.cur1.exec