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

import datetime

import pytest

from src.db.database import Database
from src.manager.jskult_batch_status_manager import JskultBatchStatusManager


class TestJskultBatchStatusManager:
    @pytest.fixture(scope='function', autouse=True)
    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')
        db.execute('DELETE FROM src07.hdke_tbl')

        yield

        # Teardown
        # ステータス管理テーブルを復元
        db.execute('DELETE FROM internal07.jskult_batch_status_manage')
        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)

        # 日付テーブルを復元
        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()

    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('start')

        # 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'] == '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
            ('unittest_process_name', '2025-05-30','unittest_process_type',
            'retry', '1', '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'

    def test_can_run_post_process_success(self):
        """
        完了しているデータ取り込み件数と受信ファイル数が一致している場合、trueが返却されること
        起動回数が+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', '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
            ('unittest_process_name', '2025-05-30', 'unittest_process_type',
            'start', '0', '0'),
            ('data_import_process_1', '2025-05-30','data_import',
            'done', '0', '0'),
            ('data_import_process_2', '2025-05-30','data_import',
            'done', '0', '0'),
            ('data_import_process_3', '2025-05-30','data_import',
            'done', '0', '0')
        """)

        # Act
        sut = JskultBatchStatusManager(
            'unittest_process_name',
            'unittest_process_type',
            3,
            3
        )
        actual = sut.can_run_post_process()

        # Assert
        assert actual is True
        actual_record = db.execute_select("""
        SELECT
            total_run_count
        FROM
            internal07.jskult_batch_status_manage
        WHERE
            process_name = 'unittest_process_name'
        AND process_date = '2025-05-30'
        """)[0]

        assert actual_record['total_run_count'] == 1