I want to test conduct testing on imported files from aws. I mock s3 using moto, in order to not mess with actual data. However, now the aws seems to be empty, thus I decided to upload some test file on mocked s3. How can I do so?
This is my setup,
Conftest.py:
@pytest.fixture(scope='function')
def aws_credentials():
"""Mocked AWS Credentials for moto."""
os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
os.environ['AWS_SECURITY_TOKEN'] = 'testing'
os.environ['AWS_SESSION_TOKEN'] = 'testing'
@pytest.fixture(scope='function')
def s3(aws_credentials):
with mock_s3():
yield boto3.client('s3', region_name='us-east-1')
Test file:
class TestLoadRecommendations:
@pytest.fixture(autouse=True)
def _setup(self, s3):
self.bucket = s3.create_bucket(Bucket=settings.SERVER_DATA_S3_BUCKET)
self.bucket.upload_file("tmp/recommendations-2021-04-13T17_28_06.csv", "tmp/recommendations-2021-04-13T17_28_06.csv")
However, instead uploading it throws an error TypeError: expected string or bytes-like object
but I am sure that I use incorrect command for file upload.
Could anybody help? Thanks!
There's multiple ways of uploading a file to S3. Your example has a combination of the S3 resource and S3 client methods, which will not work.
See the following code for an example of:
All three ways lead to Rome.
import boto3
from moto import mock_s3
BUCKET_NAME = "mybucket"
FILE_NAME = "red.jpg"
FILE_LOCATION = FILE_NAME
@mock_s3
def test_client():
create_bucket()
s3 = boto3.client('s3')
with open(FILE_LOCATION, 'rb') as data:
s3.upload_fileobj(data, BUCKET_NAME, FILE_NAME)
verify_upload()
@mock_s3
def test_resource():
s3_resource, _ = create_bucket()
s3_resource.meta.client.upload_file(FILE_LOCATION, BUCKET_NAME, FILE_NAME)
#
verify_upload()
@mock_s3
def test_bucket_resource():
_, bucket = create_bucket()
bucket.upload_file(FILE_LOCATION, FILE_NAME)
#
verify_upload()
def verify_upload():
client = boto3.client("s3")
resp = client.get_object(Bucket=BUCKET_NAME, Key=FILE_NAME)
content_length = resp["ResponseMetadata"]["HTTPHeaders"]["content-length"]
print("Content-Length: {}".format(content_length))
def create_bucket():
s3 = boto3.resource("s3")
bucket = s3.create_bucket(Bucket=BUCKET_NAME)
return s3, bucket
Note: I'm using the decorators, but these examples will work exactly the same using Moto fixtures.