from unittest.mock import MagicMock, patch import pytest from src.config.objects import LastFetchDatetime, TargetObject from src.error.exceptions import FileUploadException from src.system_var.constants import UPD_JP_NAME from src.upload_last_fetch_datetime_process import \ upload_last_fetch_datetime_process from src.util.execute_datetime import ExecuteDateTime from .test_utils.log_message import generate_log_message_tuple class TestUploadLastFetchDatetimeProcess: @pytest.fixture def bucket_name(self): return 'test-config-bucket' @pytest.fixture def prepare_bucket(self, s3_client, bucket_name): s3_client.create_bucket(Bucket=bucket_name) yield def test_run_process_success(self, s3_client, prepare_bucket, bucket_name, monkeypatch, caplog): """ Cases: 前回取得日時ファイル更新処理が正常終了し、期待通りの結果が返ること Arranges: - prepare_bucketフィクスチャで、前回取得日時ファイルを置くためのモックバケットを作る - モックバケットを指すように環境変数を設定する - オブジェクト情報を準備する - 最終更新日時インスタンスを準備する Expects: - PUTした前回取得日時ファイルが存在し内容が想定と一致する - 前回取得日時ファイル更新処理の仕様に沿った正常系ログが出力されること(デバッグログは除く) """ # Arranges # 環境変数を編集 monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name) monkeypatch.setattr('src.aws.s3.LAST_FETCH_DATE_FOLDER', 'crm/last_fetch_datetime') target_objects_dict = { 'object_name': 'Account', 'columns': [ 'Id', 'Name' ], 'is_update_last_fetch_datetime': True } last_fetch_datetime_dict = { "last_fetch_datetime_from": "1999-01-01T00:00:00.000Z", "last_fetch_datetime_to": "2100-12-31T23:59:59.000Z", } execute_datetime = ExecuteDateTime() target_object = TargetObject(target_objects_dict, execute_datetime) last_fetch_datetime = LastFetchDatetime(last_fetch_datetime_dict, execute_datetime) # Act upload_last_fetch_datetime_process(target_object, last_fetch_datetime) # Assert # ファイル確認 actual = s3_client.get_object(Bucket=bucket_name, Key=f'crm/last_fetch_datetime/Account.json') assert actual['Body'].read().decode( 'utf-8') == '{"last_fetch_datetime_from": "2100-12-31T23:59:59.000Z", "last_fetch_datetime_to": ""}' # ログの確認 assert generate_log_message_tuple(log_message='I-UPD-01 [Account] の前回取得日時ファイルの更新処理を開始します') in caplog.record_tuples assert generate_log_message_tuple(log_message='I-UPD-02 [Account] の前回取得日時ファイルの更新処理をスキップします') not in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-UPD-04 [Account] の前回取得日時ファイルの更新処理を終了します') in caplog.record_tuples def test_run_process_success_skip(self, s3_client, prepare_bucket, bucket_name, monkeypatch, caplog): """ Cases: 前回取得日時ファイル更新処理が正常終了し、期待通りの結果が返ること Arranges: - prepare_bucketフィクスチャで、前回取得日時ファイルを置くためのモックバケットを作る - モックバケットを指すように環境変数を設定する - ファイルPUT処理のモックを準備する - オブジェクト情報を準備する - 最終更新日時インスタンスを準備する Expects: - ファイルPUT処理が実行されないこと - 前回取得日時ファイル更新処理の仕様に沿った正常系ログが出力されること(デバッグログは除く) """ # Arranges # 環境変数を編集 monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name) monkeypatch.setattr('src.aws.s3.LAST_FETCH_DATE_FOLDER', 'crm/last_fetch_datetime') mock_config_bucket = MagicMock(return_value=None) target_objects_dict = { 'object_name': 'Account', 'columns': [ 'Id', 'Name' ], 'is_update_last_fetch_datetime': False } last_fetch_datetime_dict = { "last_fetch_datetime_from": "1999-01-01T00:00:00.000Z", "last_fetch_datetime_to": "2100-12-31T23:59:59.000Z", } execute_datetime = ExecuteDateTime() target_object = TargetObject(target_objects_dict, execute_datetime) last_fetch_datetime = LastFetchDatetime(last_fetch_datetime_dict, execute_datetime) # Act with patch('src.upload_last_fetch_datetime_process.ConfigBucket') as mock_config_bucket, \ patch('src.upload_last_fetch_datetime_process.JsonWriter') as mock_writer: mock_writer_inst = mock_writer.return_value mock_writer_inst.write.return_value = '' mock_config_bucket_inst = mock_config_bucket.return_value mock_config_bucket_inst.put_last_fetch_datetime_file.return_value = '' upload_last_fetch_datetime_process(target_object, last_fetch_datetime) # Assert # 処理実行確認 assert mock_config_bucket_inst.put_last_fetch_datetime_file.called is False assert mock_config_bucket.called is False # ログの確認 assert generate_log_message_tuple(log_message='I-UPD-01 [Account] の前回取得日時ファイルの更新処理を開始します') in caplog.record_tuples assert generate_log_message_tuple(log_message='I-UPD-02 [Account] の前回取得日時ファイルの更新処理をスキップします') in caplog.record_tuples assert generate_log_message_tuple(log_message=f'I-UPD-04 [Account] の前回取得日時ファイルの更新処理を終了します') not in caplog.record_tuples def test_call_depended_modules(self): """ Cases: 前回取得日時ファイル更新処理内で依存しているモジュールが正しく呼ばれていること Arranges: - 前回取得日時ファイル更新処理の依存モジュールをモック化する - オブジェクト情報を準備する - 最終更新日時インスタンスを準備する Expects: - 依存しているモジュールが正しく呼ばれている """ # Arrange target_objects_dict = { 'object_name': 'Account', 'columns': [ 'Id', 'Name' ], 'is_update_last_fetch_datetime': True } last_fetch_datetime_dict = { "last_fetch_datetime_from": "1999-01-01T00:00:00.000Z", "last_fetch_datetime_to": "2100-12-31T23:59:59.000Z", } execute_datetime = ExecuteDateTime() target_object = TargetObject(target_objects_dict, execute_datetime) last_fetch_datetime = LastFetchDatetime(last_fetch_datetime_dict, execute_datetime) # Act with patch('src.upload_last_fetch_datetime_process.ConfigBucket') as mock_config_bucket, \ patch('src.upload_last_fetch_datetime_process.JsonWriter') as mock_writer: mock_writer_inst = mock_writer.return_value mock_writer_inst.write.return_value = '' mock_config_bucket_inst = mock_config_bucket.return_value mock_config_bucket_inst.put_last_fetch_datetime_file.return_value = '' upload_last_fetch_datetime_process(target_object, last_fetch_datetime) # Assert assert mock_config_bucket_inst.put_last_fetch_datetime_file.called is True assert mock_writer_inst.write.called is True def test_raise_put_last_fetch_datetime_file(self, monkeypatch): """ Cases: 前回取得日時ファイルをアップロードできない場合、エラーが発生すること Arranges: - 前回取得日時ファイル更新処理で例外を発生させるモックを生成する - 取得処理実施結果を準備する - 実行日時取得インスタンスを生成する Expects: - 例外が発生する - ファイルがアップロードできないエラーが返却される """ # Arrange target_objects_dict = { 'object_name': 'Account', 'columns': [ 'Id', 'Name' ], 'is_update_last_fetch_datetime': True } last_fetch_datetime_dict = { "last_fetch_datetime_from": "1999-01-01T00:00:00.000Z", "last_fetch_datetime_to": "2100-12-31T23:59:59.000Z", } execute_datetime = ExecuteDateTime() target_object = TargetObject(target_objects_dict, execute_datetime) last_fetch_datetime = LastFetchDatetime(last_fetch_datetime_dict, execute_datetime) # Act with patch('src.upload_last_fetch_datetime_process.ConfigBucket') as mock_config_bucket, \ patch('src.upload_last_fetch_datetime_process.JsonWriter') as mock_writer: mock_writer_inst = mock_writer.return_value mock_writer_inst.write.return_value = '' mock_config_bucket_inst = mock_config_bucket.return_value mock_config_bucket_inst.put_last_fetch_datetime_file.side_effect = Exception('ファイルアップロードエラー') with pytest.raises(FileUploadException) as e: upload_last_fetch_datetime_process(target_object, last_fetch_datetime) # Assert assert mock_writer_inst.write.called is True assert mock_config_bucket_inst.put_last_fetch_datetime_file.called is True assert e.value.error_id == 'E-UPD-01' assert e.value.func_name == UPD_JP_NAME assert e.value.args[0] == f'[Account] 前回処理日時ファイルのアップロードに失敗しました ファイル名：[Account.json] エラー内容:[ファイルアップロードエラー]' def test_raise_put_last_fetch_datetime_file_write_local_file(self, monkeypatch): """ Cases: 前回取得日時ファイルをアップロードするためのローカルファイルの書き込みに失敗した場合、エラーが発生すること Arranges: - 前回取得日時ファイル更新処理で例外を発生させるモックを生成する - 取得処理実施結果を準備する - 実行日時取得インスタンスを生成する Expects: - 例外が発生する - ファイルが書き込めないエラーが返却される """ # Arrange target_objects_dict = { 'object_name': 'Account', 'columns': [ 'Id', 'Name' ], 'is_update_last_fetch_datetime': True } last_fetch_datetime_dict = { "last_fetch_datetime_from": "1999-01-01T00:00:00.000Z", "last_fetch_datetime_to": "2100-12-31T23:59:59.000Z", } execute_datetime = ExecuteDateTime() target_object = TargetObject(target_objects_dict, execute_datetime) last_fetch_datetime = LastFetchDatetime(last_fetch_datetime_dict, execute_datetime) # Act with patch('src.upload_last_fetch_datetime_process.ConfigBucket') as mock_config_bucket, \ patch('src.upload_last_fetch_datetime_process.JsonWriter') as mock_writer: mock_writer_inst = mock_writer.return_value mock_writer_inst.write.side_effect = Exception('ファイル書き込みエラー') mock_config_bucket_inst = mock_config_bucket.return_value mock_config_bucket_inst.put_last_fetch_datetime_file.return_value = '' with pytest.raises(FileUploadException) as e: upload_last_fetch_datetime_process(target_object, last_fetch_datetime) # Assert assert mock_writer_inst.write.called is True assert mock_config_bucket_inst.put_last_fetch_datetime_file.called is False assert e.value.error_id == 'E-UPD-01' assert e.value.func_name == UPD_JP_NAME assert e.value.args[0] == f'[Account] 前回処理日時ファイルのアップロードに失敗しました ファイル名：[Account.json] エラー内容:[ファイル書き込みエラー]'