from unittest.mock import MagicMock, patch import pytest from src import controller from src.config.objects import (FetchTargetObjects, LastFetchDatetime, TargetObject) from src.util.execute_datetime import ExecuteDateTime from .test_utils.log_message import generate_log_message_tuple TEST_OBJECT_INFO_LIST = { 'objects': [ { 'object_name': 'Account', 'upload_file_name': 'Account_YYYYMMDDHHMMSS' }, { 'object_name': 'Contact', 'upload_file_name': 'Contact_YYYYMMDDHHMMSS' }, { 'object_name': 'Call2_vod__c', 'upload_file_name': 'Call2_vod__c_YYYYMMDDHHMMSS' } ] } EXPECT_PROCESS_RESULT = { 'Account': 'success', 'Contact': 'success', 'Call2_vod__c': 'success' } class TestController: @pytest.fixture() def call_all_processes(self, s3_client, monkeypatch, caplog): """ コントロール処理内ですべてのプロセス関数が呼ばれることのテストで使用するフィクスチャ 各種プロセス関数をモック化し、正常終了させるように動く Yields: dict: プロセス関数呼び出し後のモックの辞書 """ def _func(): monkeypatch.setattr(FetchTargetObjects, '_FetchTargetObjects__validate', lambda x: None) monkeypatch.setattr(TargetObject, '_TargetObject__validate', lambda x: None) monkeypatch.setattr(LastFetchDatetime, '_LastFetchDatetime__validate', lambda x: None) mock_execute_datetime = ExecuteDateTime() mock_target_objects = FetchTargetObjects(TEST_OBJECT_INFO_LIST) mock_target_object_first = TargetObject(TEST_OBJECT_INFO_LIST['objects'][0], mock_execute_datetime) mock_target_object_second = TargetObject(TEST_OBJECT_INFO_LIST['objects'][1], mock_execute_datetime) mock_target_object_third = TargetObject(TEST_OBJECT_INFO_LIST['objects'][2], mock_execute_datetime) mock_return_values = [mock_target_object_first, mock_target_object_second, mock_target_object_third] mock_last_fetch_datetime = LastFetchDatetime({ 'last_fetch_datetime_from': '1900-01-01T00:00:00.000Z', 'last_fetch_datetime_to': '9999-12-31T59:59:59.000Z' }, mock_execute_datetime) mock_prepare_data_fetch_process = MagicMock(return_value=(mock_target_objects, mock_execute_datetime, {})) mock_check_object_info_process = MagicMock(side_effect=mock_return_values) mock_set_datetime_period_process = MagicMock(return_value=mock_last_fetch_datetime) mock_fetch_crm_data_process = MagicMock(return_value=[{}]) mock_backup_crm_data_process = MagicMock() mock_convert_crm_csv_data_process = MagicMock() mock_backup_crm_csv_data_process = MagicMock() mock_copy_crm_csv_data_process = MagicMock() mock_upload_last_fetch_datetime_process = MagicMock() mock_upload_result_data_process = MagicMock() should_call_processes = { 'prepare_data_fetch_process': mock_prepare_data_fetch_process, 'check_object_info_process': mock_check_object_info_process, 'set_datetime_period_process': mock_set_datetime_period_process, 'fetch_crm_data_process': mock_fetch_crm_data_process, 'backup_crm_data_process': mock_backup_crm_data_process, 'convert_crm_csv_data_process': mock_convert_crm_csv_data_process, 'backup_crm_csv_data_process': mock_backup_crm_csv_data_process, 'copy_crm_csv_data_process': mock_copy_crm_csv_data_process, 'upload_last_fetch_datetime_process': mock_upload_last_fetch_datetime_process, 'upload_result_data_process': mock_upload_result_data_process } with patch('src.controller.prepare_data_fetch_process', mock_prepare_data_fetch_process),\ patch('src.controller.check_object_info_process', mock_check_object_info_process),\ patch('src.controller.set_datetime_period_process', mock_set_datetime_period_process),\ patch('src.controller.fetch_crm_data_process', mock_fetch_crm_data_process),\ patch('src.controller.backup_crm_data_process', mock_backup_crm_data_process),\ patch('src.controller.convert_crm_csv_data_process', mock_convert_crm_csv_data_process),\ patch('src.controller.backup_crm_csv_data_process', mock_backup_crm_csv_data_process),\ patch('src.controller.copy_crm_csv_data_process', mock_copy_crm_csv_data_process),\ patch('src.controller.upload_last_fetch_datetime_process', mock_upload_last_fetch_datetime_process),\ patch('src.controller.upload_result_data_process', mock_upload_result_data_process): controller.controller() return should_call_processes yield _func def test_controller_call_all_processes(self, call_all_processes): """ Cases: コントロール処理からすべてのプロセスが呼ばれること Arranges: - call_all_processesフィクスチャで、コントロール処理を実行しておく Expects: - データ取得準備処理が1回のみ実行される - オブジェクト情報形式チェック処理が複数回実行される - データ取得期間設定処理が複数回実行される - CRMデータ取得処理が複数回実行される - CRM電文データバックアップ処理が複数回実行される - CSV変換処理が複数回実行される - CSVバックアップ処理が複数回実行される - CSVアップロード処理が複数回実行される - 前回取得日時ファイル更新処理が複数回実行される - 取得処理実施結果アップロード処理が1回のみ実行される """ processes = call_all_processes() processes['prepare_data_fetch_process'].assert_called_once(), 'データ取得準備処理が1回のみ実行されること' assert processes['check_object_info_process'].call_count == 3, 'オブジェクト情報形式チェック処理が複数回実行されること' assert processes['set_datetime_period_process'].call_count == 3, 'データ取得期間設定処理が複数回実行されること' assert processes['fetch_crm_data_process'].call_count == 3, 'CRMデータ取得処理が複数回実行されること' assert processes['backup_crm_data_process'].call_count == 3, 'CRM電文データバックアップ処理が複数回実行されること' assert processes['convert_crm_csv_data_process'].call_count == 3, 'CSV変換処理が複数回実行されること' assert processes['backup_crm_csv_data_process'].call_count == 3, 'CSVバックアップ処理が複数回実行されること' assert processes['copy_crm_csv_data_process'].call_count == 3, 'CSVアップロード処理が複数回実行されること' assert processes['upload_last_fetch_datetime_process'].call_count == 3, '前回取得日次ファイル更新処理が複数回実行されること' processes['upload_result_data_process'].assert_called_once(), '取得処理実施結果アップロード処理が1回のみ実行されること' def test_controller_print_normal_logs(self, call_all_processes, caplog): """ Cases: コントロール処理の正常系ログがすべての出力されていること Arranges: - call_all_processesフィクスチャで、コントロール処理を実行しておく Expects: - コントロール処理の正常系ログがすべて出力されている """ call_all_processes() # シングルトンなロガーを引き回しているので、rootロガーになる assert generate_log_message_tuple(log_message='I-CTRL-01 CRMデータ取得処理を開始します') in caplog.record_tuples assert generate_log_message_tuple(log_message='I-CTRL-02 データ取得準備処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message='I-CTRL-03 取得対象オブジェクトのループ処理開始') in caplog.record_tuples for obj in TEST_OBJECT_INFO_LIST['objects']: object_name = obj['object_name'] upload_file_name = obj['upload_file_name'] assert generate_log_message_tuple(log_message='I-CTRL-05 オブジェクト情報形式チェック処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-06 [{object_name}]のデータ取得を開始します') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-08 [{object_name}]のデータ取得期間設定処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-09 [{object_name}]のデータ取得処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-10 [{object_name}] の出力ファイル名は [{upload_file_name}] となります') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-11 [{object_name}] CRM電文データバックアップ処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-12 [{object_name}] CSV変換処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-13 [{object_name}] CSVデータバックアップ処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-14 [{object_name}] CSVデータアップロード処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-15 [{object_name}] 前回取得日時ファイル更新処理呼び出し') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-16 [{object_name}] 処理正常終了') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-17 すべてのオブジェクトの処理が終了しました 実行結果：[{EXPECT_PROCESS_RESULT}]') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-18 CRM_取得処理実施結果ファイルアップロード処理開始') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-19 すべてのデータの取得に成功しました') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-CTRL-20 CRMデータ取得処理を終了します') in caplog.record_tuples def test_controller_do_not_call_upload_process_result_process(self, s3_client, monkeypatch, caplog): """ Cases: 処理対象オブジェクトが0件の場合、取得処理実施結果アップロード処理が実行されないこと Arranges: - データ取得準備処理で返される取得対象オブジェクト情報を0件にする - 各種プロセスメソッドと内部で使用している値オブジェクトをモック化する Expects: - データ取得準備処理が1回のみ実行されること - 取得処理実施結果アップロード処理が実行されないこと - 処理対象が存在しない旨を示すログメッセージが出力されていること """ mock_execute_datetime = ExecuteDateTime() mock_prepare_data_fetch_process = MagicMock(return_value=([], mock_execute_datetime, {})) mock_check_object_info_process = MagicMock() mock_set_datetime_period_process = MagicMock() mock_fetch_crm_data_process = MagicMock() mock_backup_crm_data_process = MagicMock() mock_convert_crm_csv_data_process = MagicMock() mock_backup_crm_csv_data_process = MagicMock() mock_copy_crm_csv_data_process = MagicMock() mock_upload_last_fetch_datetime_process = MagicMock() mock_upload_result_data_process = MagicMock() with patch('src.controller.prepare_data_fetch_process', mock_prepare_data_fetch_process),\ patch('src.controller.check_object_info_process', mock_check_object_info_process),\ patch('src.controller.set_datetime_period_process', mock_set_datetime_period_process),\ patch('src.controller.fetch_crm_data_process', mock_fetch_crm_data_process),\ patch('src.controller.backup_crm_data_process', mock_backup_crm_data_process),\ patch('src.controller.convert_crm_csv_data_process', mock_convert_crm_csv_data_process),\ patch('src.controller.backup_crm_csv_data_process', mock_backup_crm_csv_data_process),\ patch('src.controller.copy_crm_csv_data_process', mock_copy_crm_csv_data_process),\ patch('src.controller.upload_last_fetch_datetime_process', mock_upload_last_fetch_datetime_process),\ patch('src.controller.upload_result_data_process', mock_upload_result_data_process): controller.controller() mock_prepare_data_fetch_process.assert_called_once(), 'データ取得準備処理が1回のみ実行されること' mock_upload_result_data_process.call_count == 0, '取得処理実施結果アップロード処理が実行されないこと' # シングルトンなロガーを引き回しているので、rootロガーになる assert generate_log_message_tuple(log_message='I-CTRL-21 処理対象のデータが存在しませんでした') in caplog.record_tuples, '処理対象が存在しない旨を示すログメッセージが出力されていること' assert generate_log_message_tuple(log_message=f'I-CTRL-20 CRMデータ取得処理を終了します') in caplog.record_tuples