单元测试 Durable Functions 和持久任务 SDK

单元测试持久性编排有助于验证业务逻辑并尽早捕获错误。 编排协调多个活动，很快变得复杂，因此测试可以在工作流发展的同时防止回归。

如果使用的是 Azure Functions，请选择与项目匹配的选项卡：Durable Functions;如果使用不带Azure Functions的独立 SDK，请选择可操作的任务 SDK。

使用 Durable Functions，可以通过模拟框架提供的上下文对象并直接调用函数来测试业务流程协调程序、活动和客户端（触发器）函数。 此方法将业务逻辑与Azure Functions运行时隔离开来。

下面是用于展示模式的简易 C# 编排器测试：

[Fact]
public async Task MyOrchestrator_CallsActivity()
{
    var contextMock = new Mock<TaskOrchestrationContext>();
    contextMock.Setup(x => x.CallActivityAsync<string>(
        It.IsAny<TaskName>(), It.IsAny<string>(), It.IsAny<TaskOptions>()))
        .ReturnsAsync("result");

    var result = await MyOrchestrator.Run(contextMock.Object);

    Assert.Equal("result", result);
}

本文的其余部分详细介绍了 C# 和 Python的此模式。

独立的 Durable Task SDK 提供 内置测试基础结构 ，用于在内存中运行业务流程，而无需外部依赖项。 你向测试工作进程注册业务流程协调程序和活动，通过测试客户端计划业务流程，并对结果进行断言。 C# 和 JavaScript 不需要模拟。 Python使用基于生成器的方法进行手动结果注入。

下面是用于显示模式的最小 C# 测试：

[Fact]
public async Task MyOrchestrator_Completes()
{
    await using var host = await DurableTaskTestHost.StartAsync(tasks =>
    {
        tasks.AddOrchestrator<MyOrchestrator>();
        tasks.AddActivity<MyActivity>();
    });

    string id = await host.Client.ScheduleNewOrchestrationInstanceAsync(nameof(MyOrchestrator));
    var result = await host.Client.WaitForInstanceCompletionAsync(id, getInputsAndOutputs: true);

    Assert.Equal(OrchestrationRuntimeStatus.Completed, result.RuntimeStatus);
}

本文的其余部分详细介绍了 C#、Python 和 JavaScript 的此模式。

先决条件

C#

• xUnit — 测试框架
• Moq — 模拟框架
• 熟悉 .NET 隔离工作者模型

Python

• Python unittest — 测试框架
• unittest.mock - 模拟库
• 熟悉 Python v2 编程模型

JavaScript

Durable Functions的 JavaScript 单元测试需要独立的 Durable Task SDK。 切换到本页顶部的 “持久任务 SDK ”选项卡，获取带有内置测试基础结构的 JavaScript 测试示例。

C#

• xUnit — 测试框架
• Microsoft.DurableTask.InProcessTestHost NuGet 包（v1.0.0 或更高版本）

Python

• pytest — 测试框架（或 unittest）
• durabletask PyPI 包

JavaScript

• Jest — 测试框架
• @microsoft/durabletask-js npm 包

测试编排器函数

协同器函数协调活动、计时器和外部事件。 它们通常包含最多的业务逻辑，并从单元测试中获益最大。

模拟编排上下文以控制任务调用的返回值。 然后直接调用你的调度器并验证输出。

C#

考虑这个编排器，它调用某项活动三次：

[Function(nameof(HelloCitiesOrchestration))]
public static async Task<List<string>> HelloCities(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    var outputs = new List<string>
    {
        await context.CallActivityAsync<string>(nameof(SayHello), "Tokyo"),
        await context.CallActivityAsync<string>(nameof(SayHello), "Seattle"),
        await context.CallActivityAsync<string>(nameof(SayHello), "London")
    };

    return outputs;
}

使用 Moq 模拟 TaskOrchestrationContext 并设置每个活动调用的预期返回值。

注释

该 It.Is<TaskName>(...) 模式是必需的，因为 CallActivityAsync 接受结构 TaskName ，而不是纯字符串。 Moq 需要显式类型匹配。

[Fact]
public async Task HelloCities_ReturnsExpectedGreetings()
{
    var contextMock = new Mock<TaskOrchestrationContext>();

    // Mock each activity call to return a known value
    contextMock.Setup(x => x.CallActivityAsync<string>(
        It.Is<TaskName>(n => n.Name == nameof(SayHello)),
        It.Is<string>(n => n == "Tokyo"),
        It.IsAny<TaskOptions>())).ReturnsAsync("Hello Tokyo!");

    contextMock.Setup(x => x.CallActivityAsync<string>(
        It.Is<TaskName>(n => n.Name == nameof(SayHello)),
        It.Is<string>(n => n == "Seattle"),
        It.IsAny<TaskOptions>())).ReturnsAsync("Hello Seattle!");

    contextMock.Setup(x => x.CallActivityAsync<string>(
        It.Is<TaskName>(n => n.Name == nameof(SayHello)),
        It.Is<string>(n => n == "London"),
        It.IsAny<TaskOptions>())).ReturnsAsync("Hello London!");

    var result = await HelloCitiesOrchestration.HelloCities(contextMock.Object);

    Assert.Equal(3, result.Count);
    Assert.Equal("Hello Tokyo!", result[0]);
    Assert.Equal("Hello Seattle!", result[1]);
    Assert.Equal("Hello London!", result[2]);
}

Python

考虑这个链接三个活动调用的业务流程协调程序：

import azure.durable_functions as df

def my_orchestrator(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity("say_hello", "Tokyo")
    result2 = yield context.call_activity("say_hello", "Seattle")
    result3 = yield context.call_activity("say_hello", "London")
    return [result1, result2, result3]

使用orchestrator_generator_wrapper模拟上下文并处理生成器。

注释

Azure Durable Functions SDK 提供 orchestrator_generator_wrapper 来模拟驱动 Python 编排器生成器的重播机制。 独立 Durable Task SDK 不包含此实用工具，因此 Durable Task SDK 选项卡改用手动 gen.send() 调用。

import unittest
from unittest.mock import Mock, call
from azure.durable_functions.testing import orchestrator_generator_wrapper

from function_app import my_orchestrator


def mock_activity(activity_name, input_value):
    mock_task = Mock()
    mock_task.result = f"Hello {input_value}!"
    return mock_task


class TestOrchestrator(unittest.TestCase):
    def test_chaining_orchestrator(self):
        # Extract the raw orchestrator function from the Azure Functions decorator wrapper
        func_call = my_orchestrator.build().get_user_function().orchestrator_function

        context = Mock()
        context.call_activity = Mock(side_effect=mock_activity)

        user_orchestrator = func_call(context)
        values = list(orchestrator_generator_wrapper(user_orchestrator))

        expected_calls = [
            call("say_hello", "Tokyo"),
            call("say_hello", "Seattle"),
            call("say_hello", "London"),
        ]
        self.assertEqual(context.call_activity.call_args_list, expected_calls)
        self.assertEqual(values[-1], ["Hello Tokyo!", "Hello Seattle!", "Hello London!"])

JavaScript

对于 JavaScript 单元测试，请切换到此页面顶部的 “持久任务 SDK ”选项卡。

C#

使用 DurableTaskTestHost 在内存中运行业务流程。 注册生产业务流程协调程序和活动类，计划业务流程，并对结果进行断言。

给定以下生产类：

class HelloCitiesOrchestrator : TaskOrchestrator<string, List<string>>
{
    public override async Task<List<string>> RunAsync(
        TaskOrchestrationContext context, string input)
    {
        var outputs = new List<string>
        {
            await context.CallActivityAsync<string>(nameof(SayHelloActivity), "Tokyo"),
            await context.CallActivityAsync<string>(nameof(SayHelloActivity), "Seattle"),
            await context.CallActivityAsync<string>(nameof(SayHelloActivity), "London")
        };
        return outputs;
    }
}

class SayHelloActivity : TaskActivity<string, string>
{
    public override Task<string> RunAsync(TaskActivityContext context, string name)
    {
        return Task.FromResult($"Hello {name}!");
    }
}

直接在测试主机中注册它们：

[Fact]
public async Task HelloCities_ReturnsExpectedGreetings()
{
    await using var host = await DurableTaskTestHost.StartAsync(tasks =>
    {
        tasks.AddOrchestrator<HelloCitiesOrchestrator>();
        tasks.AddActivity<SayHelloActivity>();
    });

    string instanceId = await host.Client.ScheduleNewOrchestrationInstanceAsync(
        nameof(HelloCitiesOrchestrator));
    OrchestrationMetadata result = await host.Client.WaitForInstanceCompletionAsync(
        instanceId, getInputsAndOutputs: true);

    Assert.Equal(OrchestrationRuntimeStatus.Completed, result.RuntimeStatus);

    var output = result.ReadOutputAs<List<string>>();
    Assert.Equal(3, output.Count);
    Assert.Equal("Hello Tokyo!", output[0]);
    Assert.Equal("Hello Seattle!", output[1]);
    Assert.Equal("Hello London!", output[2]);
}

DurableTaskTestHost 运行完整的内存业务流程引擎。 不需要外部服务或边车进程。

Python

Python Durable Task SDK 尚不提供 C# 和 JavaScript 等内置测试工具。 使用标准模拟测试业务流程协调程序逻辑。 模拟 OrchestrationContext 并通过为每个 yield 发送模拟活动结果来驱动生成器。

注释

与提供 orchestrator_generator_wrapper 的 Azure Durable Functions SDK 不同，独立 Durable Task SDK 要求使用 gen.send() 手动推进生成器。 这样就可以显式控制模拟活动结果。

from unittest.mock import Mock, call
from durabletask import task


def hello(ctx: task.ActivityContext, name: str) -> str:
    return f"Hello {name}!"


def hello_cities(ctx: task.OrchestrationContext, _):
    result1 = yield ctx.call_activity(hello, input="Tokyo")
    result2 = yield ctx.call_activity(hello, input="Seattle")
    result3 = yield ctx.call_activity(hello, input="London")
    return [result1, result2, result3]


def test_hello_cities():
    ctx = Mock(spec=task.OrchestrationContext)

    # Each call_activity returns a mock task; collect them to send results back
    mock_tasks = []
    def fake_call_activity(activity, *, input):
        t = Mock()
        t._input = input
        mock_tasks.append(t)
        return t

    ctx.call_activity = Mock(side_effect=fake_call_activity)

    # Start the generator
    gen = hello_cities(ctx, None)
    yielded = next(gen)  # first yield: call_activity(hello, input="Tokyo")

    # Send simulated results back for each yield
    try:
        yielded = gen.send("Hello Tokyo!")
        yielded = gen.send("Hello Seattle!")
        gen.send("Hello London!")
    except StopIteration as e:
        result = e.value

    # Verify the orchestrator called the right activities
    assert ctx.call_activity.call_count == 3
    assert result == ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]

此方法检查协调程序的控制流，无需依赖 SDK 内部。

JavaScript

使用 TestOrchestrationWorker 和 TestOrchestrationClient 在内存中运行业务流程：

const {
    InMemoryOrchestrationBackend,
    TestOrchestrationClient,
    TestOrchestrationWorker,
    OrchestrationStatus,
} = require("@microsoft/durabletask-js");

test("helloCities returns expected greetings", async () => {
    const backend = new InMemoryOrchestrationBackend();
    const client = new TestOrchestrationClient(backend);
    const worker = new TestOrchestrationWorker(backend);

    const sayHello = async (_, name) => `Hello ${name}!`;

    const helloCities = async function* (ctx) {
        const outputs = [];
        outputs.push(yield ctx.callActivity(sayHello, "Tokyo"));
        outputs.push(yield ctx.callActivity(sayHello, "Seattle"));
        outputs.push(yield ctx.callActivity(sayHello, "London"));
        return outputs;
    };

    worker.addOrchestrator(helloCities);
    worker.addActivity(sayHello);
    await worker.start();

    const id = await client.scheduleNewOrchestration(helloCities);
    const state = await client.waitForOrchestrationCompletion(id, true, 10);

    expect(state.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED);
    expect(JSON.parse(state.serializedOutput)).toEqual([
        "Hello Tokyo!",
        "Hello Seattle!",
        "Hello London!",
    ]);

    await worker.stop();
});

测试基础设施在进程内运行完整的编排引擎。 不需要边车或外部服务。

测试活动函数

活动函数包含实际工作 - 调用 API、处理数据或与外部系统交互。 它们是最简单的功能类型进行测试，因为它们不涉及任何特定框架的重播行为。

C#

Azure Functions 中的活动函数接收输入和可选的 FunctionContext。 像任何其他函数一样测试它们：

[Function(nameof(SayHello))]
public static string SayHello(
    [ActivityTrigger] string name, FunctionContext executionContext)
{
    return $"Hello {name}!";
}

[Fact]
public void SayHello_ReturnsExpectedGreeting()
{
    var result = HelloCitiesOrchestration.SayHello("Tokyo", Mock.Of<FunctionContext>());
    Assert.Equal("Hello Tokyo!", result);
}

Python

Azure Functions中的活动函数是常规Python函数。 直接测试它们。 有关详细信息，请参阅 Azure Functions Python 单元测试。

def say_hello(name: str) -> str:
    return f"Hello {name}!"

def test_say_hello():
    result = say_hello("Tokyo")
    assert result == "Hello Tokyo!"

JavaScript

对于 JavaScript 活动测试，请切换到此页面顶部的 “持久任务 SDK ”选项卡。

活动函数接收上下文对象和输入。 上下文提供业务流程 ID 和任务 ID 等元数据，但大多数测试都不需要它。

C#

使用协调器示例中的SayHelloActivity类，并直接使用模拟上下文调用RunAsync。

[Fact]
public async Task SayHello_ReturnsExpectedGreeting()
{
    var activity = new SayHelloActivity();
    var contextMock = new Mock<TaskActivityContext>();

    var result = await activity.RunAsync(contextMock.Object, "Tokyo");

    Assert.Equal("Hello Tokyo!", result);
}

使用 DurableTaskTestHost时，活动也会作为编排测试的一部分运行。 除非活动具有复杂的逻辑，否则不需要单独的活动测试。

Python

直接测试活动函数 - 无需特殊设置：

from durabletask import task


def hello(ctx: task.ActivityContext, name: str) -> str:
    return f"Hello {name}!"


def test_hello():
    ctx = Mock(spec=task.ActivityContext)
    result = hello(ctx, "Tokyo")
    assert result == "Hello Tokyo!"

JavaScript

直接测试活动函数：

const sayHello = async (_, name) => `Hello ${name}!`;

test("sayHello returns expected greeting", async () => {
    const result = await sayHello(undefined, "Tokyo");
    expect(result).toBe("Hello Tokyo!");
});

测试客户端函数

客户端函数（也称为触发器函数）启动业务流程和管理实例。 它们使用持久客户端绑定与业务流程引擎交互。

C#

请考虑此启动业务流程的 HTTP 触发器：

[Function("HelloCitiesOrchestration_HttpStart")]
public static async Task<HttpResponseData> HttpStart(
    [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestData req,
    [DurableClient] DurableTaskClient client,
    FunctionContext executionContext)
{
    string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
        nameof(HelloCitiesOrchestration));
    return await client.CreateCheckStatusResponseAsync(req, instanceId);
}

模拟 DurableTaskClient 以返回已知实例 ID：

[Fact]
public async Task HttpStart_ReturnsAccepted()
{
    var durableClientMock = new Mock<DurableTaskClient>("testClient");
    var functionContextMock = new Mock<FunctionContext>();
    var instanceId = "test-instance-id";

    durableClientMock
        .Setup(x => x.ScheduleNewOrchestrationInstanceAsync(
            It.IsAny<TaskName>(),
            It.IsAny<object>(),
            It.IsAny<StartOrchestrationOptions>(),
            It.IsAny<CancellationToken>()))
        .ReturnsAsync(instanceId);

    var mockRequest = CreateMockHttpRequest(functionContextMock.Object);

    var responseMock = new Mock<HttpResponseData>(functionContextMock.Object);
    responseMock.SetupGet(r => r.StatusCode).Returns(HttpStatusCode.Accepted);

    durableClientMock
        .Setup(x => x.CreateCheckStatusResponseAsync(
            It.IsAny<HttpRequestData>(),
            It.IsAny<string>(),
            It.IsAny<CancellationToken>()))
        .ReturnsAsync(responseMock.Object);

    var result = await HelloCitiesOrchestration.HttpStart(
        mockRequest, durableClientMock.Object, functionContextMock.Object);

    Assert.Equal(HttpStatusCode.Accepted, result.StatusCode);
}

Python

请考虑此启动业务流程的 HTTP 触发器：

@app.route(route="start")
@app.durable_client_input(client_name="client")
async def http_start(req: func.HttpRequest, client):
    instance_id = await client.start_new("my_orchestrator")
    return client.create_check_status_response(req, instance_id)

持久化客户端模拟：

import asyncio
import unittest
import azure.functions as func
from unittest.mock import AsyncMock, Mock

from function_app import http_start


class TestClientFunction(unittest.TestCase):
    def test_http_start(self):
        # Extract the raw client function from the Azure Functions decorator wrapper
        func_call = http_start.build().get_user_function().client_function

        req = func.HttpRequest(
            method="GET",
            body=b"{}",
            url="/api/start",
        )

        client = Mock()
        client.start_new = AsyncMock(return_value="test-instance-id")
        client.create_check_status_response = Mock(return_value="check_status_response")

        result = asyncio.run(func_call(req, client))

        client.start_new.assert_called_once_with("my_orchestrator")
        client.create_check_status_response.assert_called_once_with(req, "test-instance-id")

JavaScript

对于 JavaScript 客户端测试，请切换到此页面顶部的 “持久任务 SDK ”选项卡。

测试客户端操作

使用独立持久任务 SDK 时，客户端操作（安排业务流程、查询状态、引发事件）使用业务流程协调程序测试中已显示的相同 TestOrchestrationClient。 不存在单独的客户端函数 — 直接调用客户端 API。

C#

DurableTaskTestHost 会公开host.Client，后者是完全正常运行的 DurableTaskClient。 使用它来测试客户端级别的操作，例如调度、查询或终止编排过程：

[Fact]
public async Task Client_CanQueryOrchestrationStatus()
{
    await using var host = await DurableTaskTestHost.StartAsync(tasks =>
    {
        tasks.AddOrchestrator<HelloCitiesOrchestrator>();
        tasks.AddActivity<SayHelloActivity>();
    });

    string instanceId = await host.Client.ScheduleNewOrchestrationInstanceAsync(
        nameof(HelloCitiesOrchestrator));

    // Query status while the orchestration runs
    OrchestrationMetadata metadata = await host.Client.WaitForInstanceCompletionAsync(
        instanceId, getInputsAndOutputs: true);

    Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
    Assert.Equal(instanceId, metadata.InstanceId);
}

Python

Python Durable Task SDK 尚不提供内置测试客户端。 通过模拟TaskHubGrpcClient测试客户端级逻辑（调度、查询）

from unittest.mock import AsyncMock, Mock


def test_schedule_orchestration():
    client = Mock()
    client.schedule_new_orchestration = AsyncMock(return_value="test-id")

    # Call your application code that uses the client
    import asyncio
    instance_id = asyncio.run(client.schedule_new_orchestration("my_orchestrator"))

    assert instance_id == "test-id"
    client.schedule_new_orchestration.assert_called_once_with("my_orchestrator")

JavaScript

TestOrchestrationClient 支持与生产客户端相同的操作。 使用它测试调度、状态查询和触发事件：

const {
    InMemoryOrchestrationBackend,
    TestOrchestrationClient,
    TestOrchestrationWorker,
    OrchestrationStatus,
} = require("@microsoft/durabletask-js");

test("client can schedule and query orchestration", async () => {
    const backend = new InMemoryOrchestrationBackend();
    const client = new TestOrchestrationClient(backend);
    const worker = new TestOrchestrationWorker(backend);

    const myOrchestrator = async function* (ctx) {
        return "done";
    };

    worker.addOrchestrator(myOrchestrator);
    await worker.start();

    const id = await client.scheduleNewOrchestration(myOrchestrator);
    const state = await client.waitForOrchestrationCompletion(id, true, 10);

    expect(state.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED);
    expect(JSON.parse(state.serializedOutput)).toBe("done");

    await worker.stop();
});

相关内容

• Durable Functions 概述
• Orchestrator 函数代码约束
• 从进程内迁移到独立辅助角色 (.NET)
• Durable Functions 中的序列化和持久化

• 什么是持久任务？
• Orchestrator 函数代码约束
• 使用持久任务计划程序进行开发