import textwrap from collections import OrderedDict import pytest from src.config.objects import TargetObject from src.converter.converter import CSVStringConverter from src.util.execute_datetime import ExecuteDateTime class TestCSVStringConverter: def test_convert(self) -> str: """ Cases: 入力データがCSV形式の文字列で出力されること Arranges: - オブジェクト情報の作成 - データの作成 - 実行日時取得インスタンスの生成 - オブジェクト情報インスタンスの生成 Expects: 戻り値が期待値と一致すること """ # Arranges object_info = { "object_name": "AccountShare", "columns": [ "Id", "AccountId", "UserOrGroupId", "AccountAccessLevel", "OpportunityAccessLevel", "CaseAccessLevel", "ContactAccessLevel", "RowCause", "LastModifiedDate", "LastModifiedById", "IsDeleted" ], "is_skip": False, "is_update_last_fetch_datetime": False, "last_fetch_datetime_file_name": "AccountShare.json", "upload_file_name": "CRM_AccountShare_{execute_datetime}", "datetime_column": "LastModifiedDate" } data = [ OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST001'), ('AccountId', 'test001'), ('UserOrGroupId', None), ('AccountAccessLevel', 1), ('OpportunityAccessLevel', 2), ('CaseAccessLevel', 3), ('ContactAccessLevel', 4), ('RowCause', 'テストのため1'), ('LastModifiedDate', '2022-06-01T00:00:00.000+0000'), ('LastModifiedById', 1.234567E+6), ('IsDeleted', False) ]), OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST002'), ('AccountId', 'test002'), ('UserOrGroupId', None), ('AccountAccessLevel', 5), ('OpportunityAccessLevel', 6), ('CaseAccessLevel', 7), ('ContactAccessLevel', 8), ('RowCause', 'テストのため2'), ('LastModifiedDate', '2022-06-02T16:30:30.000+0000'), ('LastModifiedById', 2.234567E+6), ('IsDeleted', True) ]), OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST003'), ('AccountId', 'test003'), ('UserOrGroupId', None), ('AccountAccessLevel', 9), ('OpportunityAccessLevel', 10), ('CaseAccessLevel', 11), ('ContactAccessLevel', 12), ('RowCause', 'テストのため3'), ('LastModifiedDate', '2022-06-03T23:50:50.000+0000'), ('LastModifiedById', 3.234567E+6), ('IsDeleted', False) ]) ] execute_datetime = ExecuteDateTime() target_object = TargetObject(object_info, execute_datetime) # Act csv_string_converter = CSVStringConverter(target_object, data) actual = csv_string_converter.convert() # Expects expected_value = '''\ "Id","AccountId","UserOrGroupId","AccountAccessLevel","OpportunityAccessLevel","CaseAccessLevel","ContactAccessLevel","RowCause","LastModifiedDate","LastModifiedById","IsDeleted"\r\n\ "TEST001","test001","","1","2","3","4","テストのため1","2022-06-01 09:00:00","1234567","0"\r\n\ "TEST002","test002","","5","6","7","8","テストのため2","2022-06-03 01:30:30","2234567","1"\r\n\ "TEST003","test003","","9","10","11","12","テストのため3","2022-06-04 08:50:50","3234567","0"\r\n\ ''' # expected_valueのインデントが半角スペースと認識されてしまうため、`textwrap.dedent`にて補正 assert actual == textwrap.dedent(expected_value) def test_raise_convert_extract_jsons(self) -> str: """ Cases: 一部データのキー（attributes）の不足により、JSONの成形部分で例外が発生すること Arranges: - オブジェクト情報の作成 - データの作成 - 実行日時取得インスタンスの生成 - オブジェクト情報インスタンスの生成 Expects: 例外が期待値と一致すること """ # Arranges object_info = { "object_name": "AccountShare", "columns": [ "Id", "AccountId", "UserOrGroupId", "AccountAccessLevel", "OpportunityAccessLevel", "CaseAccessLevel", "ContactAccessLevel", "RowCause", "LastModifiedDate", "LastModifiedById", "IsDeleted" ], "is_skip": False, "is_update_last_fetch_datetime": False, "last_fetch_datetime_file_name": "AccountShare.json", "upload_file_name": "CRM_AccountShare_{execute_datetime}", "datetime_column": "LastModifiedDate" } data = [ OrderedDict([ ('Id', 'TEST001'), ('AccountId', 'test001'), ('UserOrGroupId', None), ('AccountAccessLevel', 1), ('OpportunityAccessLevel', 2), ('CaseAccessLevel', 3), ('ContactAccessLevel', 4), ('RowCause', 'テストのため1'), ('LastModifiedDate', '2022-06-01T00:00:00.000+0000'), ('LastModifiedById', 1.234567E+6), ('IsDeleted', False) ]), OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST002'), ('AccountId', 'test002'), ('UserOrGroupId', None), ('AccountAccessLevel', 5), ('OpportunityAccessLevel', 6), ('CaseAccessLevel', 7), ('ContactAccessLevel', 8), ('RowCause', 'テストのため2'), ('LastModifiedDate', '2022-06-02T16:30:30.000+0000'), ('LastModifiedById', 2.234567E+6), ('IsDeleted', True) ]), OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST003'), ('AccountId', 'test003'), ('UserOrGroupId', None), ('AccountAccessLevel', 9), ('OpportunityAccessLevel', 10), ('CaseAccessLevel', 11), ('ContactAccessLevel', 12), ('RowCause', 'テストのため3'), ('LastModifiedDate', '2022-06-03T23:50:50.000+0000'), ('LastModifiedById', 3.234567E+6), ('IsDeleted', False) ]) ] execute_datetime = ExecuteDateTime() target_object = TargetObject(object_info, execute_datetime) # Act csv_string_converter = CSVStringConverter(target_object, data) with pytest.raises(Exception) as e: csv_string_converter.convert() # Expects assert '必要なjsonのデータ成形に失敗しました' in str(e.value) def test_raise_convert_convert_to_csv(self) -> str: """ Cases: 一部データのカラム（Id）の不足により、CSV変換で例外が発生すること Arranges: - オブジェクト情報の作成 - データの作成 - 実行日時取得インスタンスの生成 - オブジェクト情報インスタンスの生成 Expects: 例外が期待値と一致すること """ # Arranges object_info = { "object_name": "AccountShare", "columns": [ "Id", "AccountId", "UserOrGroupId", "AccountAccessLevel", "OpportunityAccessLevel", "CaseAccessLevel", "ContactAccessLevel", "RowCause", "LastModifiedDate", "LastModifiedById", "IsDeleted" ], "is_skip": False, "is_update_last_fetch_datetime": False, "last_fetch_datetime_file_name": "AccountShare.json", "upload_file_name": "CRM_AccountShare_{execute_datetime}", "datetime_column": "LastModifiedDate" } data = [ OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('AccountId', 'test001'), ('UserOrGroupId', None), ('AccountAccessLevel', 1), ('OpportunityAccessLevel', 2), ('CaseAccessLevel', 3), ('ContactAccessLevel', 4), ('RowCause', 'テストのため1'), ('LastModifiedDate', '2022-06-01T00:00:00.000+0000'), ('LastModifiedById', 1.234567E+6), ('IsDeleted', False) ]), OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST002'), ('AccountId', 'test002'), ('UserOrGroupId', None), ('AccountAccessLevel', 5), ('OpportunityAccessLevel', 6), ('CaseAccessLevel', 7), ('ContactAccessLevel', 8), ('RowCause', 'テストのため2'), ('LastModifiedDate', '2022-06-02T16:30:30.000+0000'), ('LastModifiedById', 2.234567E+6), ('IsDeleted', True) ]), OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST003'), ('AccountId', 'test003'), ('UserOrGroupId', None), ('AccountAccessLevel', 9), ('OpportunityAccessLevel', 10), ('CaseAccessLevel', 11), ('ContactAccessLevel', 12), ('RowCause', 'テストのため3'), ('LastModifiedDate', '2022-06-03T23:50:50.000+0000'), ('LastModifiedById', 3.234567E+6), ('IsDeleted', False) ]) ] execute_datetime = ExecuteDateTime() target_object = TargetObject(object_info, execute_datetime) # Act csv_string_converter = CSVStringConverter(target_object, data) with pytest.raises(Exception) as e: csv_string_converter.convert() # Expects assert 'CSV変換に失敗しました カラム名：[Id] 行番号: [1]' in str(e.value) def test_raise_convert_write_csv_string(self, monkeypatch) -> str: """ Cases: csvデータ出力の、CSVデータ取得で例外が発生すること Arranges: - オブジェクト情報の作成 - データの作成 - 実行日時取得インスタンスの生成 - オブジェクト情報インスタンスの生成 - csvデータ出力のエラーを発生させるためのモックを準備 Expects: 例外が期待値と一致すること """ # Arranges object_info = { "object_name": "AccountShare", "columns": [ "Id", "AccountId", "UserOrGroupId", "AccountAccessLevel", "OpportunityAccessLevel", "CaseAccessLevel", "ContactAccessLevel", "RowCause", "LastModifiedDate", "LastModifiedById", "IsDeleted" ], "is_skip": False, "is_update_last_fetch_datetime": False, "last_fetch_datetime_file_name": "AccountShare.json", "upload_file_name": "CRM_AccountShare_{execute_datetime}", "datetime_column": "LastModifiedDate" } data = [ OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST001'), ('AccountId', 'test001'), ('UserOrGroupId', None), ('AccountAccessLevel', 1), ('OpportunityAccessLevel', 2), ('CaseAccessLevel', 3), ('ContactAccessLevel', 4), ('RowCause', 'テストのため1'), ('LastModifiedDate', '2022-06-01T00:00:00.000+0000'), ('LastModifiedById', 1.234567E+6), ('IsDeleted', False) ]), OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST002'), ('AccountId', 'test002'), ('UserOrGroupId', None), ('AccountAccessLevel', 5), ('OpportunityAccessLevel', 6), ('CaseAccessLevel', 7), ('ContactAccessLevel', 8), ('RowCause', 'テストのため2'), ('LastModifiedDate', '2022-06-02T16:30:30.000+0000'), ('LastModifiedById', 2.234567E+6), ('IsDeleted', True) ]), OrderedDict([ ('attributes', OrderedDict([('type', 'AccountShare'), ('url', '/services/data/v1.0/sobjects/AccountShare/test1')])), ('Id', 'TEST003'), ('AccountId', 'test003'), ('UserOrGroupId', None), ('AccountAccessLevel', 9), ('OpportunityAccessLevel', 10), ('CaseAccessLevel', 11), ('ContactAccessLevel', 12), ('RowCause', 'テストのため3'), ('LastModifiedDate', '2022-06-03T23:50:50.000+0000'), ('LastModifiedById', 3.234567E+6), ('IsDeleted', False) ]) ] execute_datetime = ExecuteDateTime() target_object = TargetObject(object_info, execute_datetime) def dummy_method(arg): raise Exception(e) # データ加工のみだと事前の処理によりエラーとなるため、csv出力モジュールをモック化する monkeypatch.setattr("csv.writer", dummy_method) # Act csv_string_converter = CSVStringConverter(target_object, data) with pytest.raises(Exception) as e: csv_string_converter.convert() # Expects assert 'CSVデータの出力に失敗しました' in str(e.value)