124 lines
6.6 KiB
Python
124 lines
6.6 KiB
Python
|
|
|
|
import csv
|
|
import decimal
|
|
from datetime import datetime
|
|
|
|
|
|
def create_vjsk_assertion_list(file_path: str) -> list:
|
|
"""DB登録期待値リストを作成する
|
|
|
|
Args:
|
|
file_path (str): DB登録期待値ファイル(tsvファイル)のパス
|
|
※DB登録期待値ファイルの前提
|
|
受領データファイルと同じ
|
|
BOM付きtsv形式
|
|
一行目はカラム名になっているヘッダ行
|
|
|
|
Returns:
|
|
List(dict) DB登録期待値辞書リスト
|
|
"""
|
|
with open(file_path, encoding='utf_8_sig', newline='') as tsv_file:
|
|
header = tsv_file.readline().strip('\n').replace('"', '').split('\t')
|
|
reader = csv.DictReader(tsv_file, fieldnames=header, delimiter='\t')
|
|
rows = [r for r in reader]
|
|
|
|
# DB抽出値と比較できるように、リテラル値をDB抽出値と同じデータフォーマットに変換
|
|
for row in rows:
|
|
for k, v in row.items():
|
|
converted_value = v
|
|
if v == 'NULL':
|
|
converted_value = None
|
|
if is_valid_date_format(v, '%Y/%m/%d') is True: # YYYY/MM/DD
|
|
converted_value = datetime.strptime(v, '%Y/%m/%d').date()
|
|
if is_valid_date_format(v, '%Y-%m-%d') is True: # YYYY-MM-DD
|
|
converted_value = datetime.strptime(v, '%Y-%m-%d').date()
|
|
if is_valid_date_format(v, '%Y/%m/%d %H:%M:%S') is True: # YYYY/MM/DD HH:MM:SS
|
|
converted_value = datetime.strptime(v, '%Y/%m/%d %H:%M:%S')
|
|
if is_valid_date_format(v, '%Y-%m-%d %H:%M:%S') is True: # YYYY-MM-DD HH:MM:SS
|
|
converted_value = datetime.strptime(v, '%Y-%m-%d %H:%M:%S')
|
|
if is_valid_date_format(v, '%y-%m-%d %H:%M:%S') is True: # YY-MM-DD HH:MM:SS
|
|
converted_value = datetime.strptime(v, '%y-%m-%d %H:%M:%S')
|
|
|
|
row[k] = converted_value
|
|
|
|
return rows
|
|
|
|
|
|
def is_valid_date_format(date_str: str, date_format):
|
|
"""日付文字列が、与えられたフォーマットにマッチするかを検査する
|
|
|
|
Args:
|
|
date_str (str): 日付文字列
|
|
date_format (str, optional): 日付のフォーマット
|
|
|
|
Returns:
|
|
_type_: 正しい日付文字列の場合、True、それ以外はFalse
|
|
"""
|
|
try:
|
|
datetime.strptime(date_str, date_format)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def assert_table_results(actual_rows: list[dict], expect_rows: list[dict], ignore_col_names: list = None, force_cast_to_str_columns: list = [], is_loaddata=False) -> None:
|
|
"""テーブル同士の取得結果突き合わせ
|
|
|
|
Args:
|
|
actual_rows (list[dict]): テスト結果の辞書リスト
|
|
expect_rows (list[dict]): 期待値の辞書リスト
|
|
ignore_col_name (list): 比較を無視するDBのカラム名. Default None.
|
|
is_loaddata (bool): LOAD DATAで読み込むテーブルかどうかのフラグ。org05の比較をするときにONにする. Default False.
|
|
"""
|
|
# 取得件数が一致すること
|
|
assert len(actual_rows) == len(expect_rows), f'レコード件数が一致しません。DBレコード数 : {len(actual_rows)} 期待値 : {len(expect_rows)}'
|
|
|
|
line_number = 0
|
|
# 1行ずつ調査
|
|
for actual_row, expect_row in zip(actual_rows, expect_rows):
|
|
line_number += 1
|
|
# 1カラムずつ調査
|
|
for actual_col_name, expect_col_name in zip(actual_row, expect_row):
|
|
# テストメソッド側で個別に確認するものはスキップさせる
|
|
if ignore_col_names is not None and actual_col_name in ignore_col_names:
|
|
continue
|
|
else:
|
|
actual_value = actual_row[actual_col_name]
|
|
expect_value = expect_row[expect_col_name]
|
|
|
|
# 期待値を、DBのデータ型(リフレクションされたpythonのデータ型)にキャストする
|
|
if actual_col_name in force_cast_to_str_columns:
|
|
# DB項目(varchar)に日付型としてキャスト可能な値が期待値である場合、force_cast_to_str_columnsに基づいて強制的に文字列キャストする
|
|
if type(expect_value).__name__ == 'date':
|
|
expect_value = expect_value.strftime('%Y-%m-%d')
|
|
elif isinstance(actual_value, (int)):
|
|
# DB項目(int)の場合、期待値もintにキャストする
|
|
expect_value = int(expect_value)
|
|
elif isinstance(actual_value, (float)):
|
|
# DB項目(float)の場合、期待値もfloatにキャストする
|
|
expect_value = float(expect_value)
|
|
elif isinstance(actual_value, (decimal.Decimal)):
|
|
# DB項目(decimal)の場合、期待値もdecimalにキャストする
|
|
expect_value = decimal.Decimal(expect_value)
|
|
elif type(actual_value).__name__ == "date":
|
|
# DB項目(date)の場合、期待値("YYYYMMDD")もdateにキャストする
|
|
if is_valid_date_format(expect_value, '%Y%m%d') is True: # YYYYMMDD
|
|
expect_value = datetime.strptime(expect_value, '%Y%m%d').date()
|
|
elif actual_value is None and expect_value == "":
|
|
# DB項目値がNULLの場合、期待値が""であればNoneに置換する
|
|
expect_value = None
|
|
elif actual_value == "0000-00-00" and expect_value == "":
|
|
# DB項目(date)がゼロ日付(NULL代替値)の場合、期待値が""であれば"0000-00-00"に置換する
|
|
expect_value = "0000-00-00"
|
|
elif actual_value == "0000-00-00 00:00:00" and expect_value == "":
|
|
# DB項目(datetime)がゼロ日付(NULL代替値)の場合、期待値が""であれば"0000-00-00 00:00:00"に置換する
|
|
expect_value = "0000-00-00 00:00:00"
|
|
elif actual_value == "" and expect_value == "" and not is_loaddata:
|
|
# DB項目値が空文字のかつ期待値が""のパターンは想定外のパターンであるため、期待値をNoneに置換して失敗させる
|
|
# LOAD文では空文字で登録されるので除外する
|
|
expect_value = None
|
|
|
|
# 検証
|
|
assert actual_value == expect_value, f'{line_number}行目:"{actual_col_name}" : "{actual_value}" ({type(actual_value)})が、期待値 "{expect_value}" ({type(expect_value)}) と一致しませんでした'
|