模拟测试 S3 上传的 python 功能

Mock testing a python functionality for S3 Upload

我想模拟下面的书面功能。为此,我不确定是否要为此添加单元测试。关于在不使用 unittest 库的情况下编写可能的单元测试有什么建议吗?非常感谢!

def file_upload(self, upload_file_bucket, file_name, file_path):
        if os.path.exists(file_path):
            with open(file_path, 'r') as f:
                xml = f.read()
        else:
            logging.error("File '%s' does not exist." % file_path)
            tools.exit_gracefully(botocore.log)
        try:
            conn = boto3.session.Session(profile_name=aws_prof_dev_qa)
            s3 = conn.resource('s3')
            object = s3.Object(upload_file_bucket, file_name)
            result = object.put(Body=xml)
            res = result.get('ResponseMetadata')
            if res.get('HTTPStatusCode') == 200:
                logging.info('File Uploaded Successfully')
            else:
                logging.info('File Not Uploaded Successfully')
            return res
        except ClientError as e:
            logging.error(e)

您可以使用 moto.

编写可靠的 end-to-end 测试

Moto 是测试 boto 的最佳实践。它在您的本地机器上模拟 boto,在本地创建存储桶,这样您就可以进行完整的 end-to-end 测试。

这是您的方法的一些示例代码(我更改了您的代码以使用配置文件名称 'aws_prof_dev_qa')。我使用 pytest fixtures 来组织我的代码,但这不是强制性的:

@pytest.fixture(scope='session')
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'
    os.environ['AWS_DEFAULT_REGION'] = 'us-east-1'

    try:
        tmp = NamedTemporaryFile(delete=False)
        # you many need to change 'aws_prof_dev_qa' to be your profile name
        tmp.write(b"""[aws_prof_dev_qa]
aws_access_key_id = testing
aws_secret_access_key = testing""")
        tmp.close()
        os.environ['AWS_SHARED_CREDENTIALS_FILE'] = str(tmp.name)
        yield
    finally:
        os.unlink(tmp.name)


@pytest.fixture(scope='function')
def empty_bucket(aws_credentials):
    moto_fake = moto.mock_s3()
    try:
        moto_fake.start()
        conn = boto3.resource('s3')
        conn.create_bucket(Bucket="MY_BUCKET")  # or the name of the bucket you use
        yield conn
    finally:
        moto_fake.stop()


def test_file_upload(empty_bucket):
    with NamedTemporaryFile() as tmp:
        tmp.write(b'Hi')
        file_name = pathlib.Path(tmp.name).name

        result = file_upload("MY_BUCKET", file_name, tmp.name)

        assert result.get('HTTPStatusCode') == 200

有关为什么 moto 比模拟更好的更多详细信息,see my lecture