データベース クラスをテストしようとしています。これはその簡単な例です。
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
問題は、テスト メソッドから呼び出されたときのトレースバックで「データベースがロックされている」ことです。次のステップとしてコードが表示されます。
- 最初の接続を開く
- データを挿入する
- コミットして接続を閉じる
- 2 番目の接続を開く (最初の接続を閉じた後)
- select を使用してステップ 2 で挿入されたデータを取得する
- データを比較する
- 入力データと選択データが等しくない場合にアサートします。
しかし...接続がデータベースと1つずつ動作する場合、データベースのブロックに関するメッセージを取得する必要はありませんよね? ライブラリ(ユニットテストまたは仮説)がスレッドを使用していると思いましたが、ドキュメントには何も見つかりませんでした。
また、通常どおり実行して、for
列挙可能なデータを挿入しようとしました。それは正常に動作します。
私が間違っていcommit
なければ、接続が開かれていても、メソッドのすべての呼び出しでデータベースのブロックを解除する必要がありますが、そうではないようです。
「データベースがロックされています」というメッセージが表示される理由を理解できる人はいますか?