Komponententest Durable Functions und dauerhafte Aufgaben-SDKs

Komponententests für dauerhafte Orchestrierungen helfen Ihnen dabei, Geschäftslogik zu überprüfen und Fehler frühzeitig abzufangen. Orchestrierungen koordinieren mehrere Aktivitäten und können schnell komplex werden, sodass Tests vor Regressionen schützen, während sich Ihr Workflow weiterentwickelt.

Wählen Sie die Registerkarte aus, die Ihrem Projekt entspricht: Durable Functions wenn Sie Azure Functions oder Durable Task SDKs verwenden, wenn Sie das eigenständige SDK ohne Azure Functions verwenden.

Mit Durable Functions testen Sie Orchestratoren, Aktivitäten und Clientfunktionen (Trigger) durch Mocken der vom Framework bereitgestellten Kontextobjekte und indem Sie Ihre Funktionen direkt aufrufen. Durch diesen Ansatz wird Ihre Geschäftslogik von der Azure Functions Laufzeit isoliert.

Hier ist ein minimaler C#-Orchestratortest zum Anzeigen des Musters:

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

Der Rest dieses Artikels behandelt dieses Muster ausführlich für C# und Python.

Die eigenständigen sdKs für dauerhafte Aufgaben bieten integrierte Testinfrastruktur , die Orchestrierungen ohne externe Abhängigkeiten im Arbeitsspeicher ausführt. Sie registrieren Orchestratoren und Aktivitäten mit einem Test-Worker, planen Orchestrierungen über einen Test-Client und bestätigen die Ergebnisse. Für C# und JavaScript ist kein Mocking erforderlich. Python verwendet einen generatorbasierten Ansatz mit manueller Ergebniseinfügung.

Hier ist ein minimaler C#-Test zum Anzeigen des Musters:

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

Der Rest dieses Artikels behandelt dieses Muster ausführlich für C#, Python und JavaScript.

Voraussetzungen

xUnit – Testframework
Moq — Simulationsframework
Vertrautheit mit dem .NET isolierten Worker-Modell

xUnit – Testframework
Das Microsoft.DurableTask.InProcessTestHost NuGet-Paket (v1.0.0 oder höher)

Testen von Orchestratorfunktionen

Orchestratorfunktionen koordinieren Aktivitäten, Zeitgeber und externe Ereignisse. Sie enthalten in der Regel die meiste Geschäftslogik und profitieren am meisten von Komponententests.

Simuliert den Orchestrierungskontext, um die Rückgabewerte von Aktivitätsaufrufen zu steuern. Rufen Sie dann Ihren Orchestrator direkt auf, und überprüfen Sie die Ausgabe.

Betrachten Sie diesen Orchestrator, der eine Aktivität dreimal aufruft:

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

Verwenden Sie Moq, um TaskOrchestrationContext zu mocken und erwartete Rückgabewerte für jeden Aktivitätsaufruf einzurichten.

Note

Das It.Is<TaskName>(...) Muster ist erforderlich, da CallActivityAsync eine TaskName Struktur akzeptiert und keine einfache Zeichenfolge. Moq benötigt die explizite Typ-Übereinstimmung.

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

Betrachten Sie diesen Orchestrator, der drei Aufrufe von Aktivitäten aneinanderreiht.

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]

Mocken Sie den Kontext und verwenden Sie orchestrator_generator_wrapper, um den Generator zu verarbeiten.

Note

Das Azure Durable Functions SDK stellt orchestrator_generator_wrapper bereit, um den Wiedergabemechanismus zu simulieren, der Python Orchestratorgeneratoren steuert. Das eigenständige Durable Task SDK enthält dieses Hilfsprogramm nicht, daher verwendet die Registerkarte "Dauerhafte Aufgabe-SDKs " stattdessen manuelle gen.send() Aufrufe.

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

Verwenden Sie DurableTaskTestHost, um Orchestrierungen im Arbeitsspeicher auszuführen. Registrieren Sie Ihre Produktionsorchestrator- und Aktivitätsklassen, planen Sie eine Orchestration, und bestätigen Sie das Ergebnis.

In Anbetracht dieser Produktionsklassen:

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

Registrieren Sie sie direkt im Testhost:

[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 führt ein vollständiges In-Memory-Orchestrierungsmodul aus. Es sind keine externen Dienste oder Sidecar-Prozesse erforderlich.

Das Python Durable Task SDK bietet noch keine integrierte Testumgebung wie C# und JavaScript. Verwenden Sie Standard Mocking zum Testen der Orchestratorlogik. Simulieren Sie den OrchestrationContext Generator und steuern Sie den Generator, indem Sie simulierte Aktivitätsergebnisse für jeden yieldzurücksenden.

Note

Im Gegensatz zum Azure Durable Functions SDK (das orchestrator_generator_wrapper bereitstellt), erfordert das eigenständige Durable Task SDK, dass Sie den Generator mit gen.send() manuell voranbringen. Dadurch erhalten Sie explizite Kontrolle über die simulierten Aktivitätsergebnisse.

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

Dieser Ansatz testet den Steuerungsfluss des Orchestrators, ohne von den internen Abläufen des SDK abhängig zu sein.

Verwenden Sie TestOrchestrationWorker und TestOrchestrationClient, um Orchestrierungen im Arbeitsspeicher auszuführen.

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

Die Testinfrastruktur führt das vollständige Orchestrierungsmodul im Prozess aus. Es sind keine Sidecars oder externe Dienste erforderlich.

Testen von Aktivitätsfunktionen

Aktivitätsfunktionen enthalten die eigentliche Arbeit – aufrufen von APIs, Verarbeiten von Daten oder Interagieren mit externen Systemen. Sie sind der einfachste Funktionstyp, der getestet werden soll, da sie kein frameworkspezifisches Wiedergabeverhalten aufweisen.

Aktivitätsfunktionen in Azure Functions empfangen eine Eingabe und gegebenenfalls ein FunctionContext. Testen Sie sie wie jede andere Funktion:

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

Aktivitätsfunktionen in Azure Functions sind normale Python Funktionen. Testen Sie sie direkt. Weitere Informationen finden Sie unter Azure Functions Python Komponententests.

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

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

Aktivitätsfunktionen empfangen ein Kontextobjekt und eine Eingabe. Der Kontext stellt Metadaten wie die Orchestrierungs-ID und die Aufgaben-ID bereit, aber die meisten Tests benötigen sie nicht.

Rufen Sie mithilfe der SayHelloActivity Klasse aus dem Orchestratorbeispiel RunAsync direkt mit einem simulierten Kontext auf:

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

Wenn Sie DurableTaskTestHost verwenden, werden Aktivitäten auch als Teil des Orchestrierungstests ausgeführt. Sie benötigen keine separaten Aktivitätstests, es sei denn, die Aktivität verfügt über komplexe Logik.

Testen Sie die Aktivitätsfunktion direkt – es ist keine spezielle Einrichtung erforderlich:

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

Testen Sie die Aktivitätsfunktion direkt:

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

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

Testen von Clientfunktionen

Clientfunktionen (auch als Triggerfunktionen bezeichnet) starten Orchestrierungen und verwalten Instanzen. Sie verwenden die dauerhafte Clientbindung, um mit dem Orchestrierungsmodul zu interagieren.

Erwägen Sie diesen HTTP-Trigger, der eine Orchestrierung startet:

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

Simuliert DurableTaskClient , um eine bekannte Instanz-ID zurückzugeben:

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

Erwägen Sie diesen HTTP-Trigger, der eine Orchestrierung startet:

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

Simuliert den dauerhaften Client:

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

Testen von Clientvorgängen

Mit den eigenständigen SDKs für dauerhafte Aufgaben nutzen Clientvorgänge (Orchestrierungen planen, Status abfragen, Ereignisse auslösen) dasselbe TestOrchestrationClient, wie es bereits in den Orchestratortests dargestellt wurde. Es ist keine separate Clientfunktion vorhanden – Sie rufen die Client-API direkt auf.

DurableTaskTestHost enthält host.Client, das eine voll funktionsfähige DurableTaskClient ist. Verwenden Sie diese, um Vorgänge auf Client-Ebene wie das Planen, Abfragen oder Beenden von Orchestrationen zu testen.

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

Das Python Durable Task SDK stellt noch keinen integrierten Testclient bereit. Testen Sie die Logik auf Clientebene (Terminplanung, Abfragevorgänge), indem Sie das Mocken von TaskHubGrpcClient verwenden.

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 unterstützt dieselben Vorgänge wie der Produktionsclient. Verwenden Sie sie zum Testen von Terminplanungs-, Statusabfragen und Auslösen von Ereignissen:

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

Feedback

War diese Seite hilfreich?

Last updated on 2026-04-28

Komponententest Durable Functions und dauerhafte Aufgaben-SDKs

Voraussetzungen

Testen von Orchestratorfunktionen

Testen von Aktivitätsfunktionen

Testen von Clientfunktionen

Testen von Clientvorgängen

Verwandte Inhalte

Feedback

Zusätzliche Ressourcen