nik/newdwh2021
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: set_process_statusのテストを実装（Updateのケース）

Browse Source
This commit is contained in:
shimoda.m@nds-tyo.co.jp 2025-05-27 13:26:55 +09:00
parent 491ba963a8
commit 274c7abec1
1 changed files with 79 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
ecs/jskult-batch/tests/manager/test_jskult_batch_status_manager.py
View File

@ -22,7 +22,7 @@ class TestJskultBatchStatusManager:
db.execute('DELETE FROM internal07.jskult_batch_status_manage')
# 日付テーブル
backup_hdke_tbl = db.execute_select('SELECT * FROM src07.hdke_tbl')[0]
backup_hdke_tbl = db.execute_select('SELECT * FROM src07.hdke_tbl')
db.execute('DELETE FROM src07.hdke_tbl')
yield
@ -30,25 +30,27 @@ class TestJskultBatchStatusManager:
# Teardown
# ステータス管理テーブルを復元
db.execute('DELETE FROM internal07.jskult_batch_status_manage')
db.execute("""
INSERT INTO internal07.jskult_batch_status_manage
(process_name,process_date,process_type,process_status,total_run_count,max_run_count_flg,ins_user,ins_date,upd_user,upd_date)
VALUES
(:process_name,:process_date,:process_type,:process_status,:total_run_count,:max_run_count_flg,:ins_user,:ins_date,:upd_user,:upd_date)
""", backup_status_manage_records)
if len(backup_status_manage_records) != 0:
db.execute("""
INSERT INTO internal07.jskult_batch_status_manage
(process_name,process_date,process_type,process_status,total_run_count,max_run_count_flg,ins_user,ins_date,upd_user,upd_date)
VALUES
(:process_name,:process_date,:process_type,:process_status,:total_run_count,:max_run_count_flg,:ins_user,:ins_date,:upd_user,:upd_date)
""", backup_status_manage_records)
# 日付テーブルを復元
db.execute('DELETE FROM src07.hdke_tbl')
db.execute("""
INSERT INTO src07.hdke_tbl
(syor_date, bch_actf, dump_sts_kbn, creater, create_date, updater, update_date)
VALUES
(:syor_date, :bch_actf, :dump_sts_kbn, CURRENT_USER(), CURRENT_TIMESTAMP(), NULL, NULL)
""", {
'syor_date': backup_hdke_tbl['syor_date'],
'bch_actf': backup_hdke_tbl['bch_actf'],
'dump_sts_kbn': backup_hdke_tbl['dump_sts_kbn'],
})
if len(backup_hdke_tbl) != 0:
db.execute('DELETE FROM src07.hdke_tbl')
db.execute("""
INSERT INTO src07.hdke_tbl
(syor_date, bch_actf, dump_sts_kbn, creater, create_date, updater, update_date)
VALUES
(:syor_date, :bch_actf, :dump_sts_kbn, CURRENT_USER(), CURRENT_TIMESTAMP(), NULL, NULL)
""", {
'syor_date': backup_hdke_tbl[0]['syor_date'],
'bch_actf': backup_hdke_tbl[0]['bch_actf'],
'dump_sts_kbn': backup_hdke_tbl[0]['dump_sts_kbn'],
})
db.disconnect()
@ -74,7 +76,7 @@ class TestJskultBatchStatusManager:
3,
33
)
sut.set_process_status('done')
sut.set_process_status('start')
# Assert
actual = db.execute_select("""
@ -91,4 +93,61 @@ class TestJskultBatchStatusManager:
assert datetime.datetime.strftime(
actual['process_date'], '%Y/%m/%d') == '2025/05/30'
assert actual['process_type'] == 'unittest_process_type'
assert actual['process_status'] == 'done'
assert actual['process_status'] == 'start'
def test_set_process_status_record_exists(self):
"""
ステータス管理テーブルのレコードが1件のとき、UPDATEされること
"""
# Arrange
# 日付テーブルを登録
db = Database.get_instance()
db.connect()
db.execute("""
INSERT INTO src07.hdke_tbl
(syor_date, bch_actf, dump_sts_kbn, creater, create_date, updater, update_date)
VALUES
('20250530', '1', '2', CURRENT_USER(), CURRENT_TIMESTAMP(), NULL, NULL)
""")
# バッチステータス管理テーブルを登録
db.execute("""
INSERT INTO internal07.jskult_batch_status_manage
(process_name,process_date,process_type,process_status,total_run_count,max_run_count_flg)
VALUES
(:process_name,:process_date,:process_type,:process_status,:total_run_count,:max_run_count_flg)
""",
{
'process_name': 'unittest_process_name',
'process_date': '2025-05-30',
'process_type': 'unittest_process_type',
'process_status': 'retry',
'total_run_count': '1',
'max_run_count_flg': '0'
})
# Act
sut = JskultBatchStatusManager(
'unittest_process_name',
'unittest_process_type',
3,
33
)
sut.set_process_status('doing')
# Assert
actual = db.execute_select("""
SELECT
process_name,
process_date,
process_type,
process_status
FROM
internal07.jskult_batch_status_manage
""")[0]
assert actual['process_name'] == 'unittest_process_name'
assert datetime.datetime.strftime(
actual['process_date'], '%Y/%m/%d') == '2025/05/30'
assert actual['process_type'] == 'unittest_process_type'
assert actual['process_status'] == 'doing'