43 lines
1.6 KiB
Python
43 lines
1.6 KiB
Python
import pytest
|
|
from src.aws.s3 import 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):
|
|
s3_client.put_object(Bucket=bucket_name, Key='hogehoge/test.txt', Body=b'aaaaaaaaaaaaaaa')
|
|
s3_resource = S3Resource(bucket_name)
|
|
actual = s3_resource.get_object('hogehoge/test.txt')
|
|
assert actual == 'aaaaaaaaaaaaaaa'
|
|
|
|
def test_put_object(self, s3_test, s3_client, bucket_name):
|
|
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() == b'aaaaaaaaaaaaaaa'
|
|
|
|
def test_copy(self, s3_test, s3_client, bucket_name):
|
|
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):
|
|
with pytest.raises(Exception) as e:
|
|
S3Resource()
|
|
assert e.value.args[0] == "__init__() missing 1 required positional argument: 'bucket_name'"
|