upload a test file using moto and boto3

Question

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!

Answer 1

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:

• S3-client - upload_fileobj
• S3-resource - upload_file
• Bucket-resource - upload_file

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.

Answer 2

In case someone is using a conftest.py file to create test artifacts, this is my solution.

Note:

1. I also show how to upload if you have your file in a folder in S3 bucket

2. My folder structure is: common tests folder at root.

3. Moto version: moto==4.2.3

       @pytest.fixture(autouse=True)
       def s3_bucket():
           """This fixture creates a local s3 bucket tests"""
           with moto.mock_s3():
               s3 = boto3.resource("s3", region_name=default_location)
               s3_resource = s3.create_bucket(
                   Bucket=os.environ["PRODUCT_IMG_BUCKET_NAME"],
                   CreateBucketConfiguration={"LocationConstraint": default_location},
               )
               s3.meta.client.head_bucket(
                   Bucket=os.environ["PRODUCT_IMG_BUCKET_NAME"]
               )
               my_bucket = s3.Bucket(
                   os.environ["PRODUCT_IMG_BUCKET_NAME"]
               )
   
               test_upload_file_path = str(Path(__file__).parent / "MyFolder" / "1.png")
   
               # upload a file into the bucket.
               s3_resource.meta.client.upload_file(
                   test_upload_file_path,
                   os.environ["PRODUCT_IMG_BUCKET_NAME"],
                   "1.png",
               )
               verify_upload(
                   os.environ["PRODUCT_IMG_BUCKET_NAME"],
                   "1.png",
               )
               yield my_bucket
   
       def verify_upload(bucket_name, file_name):
           client = boto3.client("s3")
           resp = client.get_object(Bucket=bucket_name, Key=file_name)
           content_length = resp["ResponseMetadata"]["HTTPHeaders"]["content-length"]