161 lines
6.2 KiB
Python
161 lines
6.2 KiB
Python
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from src.config.objects import TargetObject
|
|
from src.copy_crm_csv_data_process import copy_crm_csv_data_process
|
|
from src.error.exceptions import FileUploadException
|
|
from src.system_var.constants import UPLD_JP_NAME
|
|
from src.util.execute_datetime import ExecuteDateTime
|
|
|
|
from .test_utils.log_message import generate_log_message_tuple
|
|
|
|
|
|
class TestCopyCrmCsvDataProcess:
|
|
|
|
@pytest.fixture
|
|
def bucket_name(self):
|
|
return 'test-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:
|
|
CSVアップロード処理が正常終了し、期待通りの結果が返ること
|
|
Arranges:
|
|
- オブジェクト情報の準備
|
|
- コピー元のバケットのモックを生成する
|
|
- コピー先のバケットの環境変数を設定する
|
|
- コピー元にファイルを配置する
|
|
Expects:
|
|
- コピーしたしたCSVファイルが存在し内容が想定と一致する
|
|
- CSVアップロード処理の仕様に沿った正常系ログが出力されること(デバッグログは除く)
|
|
"""
|
|
|
|
# Arranges
|
|
|
|
# オブジェクト情報の準備
|
|
target_objects_dict = {
|
|
'object_name': 'Account',
|
|
'columns': [
|
|
'Id',
|
|
'Name'
|
|
]
|
|
}
|
|
|
|
execute_datetime = ExecuteDateTime()
|
|
|
|
target_object = TargetObject(target_objects_dict, execute_datetime)
|
|
|
|
# コピー元のバケットをモック化
|
|
from_copy_bucket = 'for-copy-bucket'
|
|
s3_client.create_bucket(Bucket=from_copy_bucket)
|
|
monkeypatch.setattr('src.aws.s3.CRM_BACKUP_BUCKET', from_copy_bucket)
|
|
monkeypatch.setattr('src.copy_crm_csv_data_process.CRM_IMPORT_DATA_BACKUP_FOLDER', 'data_import')
|
|
|
|
# コピー先のバケットをモック化
|
|
monkeypatch.setattr('src.aws.s3.IMPORT_DATA_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.CRM_IMPORT_DATA_FOLDER', 'crm/target')
|
|
|
|
# コピー元のモックバケットにデータを配置
|
|
s3_client.put_object(Bucket=from_copy_bucket,
|
|
Key=f'data_import/{execute_datetime.to_path()}/CRM_Account_{execute_datetime.format_date()}.csv', Body=b'test,test,test')
|
|
|
|
# Act
|
|
copy_crm_csv_data_process(target_object, execute_datetime)
|
|
|
|
# Assert
|
|
# ファイル確認
|
|
actual = s3_client.get_object(
|
|
Bucket=bucket_name, Key=f'crm/target/CRM_Account_{execute_datetime.format_date()}.csv')
|
|
assert actual['Body'].read().decode('utf-8') == 'test,test,test'
|
|
|
|
# ログの確認
|
|
assert generate_log_message_tuple(
|
|
log_message=f'I-UPLD-01 [Account] のCSVデータアップロード処理を開始します ファイル名：[CRM_Account_{execute_datetime.format_date()}.csv]') in caplog.record_tuples
|
|
assert generate_log_message_tuple(log_message=f'I-UPLD-03 [Account] のCSVデータのアップロード処理を終了します') in caplog.record_tuples
|
|
|
|
def test_call_depended_modules(self):
|
|
"""
|
|
Cases:
|
|
CSVアップロード処理内で依存しているモジュールが正しく呼ばれていること
|
|
Arranges:
|
|
- CSVアップロード処理の依存モジュールをモック化する
|
|
- オブジェクト情報の準備
|
|
Expects:
|
|
- 依存しているモジュールが正しく呼ばれている
|
|
"""
|
|
|
|
# Arrange
|
|
|
|
target_objects_dict = {
|
|
'object_name': 'Account',
|
|
'columns': [
|
|
'Id',
|
|
'Name'
|
|
]
|
|
}
|
|
|
|
execute_datetime = ExecuteDateTime()
|
|
|
|
target_object = TargetObject(target_objects_dict, execute_datetime)
|
|
|
|
mock_data_bucket = MagicMock(return_value=None)
|
|
mock_backup_bucket = MagicMock(return_value=None)
|
|
mock_copy_bucket = MagicMock(return_value=None)
|
|
|
|
# Act
|
|
with patch('src.aws.s3.DataBucket.__init__', mock_data_bucket), \
|
|
patch('src.aws.s3.BackupBucket.__init__', mock_backup_bucket), \
|
|
patch('src.aws.s3.DataBucket.put_csv_from', mock_copy_bucket):
|
|
copy_crm_csv_data_process(target_object, execute_datetime)
|
|
|
|
# Assert
|
|
assert mock_data_bucket.called is True
|
|
assert mock_backup_bucket.called is True
|
|
assert mock_copy_bucket.called is True
|
|
|
|
def test_raise_run_process_success(self, s3_client, prepare_bucket, bucket_name, monkeypatch, caplog):
|
|
"""
|
|
Cases:
|
|
CSVデータをコピーできない場合、エラーが発生すること
|
|
Arranges:
|
|
- オブジェクト情報の準備
|
|
- コピー元のバケットのモックを生成する
|
|
- コピー先のバケットの環境変数を設定する
|
|
- コピー元にファイルを配置する
|
|
Expects:
|
|
- 例外が発生する
|
|
- ファイルがアップロードできないエラーが返却される
|
|
"""
|
|
|
|
# Arranges
|
|
|
|
# オブジェクト情報の準備
|
|
target_objects_dict = {
|
|
'object_name': 'Account',
|
|
'columns': [
|
|
'Id',
|
|
'Name'
|
|
]
|
|
}
|
|
|
|
execute_datetime = ExecuteDateTime()
|
|
|
|
target_object = TargetObject(target_objects_dict, execute_datetime)
|
|
|
|
mock_copy_bucket = MagicMock(side_effect=Exception('ファイルコピーエラー'))
|
|
|
|
# Act
|
|
with patch('src.aws.s3.DataBucket.put_csv_from', mock_copy_bucket):
|
|
with pytest.raises(FileUploadException) as e:
|
|
copy_crm_csv_data_process(target_object, execute_datetime)
|
|
|
|
# Assert
|
|
assert e.value.error_id == 'E-UPLD-01'
|
|
assert e.value.func_name == UPLD_JP_NAME
|
|
assert e.value.args[0] == f'[Account] CSVデータのアップロードに失敗しました ファイル名：[CRM_Account_{execute_datetime.format_date()}.csv] エラー内容:[ファイルコピーエラー]'
|