使用 moto 和 fastapi 模拟集成的 AWS s3 功能答案

【问题标题】：Mocking integrated AWS s3 function with moto and fastapi使用 moto 和 fastapi 模拟集成的 AWS s3 功能
【发布时间】：2021-08-17 13:53:24
【问题描述】：

我正在尝试模拟一些集成在发布请求功能中的 AWS 功能。最初我尝试使用 python 模拟库。但是，调用仍然是对 AWS 进行的，例如双向身份验证。

模拟库代码如下所示。

测试文件

from unittest import mock
from fastapi.testclient import TestClient

@mock.patch("utils.io.aws.upload_audio_to_s3", return_value="test_1.wav")
def test_dummy_code(process):
    with open("tests/files/test.wav", "rb") as f:
        values = {"language_code": "nl-EN"}
        response = client.post("/mock/test", files={"file": f}, data=values)
    assert response.json() == {"status": "test_1.wav"}

带有帖子的服务器文件

from fastapi import APIRouter, File, Form, UploadFile
from utils.io.aws import upload_audio_to_s3

router = APIRouter()

@router.post("/mock/test")
async def get_mocking(file: UploadFile = File(...)):
    print(file)
    print("Files recieved")

    # s3_file_path = process_file(file)
    s3_file_path = upload_audio_to_s3(
        "test.wav", SOUND_FILE_EXTENSION, S3_FOLDER_NAME, BUCKET_NAME
    )

    print(s3_file_path)
    return {"status": s3_file_path}

aws 文件

import boto3

def connect_to_s3(region_name=REGION):
    s3_client = boto3.client("s3", region_name=region_name)
    s3_resource = boto3.resource("s3", region_name=region_name)

    # if not _validate_access(s3_client):
    #    # logging.info("Access denied. Trying with MFA")
    #    s3_client, s3_resource = _connect_to_s3_mfa()

    return s3_client, s3_resource


def upload_file_to_s3(
    file_path: str,
    filename: str,
    directory: str = None,
    bucket_name: str = "audio",
    s3_resource: boto3.resource = None,
):
    if not directory:
        s3_file_path = f"{filename}"
    else:
        s3_file_path = f"{directory}/{filename}"

    obj = s3_resource.Bucket(bucket_name)
    obj.upload_file(file_path, s3_file_path)
    return s3_file_path


def upload_audio_to_s3(audio_file_path, file_ext, folder_name, bucket_name):
    s3_client, s3_resource = connect_to_s3()

    random_file_name = create_random_file("temp")
    s3_file_path = f"{random_file_name}{file_ext}"
    s3_folder = f"{folder_name}/batch"

    s3_file_path = upload_file_to_s3(
        audio_file_path, s3_file_path, s3_folder, bucket_name, s3_resource
    )

    return s3_file_path

我真的在寻找适用于集成 s3 功能的解决方案，最好用moto。谢谢！

【问题讨论】：

标签： python unit-testing amazon-s3 fastapi moto

【解决方案1】：

添加模拟装饰器并在函数之外初始化客户端、资源和存储桶就可以了。请注意存储桶名称！这是使它工作的代码部分。

@mock_s3
def test_dummy_code():
    conn, resource = connect_to_s3('us-east-1')
    
    # We need to create the bucket since this is all in Moto's 'virtual' AWS account
    conn.create_bucket(Bucket=BUCKET_NAME)

    with open("tests/files/test.wav", "rb") as f:
        values = {"language_code": "nl-NL"}
        response = client.post("/mock/test", files={"file": f}, data=values)    
    
    assert response.status_code == 200

【讨论】：