feat: ステータス管理テーブルのユニットテスト実装中。set_process_statusのテストを実装

ecs/jskult-batch/src/error/exceptions.py

@@ -8,3 +8,7 @@ class DBException(MeDaCaException):
 
 class BatchOperationException(MeDaCaException):
     pass
+
+
+class MaxRunCountReachedException(MeDaCaException):
+    pass

ecs/jskult-batch/src/manager/jskult_batch_status_manager.py

@@ -7,19 +7,19 @@ from src.system_var import constants
 class JskultBatchStatusManager:
     """実消化&アルトマーク_バッチステータス管理テーブルを管理するクラス"""
 
-    def __init__(self, process_name: str, process_type: str, max_run_count_flg: int, receive_file_count: int):
+    def __init__(self, process_name: str, process_type: str, max_run_count: int, receive_file_count: int):
         """コンストラクタ
 
         Args:
             process_name (str): 処理名
             process_type (str): 管理区分
-            max_run_count_flg (int): 最大起動回数
+            max_run_count (int): 最大起動回数
             receive_file_count (int): 受信ファイル数
         """
 
         self._process_name: str = process_name
         self._process_type: str = process_type
-        self._max_run_count_flg: str = max_run_count_flg
+        self._max_run_count: str = max_run_count
         self._receive_file_count: str = receive_file_count
 
         # DB接続モジュールを初期化
@@ -392,7 +392,7 @@ class JskultBatchStatusManager:
             run_count = record[0]['run_count']
 
             # 取得した起動回数とフィールド変数の最大起動回数が一致を確認
-            return run_count == self._max_run_count_flg
+            return run_count == self._max_run_count
         except Exception as e:
             raise BatchOperationException(e)
         finally:

ecs/jskult-batch/tests/manager/test_jskult_batch_status_manager.py

@@ -1,3 +1,94 @@
+import datetime
+
+import pytest
+
+from src.db.database import Database
+from src.manager.jskult_batch_status_manager import JskultBatchStatusManager
+
+
 class TestJskultBatchStatusManager:
-    def test_1(self):
+    @pytest.fixture(scope='function', autouse=True)
-        pass
+    def backup_hdke_tbl_record(self):
+        """
+        テスト実行前にテーブルのバックアップを取得する。
+        テスト実行後にバックアップから復元する。
+        """
+        # Setup
+        db = Database.get_instance()
+        db.connect()
+        # ステータス管理テーブル
+        backup_status_manage_records = db.execute_select(
+            'SELECT * FROM internal07.jskult_batch_status_manage')
+        db.execute('DELETE FROM internal07.jskult_batch_status_manage')
+
+        # 日付テーブル
+        backup_hdke_tbl = db.execute_select('SELECT * FROM src07.hdke_tbl')[0]
+        db.execute('DELETE FROM src07.hdke_tbl')
+
+        yield
+
+        # 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)
+
+        # 日付テーブルを復元
+        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_set_process_status_record_not_exists(self):
+        """
+        ステータス管理テーブルのレコードが0件のとき、INSERTされること
+        """
+        # 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 = JskultBatchStatusManager(
+            'unittest_process_name',
+            'unittest_process_type',
+            3,
+            33
+        )
+        sut.set_process_status('done')
+
+        # 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'] == 'done'