516 lines
20 KiB
Python
516 lines
20 KiB
Python
import os
|
|
|
|
import pytest
|
|
|
|
from src.aws.s3 import BackupBucket, ConfigBucket, DataBucket, S3Client
|
|
|
|
|
|
@pytest.fixture
|
|
def bucket_name():
|
|
return 'test-bucket'
|
|
|
|
|
|
@pytest.fixture
|
|
def s3_test(s3_client, bucket_name):
|
|
s3_client.create_bucket(Bucket=bucket_name)
|
|
yield
|
|
|
|
|
|
class TestS3Client:
|
|
|
|
def test_get_object(self, s3_test, s3_client, bucket_name):
|
|
"""
|
|
Cases:
|
|
- S3からオブジェクトが取得できるか
|
|
Arranges:
|
|
- S3をモック化する
|
|
- 期待値となるファイルを配置する
|
|
Expects:
|
|
- オブジェクトが取得でき、期待値と正しいこと
|
|
"""
|
|
# Arrange
|
|
s3_client.put_object(Bucket=bucket_name, Key='hogehoge/test.txt', Body=b'aaaaaaaaaaaaaaa')
|
|
|
|
# Act
|
|
sut = S3Client(bucket_name)
|
|
actual = sut.get_object('hogehoge/test.txt')
|
|
|
|
# Assert
|
|
assert actual == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_raise_get_object(self, s3_test, s3_client, bucket_name):
|
|
"""
|
|
Cases:
|
|
- オブジェクトのキーが存在しない場合エラーになること
|
|
Arranges:
|
|
- S3をモック化する
|
|
Expects:
|
|
- オブジェクト取得時にエラーになること
|
|
"""
|
|
# Arrange
|
|
# Act
|
|
sut = S3Client(bucket_name)
|
|
with pytest.raises(Exception):
|
|
# Assert
|
|
sut.get_object('hogehoge/test.txt')
|
|
|
|
def test_put_object(self, s3_test, s3_client, bucket_name, tmpdir):
|
|
"""
|
|
Cases:
|
|
- S3にオブジェクトをPUTできるか
|
|
Arranges:
|
|
- S3をモック化する
|
|
Expects:
|
|
- PUTされたファイルが存在する
|
|
- PUTされたファイルの内容が期待値と一致する
|
|
"""
|
|
|
|
file_path = os.path.join(tmpdir, 'test.txt')
|
|
with open(file_path, mode='w') as f:
|
|
f.write('aaaaaaaaaaaaaaa')
|
|
|
|
sut = S3Client(bucket_name)
|
|
|
|
sut.put_object('hogehoge/test.txt', file_path)
|
|
actual = s3_client.get_object(Bucket=bucket_name, Key='hogehoge/test.txt')
|
|
|
|
assert actual['Body'].read().decode('utf-8') == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_raise_put_object(self, s3_client, bucket_name):
|
|
"""
|
|
Cases:
|
|
- バケットが存在しない場合エラーになること
|
|
Arranges:
|
|
- S3をモック化する
|
|
- バケットを作成しない
|
|
Expects:
|
|
- オブジェクト登録時にエラーになること
|
|
"""
|
|
# Arrange
|
|
# Act
|
|
sut = S3Client(bucket_name)
|
|
with pytest.raises(Exception):
|
|
# Assert
|
|
sut.put_object('hogehoge/test.txt', 'aaaaaaaaaaaaaaa')
|
|
|
|
def test_copy(self, s3_test, s3_client, bucket_name):
|
|
"""
|
|
Cases:
|
|
- S3内のオブジェクトを別バケットにコピーできるか
|
|
Arranges:
|
|
- S3をモック化する
|
|
- コピー先バケットを作成する
|
|
- 期待値となるファイルをコピー元バケットに配置する
|
|
Expects:
|
|
- コピーされたファイルが存在する
|
|
- コピーされたファイルの内容が期待値と一致する
|
|
"""
|
|
for_copy_bucket = 'for-copy-bucket'
|
|
s3_client.create_bucket(Bucket=for_copy_bucket)
|
|
s3_client.put_object(Bucket=bucket_name, Key='hogehoge/test.txt', Body=b'aaaaaaaaaaaaaaa')
|
|
|
|
sut = S3Client(bucket_name)
|
|
sut.copy(bucket_name, 'hogehoge/test.txt', for_copy_bucket, 'test.txt')
|
|
|
|
actual = s3_client.get_object(Bucket=for_copy_bucket, Key='test.txt')
|
|
assert actual['Body'].read() == b'aaaaaaaaaaaaaaa'
|
|
|
|
def test_raise_copy(self, s3_client, s3_test, bucket_name):
|
|
"""
|
|
Cases:
|
|
- コピー対象のキーが存在しない場合エラーになること
|
|
Arranges:
|
|
- S3をモック化する
|
|
Expects:
|
|
- オブジェクト登録時にエラーになること
|
|
"""
|
|
# Arrange
|
|
# Act
|
|
sut = S3Client(bucket_name)
|
|
with pytest.raises(Exception):
|
|
# Assert
|
|
sut.copy(bucket_name, 'hogehoge/test.txt', 'for_copy_bucket', 'test.txt')
|
|
|
|
def test_raise_init_no_provide_bucket_name(self):
|
|
"""
|
|
Cases:
|
|
- バケット名を指定しない場合、例外となること
|
|
Arranges:
|
|
|
|
Expects:
|
|
- インスタンス生成時に例外が発生すること
|
|
"""
|
|
with pytest.raises(Exception) as e:
|
|
S3Client()
|
|
assert e.value.args[0] == "S3Client.__init__() missing 1 required positional argument: 'bucket_name'"
|
|
|
|
|
|
class TestConfigBucket:
|
|
|
|
def test_get_object_info_file(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- オブジェクト情報ファイルが取得できること
|
|
Arranges:
|
|
- 環境変数を置き換える
|
|
- オブジェクト情報ファイルを配置する
|
|
Expects:
|
|
- オブジェクト情報ファイルが文字列として取得でき、期待値と一致する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.OBJECT_INFO_FOLDER', 'crm')
|
|
monkeypatch.setattr('src.aws.s3.OBJECT_INFO_FILENAME', 'objects.json')
|
|
s3_client.put_object(Bucket=bucket_name, Key=f'crm/objects.json', Body=b'aaaaaaaaaaaaaaa')
|
|
|
|
sut = ConfigBucket()
|
|
actual = sut.get_object_info_file()
|
|
|
|
assert actual == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_raise_get_object_info_file(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- オブジェクト情報ファイルが存在しない場合、エラーになること
|
|
Arranges:
|
|
- s3_testフィクスチャを引数に指定しない
|
|
- 環境変数を置き換える
|
|
Expects:
|
|
- オブジェクト情報ファイルが取得できず、例外が発生する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.OBJECT_INFO_FOLDER', 'crm')
|
|
monkeypatch.setattr('src.aws.s3.OBJECT_INFO_FILENAME', 'objects.json')
|
|
|
|
sut = ConfigBucket()
|
|
with pytest.raises(Exception):
|
|
sut.get_object_info_file()
|
|
|
|
def test_get_last_fetch_datetime_file(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- オブジェクト最終更新日時ファイルが取得できること
|
|
Arranges:
|
|
- S3をモック化する
|
|
- オブジェクト最終更新日時ファイルを配置する
|
|
Expects:
|
|
- オブジェクト最終更新日時ファイルが文字列として取得でき、期待値と一致する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.LAST_FETCH_DATE_FOLDER', 'crm')
|
|
s3_client.put_object(Bucket=bucket_name, Key=f'crm/Object.json', Body=b'aaaaaaaaaaaaaaa')
|
|
|
|
sut = ConfigBucket()
|
|
actual = sut.get_last_fetch_datetime_file('Object.json')
|
|
|
|
assert actual == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_raise_get_last_fetch_datetime_file(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- オブジェクト最終更新日時ファイルが存在しない場合、エラーになること
|
|
Arranges:
|
|
- s3_testフィクスチャを引数に指定しない
|
|
- 環境変数を置き換える
|
|
Expects:
|
|
- オブジェクト最終更新日時ファイルが取得できず、例外が発生する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.LAST_FETCH_DATE_FOLDER', 'crm')
|
|
|
|
sut = ConfigBucket()
|
|
with pytest.raises(Exception):
|
|
sut.get_last_fetch_datetime_file('Object.json')
|
|
|
|
def test_put_last_fetch_datetime_file(self, s3_test, s3_client, bucket_name, monkeypatch, tmpdir):
|
|
"""
|
|
Cases:
|
|
- オブジェクト最終更新日時ファイルをPUTできること
|
|
Arranges:
|
|
- S3をモック化する
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- オブジェクト最終更新日時ファイルが存在する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.LAST_FETCH_DATE_FOLDER', 'crm')
|
|
|
|
file_path = os.path.join(tmpdir, 'Object.json')
|
|
with open(file_path, mode='w') as f:
|
|
f.write('aaaaaaaaaaaaaaa')
|
|
|
|
sut = ConfigBucket()
|
|
sut.put_last_fetch_datetime_file('Object.json', file_path)
|
|
|
|
actual = s3_client.get_object(Bucket=bucket_name, Key=f'crm/Object.json')
|
|
assert actual['Body'].read().decode('utf-8') == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_raise_put_last_fetch_datetime_file(self, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- オブジェクト最終更新日時ファイルをPUTできない場合、エラーになること
|
|
Arranges:
|
|
- s3_testフィクスチャを引数に指定しない
|
|
- 環境変数を置き換える
|
|
Expects:
|
|
- オブジェクト最終更新日時ファイルをPUTできず、例外が発生する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.LAST_FETCH_DATE_FOLDER', 'crm')
|
|
|
|
sut = ConfigBucket()
|
|
with pytest.raises(Exception):
|
|
sut.put_last_fetch_datetime_file('Object.json', 'aaaaaaaaaaaaaaa')
|
|
|
|
def test_config_bucket_str(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- 設定ファイル配置バケットを文字列化したとき、バケット名が取得できること
|
|
Arranges:
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- 環境変数の設定ファイル配置バケット名と一致する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name)
|
|
|
|
sut = ConfigBucket()
|
|
|
|
assert str(sut) == bucket_name
|
|
|
|
|
|
class TestDataBucket:
|
|
|
|
def test_put_csv(self, s3_test, s3_client, bucket_name, monkeypatch, tmpdir):
|
|
"""
|
|
Cases:
|
|
- CSVファイルをPUTできること
|
|
Arranges:
|
|
- S3をモック化する
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- PUTしたファイルが存在する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.IMPORT_DATA_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.CRM_IMPORT_DATA_FOLDER', 'crm/target')
|
|
|
|
file_path = os.path.join(tmpdir, 'test.json')
|
|
with open(file_path, mode='w') as f:
|
|
f.write('test,test,test')
|
|
|
|
sut = DataBucket()
|
|
sut.put_csv('test.csv', file_path)
|
|
|
|
actual = s3_client.get_object(Bucket=bucket_name, Key=f'crm/target/test.csv')
|
|
assert actual['Body'].read().decode('utf-8') == 'test,test,test'
|
|
|
|
def test_raise_put_csv(self, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- CSVファイルをPUTできず、エラーになること
|
|
Arranges:
|
|
- s3_testフィクスチャを引数に指定しない
|
|
- 環境変数を置き換える
|
|
Expects:
|
|
- CSVファイルをPUTできず、例外が発生する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.IMPORT_DATA_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.CRM_IMPORT_DATA_FOLDER', 'crm/target')
|
|
|
|
sut = DataBucket()
|
|
with pytest.raises(Exception):
|
|
sut.put_csv('test.csv', 'test,test,test')
|
|
|
|
def test_put_csv_from(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- 他のバケットからCSVファイルをコピーできること
|
|
Arranges:
|
|
- S3をモック化する
|
|
- コピー先バケットを作成する
|
|
- 期待値となるファイルコピー元バケットに配置する
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- コピーしたファイルが存在する
|
|
"""
|
|
for_copy_bucket = 'for-copy-bucket'
|
|
s3_client.create_bucket(Bucket=for_copy_bucket)
|
|
s3_client.put_object(Bucket=bucket_name, Key='hogehoge/test.csv', Body=b'test,test,test')
|
|
|
|
monkeypatch.setattr('src.aws.s3.IMPORT_DATA_BUCKET', for_copy_bucket)
|
|
monkeypatch.setattr('src.aws.s3.CRM_IMPORT_DATA_FOLDER', 'crm/target')
|
|
|
|
sut = DataBucket()
|
|
sut.put_csv_from(bucket_name, 'hogehoge/test.csv')
|
|
actual = s3_client.get_object(Bucket=for_copy_bucket, Key=f'crm/target/test.csv')
|
|
|
|
assert actual['Body'].read().decode('utf-8') == 'test,test,test'
|
|
|
|
def test_raise_put_csv_from(self, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- 他のバケットからCSVファイルをコピーできず、エラーになること
|
|
Arranges:
|
|
- s3_testフィクスチャを引数に指定しない
|
|
- 環境変数を置き換える
|
|
Expects:
|
|
- コピーできず、例外が発生する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.IMPORT_DATA_BUCKET', 'for_copy_bucket')
|
|
monkeypatch.setattr('src.aws.s3.CRM_IMPORT_DATA_FOLDER', 'crm/target')
|
|
|
|
sut = DataBucket()
|
|
with pytest.raises(Exception):
|
|
sut.put_csv_from(bucket_name, 'hogehoge/test.csv')
|
|
|
|
def test_data_bucket_str(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- データ登録バケットを文字列化したとき、バケット名が取得できること
|
|
Arranges:
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- 環境変数のデータ登録バケット名と一致する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.IMPORT_DATA_BUCKET', bucket_name)
|
|
|
|
sut = DataBucket()
|
|
|
|
assert str(sut) == bucket_name
|
|
|
|
|
|
class TestBackupBucket:
|
|
|
|
def test_put_csv(self, s3_test, s3_client, bucket_name, monkeypatch, tmpdir):
|
|
"""
|
|
Cases:
|
|
- CSVファイルをPUTできること
|
|
Arranges:
|
|
- S3をモック化する
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- PUTしたファイルが存在する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_BACKUP_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.CRM_IMPORT_DATA_BACKUP_FOLDER', 'data_import')
|
|
|
|
file_path = os.path.join(tmpdir, 'test.json')
|
|
with open(file_path, mode='w') as f:
|
|
f.write('test,test,test')
|
|
|
|
sut = BackupBucket()
|
|
sut.put_csv('test.csv', file_path)
|
|
|
|
actual = s3_client.get_object(Bucket=bucket_name, Key=f'data_import/test.csv')
|
|
assert actual['Body'].read().decode('utf-8') == 'test,test,test'
|
|
|
|
def test_raise_put_csv_from(self, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- CSVファイルをPUTできず、エラーになること
|
|
Arranges:
|
|
- s3_testフィクスチャを引数に指定しない
|
|
- 環境変数を置き換える
|
|
Expects:
|
|
- CSVファイルがPUTできず、例外が発生する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.IMPORT_DATA_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.CRM_IMPORT_DATA_FOLDER', 'crm/target')
|
|
|
|
sut = BackupBucket()
|
|
with pytest.raises(Exception):
|
|
sut.put_csv('test.csv', 'test,test,test')
|
|
|
|
def test_put_response_json(self, s3_test, s3_client, bucket_name, monkeypatch, tmpdir):
|
|
"""
|
|
Cases:
|
|
- JSONファイルをPUTできること
|
|
Arranges:
|
|
- S3をモック化する
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- PUTしたファイルが存在する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_BACKUP_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.RESPONSE_JSON_BACKUP_FOLDER', 'response_json')
|
|
|
|
file_path = os.path.join(tmpdir, 'test.json')
|
|
with open(file_path, mode='w') as f:
|
|
f.write('{"test": "test"}')
|
|
|
|
sut = BackupBucket()
|
|
sut.put_response_json('test.json', file_path)
|
|
|
|
actual = s3_client.get_object(Bucket=bucket_name, Key=f'response_json/test.json')
|
|
assert actual['Body'].read().decode('utf-8') == '{"test": "test"}'
|
|
|
|
def test_raise_response_json(self, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- JSONファイルをPUTできず、エラーになること
|
|
Arranges:
|
|
- s3_testフィクスチャを引数に指定しない
|
|
- 環境変数を置き換える
|
|
Expects:
|
|
- JSONファイルをPUTできず、例外が発生する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_BACKUP_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.RESPONSE_JSON_BACKUP_FOLDER', 'response_json')
|
|
|
|
sut = BackupBucket()
|
|
with pytest.raises(Exception):
|
|
sut.put_response_json('test.json', {"test": "test"})
|
|
|
|
def test_put_result_json(self, s3_test, s3_client, bucket_name, monkeypatch, tmpdir):
|
|
"""
|
|
Cases:
|
|
- 結果のJSONファイルをPUTできること
|
|
Arranges:
|
|
- S3をモック化する
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- PUTしたファイルが存在する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_BACKUP_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.PROCESS_RESULT_FOLDER', 'data_import')
|
|
|
|
file_path = os.path.join(tmpdir, 'test.json')
|
|
with open(file_path, mode='w') as f:
|
|
f.write('{"test": "test"}')
|
|
|
|
sut = BackupBucket()
|
|
sut.put_result_json('result.json', file_path)
|
|
|
|
actual = s3_client.get_object(Bucket=bucket_name, Key=f'data_import/result.json')
|
|
assert actual['Body'].read().decode('utf-8') == '{"test": "test"}'
|
|
|
|
def test_raise_result_json(self, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- 結果のJSONファイルをPUTできず、エラーになること
|
|
Arranges:
|
|
- s3_testフィクスチャを引数に指定しない
|
|
- 環境変数を置き換える
|
|
Expects:
|
|
- 結果のJSONファイルをPUTできず、例外が発生する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_BACKUP_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.PROCESS_RESULT_FOLDER', 'data_import')
|
|
|
|
sut = BackupBucket()
|
|
with pytest.raises(Exception):
|
|
sut.put_result_json('result.json', {"test": "test"})
|
|
|
|
def test_backup_bucket_str(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
- CRMデータバックアップバケットを文字列化したとき、バケット名が取得できること
|
|
Arranges:
|
|
- 環境変数をテスト用の値に置き換える
|
|
Expects:
|
|
- 環境変数のCRMデータバックアップバケット名と一致する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_BACKUP_BUCKET', bucket_name)
|
|
|
|
sut = BackupBucket()
|
|
|
|
assert str(sut) == bucket_name
|