140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
"""テスト用共通処理関数"""
|
|
|
|
import io
|
|
from datetime import datetime
|
|
from types import ModuleType
|
|
|
|
from src.batch.ultmarc.datfile import DatFile
|
|
|
|
|
|
def create_ultmarc_test_csv(*csv_rows: str) -> DatFile:
|
|
"""アルトマーク取込テストのCSVを作成
|
|
Args:
|
|
csv_rows (tuple[str]): CSV文字列のリスト
|
|
Returns:
|
|
list[list[str]]: CSVデータ
|
|
"""
|
|
string_io = io.StringIO()
|
|
string_io.writelines(csv_rows)
|
|
string_io.seek(0)
|
|
dat_file = DatFile(string_io)
|
|
|
|
return dat_file
|
|
|
|
|
|
def create_ultmarc_common_column_names() -> list[str]:
|
|
"""アルトマークテーブル共通のカラム名を作成
|
|
Returns:
|
|
list[str]: 共通カラム名
|
|
"""
|
|
|
|
return [
|
|
'regist_ymd',
|
|
'update_ymd',
|
|
'delete_ymd',
|
|
'regist_date',
|
|
'create_user',
|
|
'update_date',
|
|
'update_user',
|
|
'sys_regist_date',
|
|
'regist_prgm_id',
|
|
'sys_update_date',
|
|
'update_prgm_id'
|
|
]
|
|
|
|
|
|
def create_insert_sql_with_parameter(table_name: str, column_names: list[str], test_data: list[str]) -> tuple[str, dict]:
|
|
"""INSERT文と登録値のパラメータを返す
|
|
|
|
Args:
|
|
table_name (str): スキーマ完全修飾のテーブル名(例:src05.com_alma)
|
|
column_names (list[str]): カラム名のリスト
|
|
test_data (list[str]): 値のリスト
|
|
|
|
Returns:
|
|
tuple[str, dict]: [0]→INSERT文,[1]→値のパラメータ
|
|
"""
|
|
placeholders = ','.join([f':{column_name}' for column_name in column_names])
|
|
insert_sql = f"INSERT INTO {table_name} ({','.join(column_names)}) VALUES({placeholders})"
|
|
parameter = {k: v for k, v in zip(column_names, test_data)}
|
|
|
|
return insert_sql, parameter
|
|
|
|
|
|
def create_delete_sql_with_parameter(table_name: str, delete_parameter: dict[str, str]):
|
|
"""DELETE文と削除条件値のパラメータを返す
|
|
|
|
Args:
|
|
table_name (str): スキーマ完全修飾のテーブル名(例:src05.com_alma)
|
|
delete_parameter (dict[str, str]): 削除条件に使用するカラム名と値の辞書
|
|
|
|
Returns:
|
|
tuple[str, dict]: [0]→DELETE文,[1]→値のパラメータ
|
|
"""
|
|
where_clause_list = []
|
|
for k in delete_parameter:
|
|
where_clause_list.append(f'{k} = :{k}')
|
|
where_clauses = ' AND '.join(where_clause_list)
|
|
delete_sql = f"DELETE FROM {table_name} WHERE {where_clauses}"
|
|
|
|
return delete_sql, delete_parameter
|
|
|
|
|
|
def create_ultmarc_common_column_values(expect_datetime: datetime, expect_date_str: str, module_name: str) -> list:
|
|
"""アルトマークテーブル共通のカラムを作成
|
|
|
|
Args:
|
|
expect_datetime (datetime): テスト実行年月日時分秒
|
|
expect_date_str (str): テスト実行年月日の文字列
|
|
module_name (str): モジュール名
|
|
|
|
Returns:
|
|
list: 共通カラム値
|
|
"""
|
|
|
|
return [
|
|
expect_date_str, # 登録年月日(regist_ymd)
|
|
None, # 更新年月日(update_ymd)
|
|
None, # 削除年月日(delete_ymd)
|
|
None, # 登録日時(regist_date)
|
|
None, # 登録者(create_user)
|
|
None, # 更新日時(update_date)
|
|
None, # 更新者(update_user)
|
|
expect_datetime, # システム登録日時(sys_regist_date)
|
|
module_name, # 登録プログラムid(regist_prgm_id)
|
|
expect_datetime, # システム更新日時(sys_update_date)
|
|
module_name # 更新プログラムid(update_prgm_id)
|
|
]
|
|
|
|
|
|
def get_module_name(module: ModuleType) -> str:
|
|
"""登録プログラムID、更新プログラムIDに登録するモジュール名を作成
|
|
|
|
Args:
|
|
module (ModuleType): pythonモジュール
|
|
|
|
Returns:
|
|
str: モジュール名
|
|
"""
|
|
return module.__name__.split('.')[-1]
|
|
|
|
|
|
def assert_table_results(actual_rows: list[dict], expect_rows: list[dict]) -> None:
|
|
"""テーブル同士の取得結果突き合わせ
|
|
|
|
Args:
|
|
actual_rows (list[dict]): テスト結果の辞書リスト
|
|
expect_rows (list[dict]): 期待値の辞書リスト
|
|
"""
|
|
# 取得件数が一致すること
|
|
assert len(actual_rows) == len(expect_rows)
|
|
|
|
# 1カラムずつ調査
|
|
for actual_row, expect_row in zip(actual_rows, expect_rows):
|
|
for actual_col_name, expect_col_name in zip(actual_row, expect_row):
|
|
# システム日付が設定されるカラムは、期待値以前になっていること
|
|
if actual_col_name in ['sys_regist_date', 'sys_update_date']:
|
|
assert actual_row[actual_col_name] < expect_row[expect_col_name], f'{actual_col_name}が、期待値の日時以前であること'
|
|
else:
|
|
assert actual_row[actual_col_name] == expect_row[expect_col_name], f'{actual_col_name}が、期待値と一致すること'
|