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

import pytest

from src.db.database import Database
from src.manager.jskult_hdke_tbl_manager import JskultHdkeTblManager


class TestJskultHdkeTblManager:
    @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

        # 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