對耐用函式與耐久任務 SDK 進行單元測試

單元測試持久性協調有助於你驗證業務邏輯並及早發現錯誤。編排協調多項活動，且可能迅速複雜，因此測試能防止隨著工作流程演進而出現迴歸。

選擇與你專案相符的分頁：如果你使用 Azure Functions，請選擇 Durable Functions;若使用獨立 SDK 且未Azure Functions，則選擇 Durable Task 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 的這個模式。

先決條件

xUnit — 測試框架
Moq — 模擬框架
熟悉.NET孤立工人模型

xUnit — 測試框架
Microsoft.DurableTask.InProcessTestHost NuGet 套件（v1.0.0 或更新版本）

測試編排器功能

編排器功能協調活動、計時器及外部事件。它們通常包含最多的商業邏輯，並從單元測試中獲益最大。

模擬協調流程內容，以控制活動呼叫的傳回值。然後直接打電話給你的編排器確認輸出。

請考慮這個會呼叫活動三次的協調器：

[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 並設定每個活動呼叫的期望回報值。

Note

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]);
}

請考慮這個會鏈結三個活動呼叫的協調器：

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 處理生成器。

Note

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!"])

用 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 運行完整的記憶體內編排引擎。不需要外部服務或 Sidecar 處理序。

Python Durable Task SDK 目前還沒有提供像 C# 和 JavaScript 那樣的內建測試工具。使用標準模擬來測試協調器邏輯。模擬OrchestrationContext，並針對每個yield回傳的模擬活動結果驅動產生器。

Note

與提供 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 內部結構。

使用 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、處理資料或與外部系統互動。它們是最容易測試的函式類型，因為沒有架構特定的重新執行行為。

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);
}

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!"

活動函式接收上下文物件與輸入。上下文會提供像是編排 ID 和任務 ID 這類的元資料，但大多數測試並不需要。

在編排器範例中使用 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時，活動也會作為編排測試的一部分執行。除非活動邏輯複雜，否則不需要另外的活動測試。

直接測試活動函數——不需要特殊設定：

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!"

直接測試活動函數：

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

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

測試用戶端功能

用戶端函式（也稱為觸發函式）啟動編排並管理實例。他們使用持久客戶端綁定來與編排引擎互動。

考慮這個啟動編排的 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);
}

考慮這個啟動編排的 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")

測試客戶端操作

使用獨立的耐久任務 SDK 時，客戶端作業 (排程協調流程、查詢狀態、引發事件) 會使用與協調器測試中已經展示的相同TestOrchestrationClient。沒有獨立的客戶端函式——你直接呼叫客戶端 API。

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 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")

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();
});

意見反應

此頁面對您有幫助嗎？

上次更新時間： 2026-06-11

對耐用函式與耐久任務 SDK 進行單元測試

先決條件

測試編排器功能

測試活動功能

測試用戶端功能

測試客戶端操作

相關內容

意見反應

其他資源