使用 moto 和 boto3 上传测试文件答案

【问题标题】：upload a test file using moto and boto3使用 moto 和 boto3 上传测试文件
【发布时间】：2021-04-15 08:04:45
【问题描述】：

我想对从 aws 导入的文件进行测试。我使用 moto 模拟 s3，以免弄乱实际数据。但是，现在 aws 似乎是空的，因此我决定在 mocked s3 上上传一些测试文件。我该怎么做？

这是我的设置，
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')

测试文件：

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")

但是，上传它会引发错误TypeError: expected string or bytes-like object，但我确信我使用了不正确的命令进行文件上传。有人可以帮忙吗？谢谢！

【问题讨论】：

你从哪里得到那个 TypeError？
s3.create_bucket 不返回桶，它返回一个响应对象。您需要在 s3 客户端上调用 upload_file，而不是在存储桶上：s3.upload_file(...)

标签： django amazon-web-services pytest boto3 moto

【解决方案1】：

有多种方法可以将文件上传到 S3。您的示例结合了 S3 资源和 S3 客户端方法，这将不起作用。

查看以下代码示例：

S3 客户端 - upload_fileobj
S3 资源 - 上传文件
存储桶资源 - 上传文件

所有三种方式都通向罗马。

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

注意：我使用的是装饰器，但这些示例在使用 Moto 固定装置时的工作方式完全相同。

【讨论】：