177 lines
6.1 KiB
Python
177 lines
6.1 KiB
Python
"""テスト用共通処理関数"""
|
|
|
|
import io
|
|
from types import ModuleType
|
|
|
|
from src.batch.ultmarc.datfile import DatFile, DatFileLine
|
|
from src.batch.ultmarc.utmp_tables.table_mapper.ultmarc_table_mapper import \
|
|
UltmarcTableMapper
|
|
from src.batch.ultmarc.utmp_tables.ultmarc_table_mapper_factory import \
|
|
UltmarcTableMapperFactory
|
|
from src.db.database import Database
|
|
|
|
|
|
def create_ultmarc_test_csv(csv_rows: list[str]) -> DatFile:
|
|
"""アルトマーク取込テストのCSVを作成
|
|
Args:
|
|
csv_rows (tuple[str]): CSV文字列のリスト
|
|
Returns:
|
|
list[list[str]]: CSVデータ
|
|
"""
|
|
string_io = io.StringIO()
|
|
for csv_row in csv_rows:
|
|
string_io.write(csv_row)
|
|
string_io.write('\n')
|
|
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(**kwargs) -> list:
|
|
"""アルトマークテーブル共通のカラムを作成
|
|
|
|
Args:
|
|
kwargs 有効なキー一覧
|
|
regist_ymd (str): 登録年月日
|
|
update_ymd (str): 更新年月日
|
|
delete_ymd (str): 削除年月日
|
|
regist_date (datetime): 登録日時
|
|
create_user (str): 登録者
|
|
update_date (datetime): 更新日時
|
|
update_user (str): 更新者
|
|
sys_regist_date (datetime): システム登録日時
|
|
regist_prgm_id (str): 登録プログラムid
|
|
sys_update_date (datetime): システム更新日時
|
|
update_prgm_id (str): 更新プログラムid
|
|
Returns:
|
|
list: 共通カラム値
|
|
"""
|
|
|
|
return [
|
|
kwargs.get('regist_ymd'), # 登録年月日(regist_ymd)
|
|
kwargs.get('update_ymd'), # 更新年月日(update_ymd)
|
|
kwargs.get('delete_ymd'), # 削除年月日(delete_ymd)
|
|
kwargs.get('regist_date'), # 登録日時(regist_date)
|
|
kwargs.get('create_user'), # 登録者(create_user)
|
|
kwargs.get('update_date'), # 更新日時(update_date)
|
|
kwargs.get('update_user'), # 更新者(update_user)
|
|
kwargs.get('sys_regist_date'), # システム登録日時(sys_regist_date)
|
|
kwargs.get('regist_prgm_id'), # 登録プログラムid(regist_prgm_id)
|
|
kwargs.get('sys_update_date'), # システム更新日時(sys_update_date)
|
|
kwargs.get('update_prgm_id') # 更新プログラムid(update_prgm_id)
|
|
]
|
|
|
|
|
|
def create_ultmarc_table_mapper_sut(line: DatFileLine, db: Database) -> UltmarcTableMapper:
|
|
"""アルトマークテーブルマッパーのインスタンスを返す
|
|
|
|
Args:
|
|
line (DatFileLine): テストデータの1行
|
|
db (Database): 接続済みDBインスタンス
|
|
|
|
Returns:
|
|
UltmarcTableMapper: マッパークラス
|
|
"""
|
|
layout_class = line.layout_class
|
|
record_id = line.record_id
|
|
factory = UltmarcTableMapperFactory()
|
|
sut = factory.create(
|
|
layout_class=layout_class,
|
|
record_id=record_id,
|
|
records=line.records,
|
|
db=db
|
|
)
|
|
|
|
return sut
|
|
|
|
|
|
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], line_number: int, ignore_col_name: list = None) -> 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 ignore_col_name is not None and actual_col_name in ignore_col_name:
|
|
continue
|
|
else:
|
|
assert actual_row[actual_col_name] == expect_row[expect_col_name], f'{line_number}行目:{actual_col_name}が、期待値と一致すること'
|