python - Mocking integrated AWS s3 function with moto and fastapi
问题描述
I'm trying to mock some AWS functionalities that are integrated in a post request function. Initially I tried with python mock lib. However, the calls are still made to AWS such as the two way authentication.
The mock lib code looks like this.
Test file
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"}
Server file with post
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 file
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
Im really looking for a solution which works for integrated s3 functions, preferably with moto. Thanks!
解决方案
Adding the mock decorator and initializing the client, resource and bucket outside of the function did the trick. Beware of the bucket name! Here is the part of the code that made it work.
@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
推荐阅读
- angular - 可观察对象过滤可观察对象并映射到不同的可观察对象:Angular 9
- android-studio - Android Studio Widget 更新更快
- python - 如何在循环中使用 to_json 为每个键、值对创建 json,其中键是 json 文件名,值是要转换为 json 的数据帧?
- c# - 父控件绑定命令到 ContextMenu MenuItem
- c# - 如何以最佳方式在 Unity 中实例化 UI 图像的静态网格?
- r - 替换空(NA)数据框中的行而不将因子转换为整数
- node.js - Neo4jError:消息“提交”不能由处于就绪状态的会话处理
- android - 如何在 Android Studio 中解决这个问题
- c++ - 将二进制(exe、zip 等)文件读入 char*、c++
- javascript - 为什么这个 forEach 映射是未定义的?