160 lines
5.8 KiB
Python
160 lines
5.8 KiB
Python
import pytest
|
|
from src.aws.s3 import ConfigBucket, S3Resource
|
|
|
|
|
|
@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 TestS3Resource:
|
|
|
|
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')
|
|
|
|
# ActAssert
|
|
s3_resource = S3Resource(bucket_name)
|
|
actual = s3_resource.get_object('hogehoge/test.txt')
|
|
|
|
# Assert
|
|
assert actual == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_put_object(self, s3_test, s3_client, bucket_name):
|
|
"""
|
|
Cases:
|
|
S3にオブジェクトをPUTできるか
|
|
Arranges:
|
|
- S3をモック化する
|
|
Expects:
|
|
オブジェクトがPUTできること
|
|
"""
|
|
s3_resource = S3Resource(bucket_name)
|
|
|
|
s3_resource.put_object('hogehoge/test.txt', 'aaaaaaaaaaaaaaa')
|
|
actual = s3_client.get_object(Bucket=bucket_name, Key='hogehoge/test.txt')
|
|
|
|
assert actual['Body'].read().decode('utf-8') == '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')
|
|
|
|
s3_resource = S3Resource(bucket_name)
|
|
s3_resource.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_init_raise_no_provide_bucket_name(self):
|
|
"""
|
|
Cases:
|
|
バケット名を指定しない場合、例外となること
|
|
Arranges:
|
|
|
|
Expects:
|
|
例外が発生すること
|
|
"""
|
|
with pytest.raises(Exception) as e:
|
|
S3Resource()
|
|
assert e.value.args[0] == "__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')
|
|
|
|
config_bucket = ConfigBucket()
|
|
actual = config_bucket.get_object_info_file()
|
|
|
|
assert actual == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_get_last_fetch_datetime_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.LAST_FETCH_DATE_FOLDER', 'crm')
|
|
s3_client.put_object(Bucket=bucket_name, Key=f'crm/Object.json', Body=b'aaaaaaaaaaaaaaa')
|
|
|
|
config_bucket = ConfigBucket()
|
|
actual = config_bucket.get_last_fetch_datetime_file('Object.json')
|
|
|
|
assert actual == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_put_last_fetch_datetime_file(self, s3_test, s3_client, bucket_name, monkeypatch):
|
|
"""
|
|
Cases:
|
|
オブジェクト最終更新日時ファイルをPUTできること
|
|
Arranges:
|
|
|
|
Expects:
|
|
オブジェクト最終更新日時ファイルが存在する
|
|
"""
|
|
monkeypatch.setattr('src.aws.s3.CRM_CONFIG_BUCKET', bucket_name)
|
|
monkeypatch.setattr('src.aws.s3.LAST_FETCH_DATE_FOLDER', 'crm')
|
|
|
|
config_bucket = ConfigBucket()
|
|
config_bucket.put_last_fetch_datetime_file('Object.json', 'aaaaaaaaaaaaaaa')
|
|
|
|
actual = s3_client.get_object(Bucket=bucket_name, Key=f'crm/Object.json')
|
|
assert actual['Body'].read().decode('utf-8') == '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)
|
|
|
|
config_bucket = ConfigBucket()
|
|
|
|
assert str(config_bucket) == bucket_name
|