如何为FastAPI+CosmosDB的Azure函数编写Pytest单元测试？

我有一段结合FastAPI、Azure Functions与CosmosDB的后端代码，已完成功能测试，但编写Pytest单元测试时遇到阻碍，请求技术协助。

后端源码 #

import logging
import json
import azure.functions as func
from azure.cosmos import CosmosClient
import fastapi
import os

app = fastapi.FastAPI()
client = CosmosClient.from_connection_string(os.environ["CosmosDBConnStr"])

database_name = "dmdb"
container_name = "DataContract"

database = client.get_database_client(database_name)
container = database.get_container_client(container_name)
  

@app.get("/marketplace/{V}")
async def get_data_by_version(V: str):
    logging.info("all data contracts are coming....")
    query = f''' SELECT c.name,c.version, c.Theme, c.title, c.description, udf.tsToDate(c._ts) as updated_at
                FROM c 
                WHERE c.version = '{V}'
                ORDER BY c._ts DESC
                OFFSET 0 LIMIT 20
               '''
 
    items = list(container.query_items(query, enable_cross_partition_query=True))
    logging.info(f"items: {items}")
    response_body = json.dumps(items)
    return func.HttpResponse(body=response_body, mimetype="application/json")


@app.get("/marketplace/{V}/themes")
async def get_existing_themes_by_version(V: str):
    query = f"SELECT c.Theme FROM c WHERE c.version = '{V}'"
    items = list(container.query_items(query, enable_cross_partition_query=True))
    unique_themes = list(set(item["Theme"] for item in items))
    response_body = json.dumps(unique_themes)
    return func.HttpResponse(body=response_body, mimetype="application/json")

@app.get("/marketplace/{V}/{id}")
async def get_data_by_version_and_id(V: str, id: str):
   
    query = f"SELECT * FROM c WHERE c.version = '{V}' AND c.id = '{id}'"
    items = list(container.query_items(query, enable_cross_partition_query=True))
    
    if not items:
        return func.HttpResponse(
            f"No item with version '{V}' and id '{id}' was found",
            status_code=404
        )
    
    response_body = json.dumps(items[0])
    return func.HttpResponse(body=response_body, mimetype="application/json")


async def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    return await func.AsgiMiddleware(app).handle_async(req, context)

各接口实际返回示例 #

• 全量摘要数据接口http://localhost:7071/marketplace/V1返回：

{"_HttpResponse__status_code":200,"_HttpResponse__mimetype":"application/json","_HttpResponse__charset":"utf-8","_HttpResponse__headers":{},"_HttpResponse__body":"[{\"name\": \"demo_contrac1\", \"version\": \"V1\", \"Theme\": \"Theme1\", \"title\": \"title1\", \"description\": \"test data contract management2\", \"updated_at\": \"2023-06-02T17:06:44.000Z\"}]"}

• 主题列表接口http://localhost:7071/marketplace/V1/themes返回：

{"_HttpResponse__status_code":200,"_HttpResponse__mimetype":"application/json","_HttpResponse__charset":"utf-8","_HttpResponse__headers":{},"_HttpResponse__body":"[\"Theme1\"]"}

• 按ID查询接口http://localhost:7071/marketplace/V1/123返回：

{"_HttpResponse__status_code":200,"_HttpResponse__mimetype":"application/json","_HttpResponse__charset":"utf-8","_HttpResponse__headers":{},"_HttpResponse__body":"{\"version\": \"V1\", \"name\": \"demo_contrac1\", \"title\": \"title1\", \"Theme\": \"Theme1\", \"description\": \"test data contract management2\", \"data owner\": \"j.jansen@amsterdam.nl\", \"confidentiality\": \"open\", \"table1\": {\"description:\": \"testen\", \"attribute_1\": {\"type\": \"int\", \"description:\": \"testen\", \"identifiability\": \"identifiable\"}}, \"id\": \"f895357f-efe0-4e64-a3b4-868f54e266c1\", \"_rid\": \"R69IAIfCVFQ7AAAAAAAAAA==\", \"_self\": \"dbs/R69IAA==/colls/R69IAIfCVFQ=/docs/R69IAIfCVFQ7AAAAAAAAAA==/\", \"_etag\": \"\\\"00000000-0000-0000-9574-9bf76b6101d9\\\"\", \"_attachments\": \"attachments/\", \"_ts\": 1685725604}"}

内容的提问来源于stack exchange，提问作者aca1803