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 = None, is_loaddata=False) -> None: """テーブル同士の取得結果突き合わせ Args: actual_rows (list[dict]): テスト結果の辞書リスト expect_rows (list[dict]): 期待値の辞書リスト ignore_col_name (list): 比較を無視するDBのカラム名. Default None. force_cast_to_str_columns (list): 強制的に文字列にキャストする項目のリスト 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 force_cast_to_str_columns is not None and 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 type(expect_value).__name__ == 'datetime': expect_value = expect_value.strftime('%Y-%m-%d %H:%M:%S') elif not is_loaddata and expect_value == "": expect_value = None 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)}) と一致しませんでした'