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

feat: 日付テーブル操作クラスのユニットテストを実装

Browse Source
This commit is contained in:
shimoda.m@nds-tyo.co.jp 2025-05-26 17:26:11 +09:00
parent 306e393ccf
commit 9ec8b764ae
3 changed files with 174 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ecs/jskult-batch/src/db/database.py
View File

@ -181,7 +181,7 @@ class Database:
self.__connection = None
def to_jst(self):
self.execute('SET time_zone = "+9:00"')
self.execute('SET time_zone = "+9:00"')
def __execute_with_transaction(self, query: str, parameters: dict):
# トランザクションを開始してクエリを実行する

2
ecs/jskult-batch/src/system_var/constants.py
View File

@ -4,7 +4,7 @@ BATCH_EXIT_CODE_SUCCESS = 0
# バッチ処理中フラグ：未処理
BATCH_ACTF_BATCH_UNPROCESSED = '0'
# バッチ処理中フラグ：処理中
BATCH_ACTF_BATCH_IN_PROCESSING = '1'
BATCH_ACTF_BATCH_START = '1'
# dump取得状態区分：未処理
DUMP_STATUS_KBN_UNPROCESSED = '0'
# dump取得状態区分：dump取得正常終了

174
ecs/jskult-batch/tests/manager/test_jskult_hdke_tbl_manager.py
View File

@ -1,3 +1,173 @@
import pytest
from src.db.database import Database
from src.manager.jskult_hdke_tbl_manager import JskultHdkeTblManager
class TestJskultHdkeTblManager:
def test_1(self):
pass
@pytest.fixture(scope='function', autouse=True)
def backup_hdke_tbl_record(self):
"""
テスト実行前に日付テーブルのバックアップを取得する。
テスト実行後にバックアップから復元する。
"""
# Setup
db = Database.get_instance()
db.connect()
backup_hdke_tbl = db.execute_select('SELECT * FROM src07.hdke_tbl')[0]
db.execute('DELETE FROM src07.hdke_tbl')
yield backup_hdke_tbl
# Teardown
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'],
})
db.disconnect()
def test_get_batch_statuses(self):
"""
日付テーブルから以下を取得できること
- 日次バッチ処理中フラグ
- dump取得状況区分
- 処理日(yyyy/mm/dd)
"""
# 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)
""")
# Act
sut = JskultHdkeTblManager()
bch_actf, dump_sts_kbn, syor_date = sut.get_batch_statuses()
# Assert
assert bch_actf == '1'
assert dump_sts_kbn == '2'
assert syor_date == '2025/05/30'
def test_update_batch_process_start(self):
"""
日付テーブルの日次バッチ処理中フラグを1(処理中)に更新できること
"""
# 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', '0', '2', CURRENT_USER(), CURRENT_TIMESTAMP(), NULL, NULL)
""")
# Act
sut = JskultHdkeTblManager()
sut.update_batch_process_start()
# Assert
actual_record = db.execute_select('SELECT * FROM src07.hdke_tbl')[0]
assert actual_record['bch_actf'] == '1'
assert actual_record['dump_sts_kbn'] == '2'
def test_update_batch_process_complete(self):
"""
日付テーブルを以下の通り更新できること
- 日次バッチ処理中フラグを0(未処理)に更新
- dump取得状態区分を0(未処理)に更新
"""
# 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)
""")
# Act
sut = JskultHdkeTblManager()
sut.update_batch_process_complete()
# Assert
actual_record = db.execute_select('SELECT * FROM src07.hdke_tbl')[0]
assert actual_record['bch_actf'] == '0'
assert actual_record['dump_sts_kbn'] == '0'
def test_can_run_process_true(self):
"""
以下の条件の時、Trueが返却されること
- 日次バッチ処理中フラグが1(処理中)
- dump取得状態区分が2(正常終了)
"""
# 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)
""")
# Act
sut = JskultHdkeTblManager()
actual = sut.can_run_process()
# Assert
assert actual is True
def test_can_run_process_false_because_bch_actf_off(self):
"""
日次バッチ処理中フラグが1ではないため、Falseが返却されること
"""
# 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', '0', '2', CURRENT_USER(), CURRENT_TIMESTAMP(), NULL, NULL)
""")
# Act
sut = JskultHdkeTblManager()
actual = sut.can_run_process()
# Assert
assert actual is False
def test_can_run_process_false_because_dump_sts_kbn_not_complete(self):
"""
dump取得状態区分が2ではないため、Falseが返却されること
"""
# 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', '0', CURRENT_USER(), CURRENT_TIMESTAMP(), NULL, NULL)
""")
# Act
sut = JskultHdkeTblManager()
actual = sut.can_run_process()
# Assert
assert actual is False