我正在尝试测试我的数据库类。这是它的简化示例。
class Database:
""" it has more methods but I show only the most important """
def __init__(self, name):
# let's think the db-file exists with tables
self.conn = sqlite3.connect(name)
self.cursor = self.conn.cursor()
def __del__(self):
""" Here I close connection if the object was destroyed """
self.conn.close()
def insert(self, col1, col2, col3):
""" The key method where problem is """
self.cursor.execute(insert_query.format(col1, col2, col3))
self.conn.commit() # here I do commit to apply changes with DB
所以,我想检查insert
方法。测试用例类是:
class DatabaseTestCase(unittest.TestCase):
""" it has other methods but the problem is here """
@given(col1=text(col1_params), col2=text(col2_params), col3=text(col3_params))
def test_db_insert(self, col1, col2, col3):
db = Database("test.db")
input_data = col1, col2, col3
# insert with commit (see Database example above)
db.insert(*input_data)
# delete object and close connection
del db
# recreate the object to get sure my data was added and
# the changes were commited
db = Database("test.db")
# I use the way not to use my own methods of Database object
cursor = db.conn.execute("SELECT * FROM mytable WHERE col1 = '{}'".format(col1))
result = cursor.fetchone()
for input_item, row_item in zip(input_data, result):
pass # assert here
# close connection with deleting of the db object
del db
db.insert
问题是从测试方法调用时回溯中的“数据库被锁定” 。我将代码视为下一步:
- 打开第一个连接
- 插入数据
- 提交并关闭连接
- 打开第二个连接(在第一个被关闭之后)
- 使用 select 获取在步骤 2 中插入的数据
- 比较数据
- 如果输入和选择的数据不相等,则断言。
但是...如果连接一个一个地与数据库一起工作,我就不会收到有关数据库阻塞的消息,是吗?我知道库(单元测试或假设)使用线程,但我在文档中一无所获。
我也尝试照常运行它for
并插入可枚举的数据。它工作正常。
如果我没有错,commit
即使打开连接,每次调用方法都必须解除对数据库的阻塞,但似乎没有发生。
谁能帮我理解为什么我看到“数据库已锁定”消息?