如何为 Durable Azure Functions 编写单元测试？

Question

我正在编写一个 Azure 持久函数，我想为整个 Azure 函数编写一些单元测试。

我试图触发客户端功能（“开始”功能，因为它通常被称为），但我无法使其工作。

我这样做有两个原因：

1. 通过运行“func host start”（或按 F5）来运行 Azure 函数代码令人沮丧，然后转到我的浏览器，找到正确的选项卡，转到 http://localhost:7071/api/orchestrators/FooOrchestrator并返回 VS Code 调试我的代码。
2. 我想编写一些单元测试来确保我的项目代码的质量。因此我愿意接受建议，也许只测试 Activity 函数的执行会更容易。

客户端功能代码

这是我的客户端函数的代码，主要是样板代码，如this one

import logging
import azure.functions as func
import azure.durable_functions as df

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    # 'starter' seems to contains the JSON data about
    # the URLs to monitor, stop, etc, the Durable Function
    client = df.DurableOrchestrationClient(starter)

    # The Client function knows which orchestrator to call
    # according to 'function_name'
    function_name = req.route_params["functionName"]

    # This part fails with a ClientConnectorError
    # with the message: "Cannot connect to host 127.0.0.1:17071 ssl:default"
    instance_id = await client.start_new(function_name, None, None)

    logging.info(f"Orchestration '{function_name}' starter with ID = '{instance_id}'.")

    return client.create_check_status_response(req, instance_id)

单元测试尝试

然后我尝试编写一些代码来触发这个客户端函数，就像我为一些“经典”Azure 函数所做的那样：

import asyncio
import json

if __name__ == "__main__":
    # Build a simple request to trigger the Client function
    req = func.HttpRequest(
        method="GET",
        body=None,
        url="don't care?",
        # What orchestrator do you want to trigger?
        route_params={"functionName": "FooOrchestrator"},
    )

    # I copy pasted the data that I obtained when I ran the Durable Function
    # with "func host start"
    starter = {
        "taskHubName": "TestHubName",
        "creationUrls": {
            "createNewInstancePostUri": "http://localhost:7071/runtime/webhooks/durabletask/orchestrators/{functionName}[/{instanceId}]?code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
            "createAndWaitOnNewInstancePostUri": "http://localhost:7071/runtime/webhooks/durabletask/orchestrators/{functionName}[/{instanceId}]?timeout={timeoutInSeconds}&pollingInterval={intervalInSeconds}&code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
        },
        "managementUrls": {
            "id": "INSTANCEID",
            "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID?taskHub=TestHubName&connection=Storage&code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
            "sendEventPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/raiseEvent/{eventName}?taskHub=TestHubName&connection=Storage&code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
            "terminatePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/terminate?reason={text}&taskHub=TestHubName&connection=Storage&code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
            "rewindPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/rewind?reason={text}&taskHub=TestHubName&connection=Storage&code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
            "purgeHistoryDeleteUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID?taskHub=TestHubName&connection=Storage&code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
            "restartPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/restart?taskHub=TestHubName&connection=Storage&code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
        },
        "baseUrl": "http://localhost:7071/runtime/webhooks/durabletask",
        "requiredQueryStringParameters": "code=aakw1DfReOkYCTFMdKPaA1Q6bSfnHZ/0lzvKsS6MVXCJdp4zhHKDJA==",
        "rpcBaseUrl": "http://127.0.0.1:17071/durabletask/",
    }

    # I need to use async methods because the "main" of the Client
    # uses async.
    reponse = asyncio.get_event_loop().run_until_complete(
        main(req, starter=json.dumps(starter))
    )

但不幸的是，客户端功能在 await client.start_new(function_name, None, None) 部分仍然失败。

如何在 Python 中为我的 Durable Azure Function 编写一些单元测试？

技术资料

• Python 版本：3.9
• Azure Functions 核心工具版本 4.0.3971
• 函数运行时版本：4.0.1.16815

Answer 1

不确定这是否会有所帮助，这是 Microsoft 提供的有关您正在寻找的单元测试的官方文档 - https://github.com/kemurayama/durable-functions-for-python-unittest-sample