hypothesis+unittest 测试锁定sqlite数据库
hypothesis+unittest tests lock sqlite database
我正在尝试测试我的数据库 class。这是它的简化示例。
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为:
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
时,问题是 "database is locked" 在回溯中。我将代码视为后续步骤:
- 打开第一个连接
- 插入数据
- 提交并关闭连接
- 打开第二个连接(在第一个连接关闭后)
- 使用 select
在步骤 2 中插入数据
- 比较数据
- 如果输入和 selected 数据不相等则断言。
但是...如果连接一个接一个地与数据库一起工作,我不会收到有关数据库阻塞的消息,对吗?我知道库(单元测试或假设)使用线程,但我在文档中找不到任何内容。
我还尝试 运行 通常 for
并插入可枚举数据。它工作正常。
如果我没记错的话,即使连接打开,每次调用 commit
方法都必须解锁数据库,但似乎并没有发生。
谁能帮我理解为什么我会看到 "database is locked" 消息?
我怀疑您的数据库连接实际上并未关闭。无论如何,你不应该在测试运行之间使用相同的数据库文件 - 假设测试是可重复的很重要 - 所以最简单的事情是在你的测试中创建一个临时文件并使用它而不是固定的 test.db
看看问题是否消失。
更一般地说,我认为依赖 del
中关闭的东西往往是奇怪错误的来源,我鼓励使用明确的 context manager 或类似的。
db
对象实际上 运行 __del__
直到它被垃圾收集,您不应该依赖于在任何特定时间发生。正如@DRMacIver 建议的那样,最好为此使用上下文管理器。
您可以通过在 del db
之后的行中添加 import gc; gc.collect()
来检查这是否是真正的问题。
我正在尝试测试我的数据库 class。这是它的简化示例。
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为:
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
时,问题是 "database is locked" 在回溯中。我将代码视为后续步骤:
- 打开第一个连接
- 插入数据
- 提交并关闭连接
- 打开第二个连接(在第一个连接关闭后)
- 使用 select 在步骤 2 中插入数据
- 比较数据
- 如果输入和 selected 数据不相等则断言。
但是...如果连接一个接一个地与数据库一起工作,我不会收到有关数据库阻塞的消息,对吗?我知道库(单元测试或假设)使用线程,但我在文档中找不到任何内容。
我还尝试 运行 通常 for
并插入可枚举数据。它工作正常。
如果我没记错的话,即使连接打开,每次调用 commit
方法都必须解锁数据库,但似乎并没有发生。
谁能帮我理解为什么我会看到 "database is locked" 消息?
我怀疑您的数据库连接实际上并未关闭。无论如何,你不应该在测试运行之间使用相同的数据库文件 - 假设测试是可重复的很重要 - 所以最简单的事情是在你的测试中创建一个临时文件并使用它而不是固定的 test.db
看看问题是否消失。
更一般地说,我认为依赖 del
中关闭的东西往往是奇怪错误的来源,我鼓励使用明确的 context manager 或类似的。
db
对象实际上 运行 __del__
直到它被垃圾收集,您不应该依赖于在任何特定时间发生。正如@DRMacIver 建议的那样,最好为此使用上下文管理器。
您可以通过在 del db
之后的行中添加 import gc; gc.collect()
来检查这是否是真正的问题。