概要
タイトルの通りですが、アプリケーションを作っていると必ずテストがあるのですが、DBの登録内容確認とかが目視確認だと限界があるし何よりバグある度に何度もやり直したくないのでなんとか楽にならないかなと思ってPythonでアサーションチェックコードを書いてみました。
この記事では、Pythonを使用して特定の条件に一致するデータベースレコードをチェックする自動化サンプルプログラムを紹介します。CSVファイルから期待値を読み込み、チェック結果を出力します。データベースの品質保証やテスト効率化に役立つテクニックを1つとして誰かのお役に立てれば幸いです。目指せテスト自動化。
チェック要件
- 特定の条件に一致するDBレコードだけチェックしたい
- 条件に一致したレコードの内、特定の列の値だけチェックしたい
- チェックに一致しない行／列があってもチェックは最後までやり切りたい
サンプルプログラムでの対応内容
要件に対し以下の内容で実現しました。
- チェックの期待値を書くCSVファイルを用意して、CSVの先頭列とDBのレコードに一致した行だけチェックする仕様にしました
- CSVファイルの中身は、チェックしたい特定の列だけ記載して記載がある列だけチェックする仕様としました
- assert文だとNGが1つあった段階でエラーになってしまうので、大人しくif文でチェックしました
チェック用CSVファイルはこんな感じです。
"KEY","TYPE","STATUS"
"0001","1","1"
"1000","2","0"
先頭列の"KEY"に一致したレコードだけチェックします。KEY=0002等のデータもDBにもちろんあるのですが、チェック対象は、KEY=0001とKEY=1000になります。
DBの列は、KEY,TYPE,STATUSの他にも様々な列があるのですが、あくまでもチェック対象はKEY,TYPE,STATUSの3つになります。
動作環境
- Windows
- DB(Oracle 19c)
- Python 3.10.11
- Pythonライブラリ(oracledb 1.3.2/pandas 2.0.3)
事前準備
Pythonライブラリoracledbインストール
以下を参考にoracledbをインストール
python -m pip install oracledb --upgrade
oracledbの実際の使い方は以下が参考になります。
Pythonライブラリpandasインストール
以下コマンドでpandasをインストール
pip install pandas
pandasの説明は割愛します。。。
私は以下の本で勉強しました。
（私のPythonレベルはこの本の通り2年生程度です^^;）
サンプルプログラム
以下のサンプルプログラムでチェック可能です。
DB接続情報、レコード抽出SQL、レコード特定項目名、CSVファイル名は環境によって適切に書き換えてください。
import oracledb
import pandas as pd
# DB接続情報
DB_HOST = 'hostname'
DB_PORT = '1521'
DB_SID = 'sid'
DB_USER = 'user'
DB_PASS = 'pass'
# レコード抽出SQL
SQL = 'SELECT * FROM USERTABLE'
# レコード特定項目名
DB_KEY = 'KEY'
# CSVファイル名
CSV_FILE = 'check.csv'
# DBレコード取得
def select_db():
connection = oracledb.connect(user=DB_USER, password=DB_PASS, dsn=f"{DB_HOST}:{DB_PORT}/{DB_SID}")
cursor = connection.cursor()
cursor.execute(SQL)
columns = [col[0] for col in cursor.description] # 列名を取得
data = cursor.fetchall() # データを取得
# クエリ結果をデータフレームに格納
df = pd.DataFrame(data, columns=columns)
cursor.close()
connection.close()
return df
# 値確認
def check_assert(df_exp, df_act):
result_flag = True
# ヘッダ行の列名を取得
df_exp_header = list(df_exp.columns)
# 条件に一致する行ごとにアサーションを行う
for df_exp_row_index, df_exp_row in df_exp.iterrows():
filtered_df_act = df_act[df_act[DB_KEY] == df_exp_row[DB_KEY]] # KEY列が一致する行のチェックを行う
for filtered_df_act_index, filtered_df_act_row in filtered_df_act.iterrows():
if filtered_df_act.empty:
result_flag = False
print("NG:NoRecord " + str(df_exp_row[DB_KEY]))
else:
for column in df_exp_header:
if df_exp_row[column] == filtered_df_act_row[column]:
print("OK:" + str(df_exp_row_index + 1))
else:
result_flag = False
print("NG:" + str(df_exp_row_index + 1) + " exp:" + str(df_exp_row[column]) + " act:" + str(filtered_df_act_row[column]))
return result_flag
def main():
csv_path = CSV_FILE
df_expect = pd.read_csv(csv_path, header=0)
df_expect[DB_KEY] = df_expect[DB_KEY].astype(str) # 列'KEY'のデータ型を文字列型に変更
# 実際の値を取得する
df_actual = select_db()
if check_assert(df_expect, df_actual):
print("CHECK SUCCESS END.")
else:
print("CHECK ERROR END.")
if __name__ == '__main__':
main()
注意事項
SQLは全件検索になっているためレコードがたくさんある場合は、SQL条件を見直しましょう。この条件もCSVファイルインプットにするとさらに柔軟になると思います。
# レコード抽出SQL
SQL = 'SELECT * FROM USERTABLE'
fetch件数も現在allになっているので、レコード数に応じてメンテしましょう。
data = cursor.fetchall() # データを取得
pandasのデータフレームはデフォルトでは各列のデータ型が自動的に推測されます。
そのため、チェック対象のCSVファイルは各列のデータ型が自動的に推測されます。
一方、OracleのデータフレームはDBの型に依存して正しくDB型通りになるようでした。
型がずれていると正しく一致チェックが動作しないため、型が正しく認識されていない場合は、意識的に正しく指定しましょう。
サンプルは以下の1文で'KEY'のデータ型を文字列型に変更しています。
df_expect[DB_KEY] = df_expect[DB_KEY].astype(str) # 列'KEY'のデータ型を文字列型に変更
ちなみにpandasのデータフレームの各列の型は以下で確認できます。
print(df.dtypes)
まとめ
importや定数定義部分を除く、処理だけだとソースコードはおよそ50行程度。
CSVファイル読み込み、DB接続して、項目チェックしているので、それなりに処理はあるのですが、こんな少なく書けるのはさすがPythonという気がします。
個人的には項目チェックを1行 x 1列泥臭くやっているのでもっと簡単に出来ないものかと思いましたが、私の力量ではこの程度でした(^^;)
良いアドバイスある方いたらコメントください！
おまけ
同じ要件をChatGPTに依頼してみた結果です。
こちらの動作確認はしていません。ただ、これが一瞬で出てくることが凄いですね。。。
依頼文
あなたはアプリケーション上級開発者です。アプリケーションのテスト自動化のためにPythonでDBの登録内容を自動的にチェックするプログラムを作りたいのですが、Pythonでどんなプログラムができますか？具体的なサンプルプログラムを回答してください。
要件は以下の通りです。
- 特定の条件に一致するDBレコードだけチェックしたい
- 条件に一致したレコードの内、特定の列の値だけチェックしたい
- チェックの行や列の期待値はCSVファイルに定義する
- チェックに一致しない行／列があってもチェックは最後までやり切りたい
回答文
以下に、具体的なソースコードのサンプルを示します。このサンプルでは、Oracleデータベースを使用し、特定の条件に一致するレコードの特定の列の値をチェックするプログラムです。CSVファイルから期待値を読み込み、チェック結果をログファイルに出力します。
import csv
import cx_Oracle
# CSVファイルから期待値を読み込む
def read_expectations(file_path):
expectations = {}
with open(file_path, 'r') as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
column = row['Column']
expected_value = row['Expected Value']
expectations[column] = expected_value
return expectations
# データベースへの接続とクエリ実行
def execute_query(host, port, sid, username, password, query):
dsn = cx_Oracle.makedsn(host, port, sid)
connection = cx_Oracle.connect(username, password, dsn)
cursor = connection.cursor()
cursor.execute(query)
records = cursor.fetchall()
cursor.close()
connection.close()
return records
# データのチェックとログ出力
def check_data(records, expectations, log_file):
with open(log_file, 'w') as logfile:
for record in records:
column_value = record[0] # チェックする列のインデックスを指定
expected_value = expectations.get('Column', '')
if column_value == expected_value:
logfile.write(f"OK - {column_value}\n")
else:
logfile.write(f"ERROR - Expected: {expected_value}, Actual: {column_value}\n")
# メイン関数
def main():
# CSVから期待値を読み込む
expectations = read_expectations('expectations.csv')
# データベースへの接続とクエリ実行
host = 'localhost'
port = 1521
sid = 'ORCL'
username = 'username'
password = 'password'
query = 'SELECT column FROM table WHERE condition = value'
records = execute_query(host, port, sid, username, password, query)
# データのチェックとログ出力
log_file = 'test_log.txt'
check_data(records, expectations, log_file)
if __name__ == '__main__':
main()
上記のサンプルコードは、基本的なフレームワークとして使用できますが、実際の使用環境に合わせてカスタマイズする必要があります。データベースの接続情報やクエリ、チェックする列などを適切に設定し、期待値のCSVファイルを作成してください。また、必要に応じてエラーハンドリングやログの詳細な出力などを追加することもできます。
以上です！！