Unit test per Durable Functions e Durable Task SDK

L'esecuzione di unit test delle orchestrazioni durevoli consente di verificare la logica di business e rilevare gli errori in anticipo. Le orchestrazioni coordinano più attività e possono aumentare rapidamente le attività complesse, quindi i test proteggono dalle regressioni man mano che il flusso di lavoro si evolve.

Selezionare la scheda corrispondente al progetto: Durable Functions se si usa Funzioni di Azure o Durable Task SDK se si usa l'SDK autonomo senza Funzioni di Azure.

Con Durable Functions, è possibile testare orchestratori, attività e funzioni client (trigger) simulando gli oggetti di contesto forniti dal framework e chiamare direttamente le funzioni. Questo approccio isola la logica di business dal runtime di Funzioni di Azure.

Ecco un test minimo dell'agente di orchestrazione C# per mostrare il modello:

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

Il resto di questo articolo illustra in dettaglio questo modello per C# e Python.

Gli SDK di Durable Task autonomi forniscono un'infrastruttura di test predefinita che esegue orchestrazioni in memoria senza dipendenze esterne. Registrare agenti di orchestrazione e attività con un processo di lavoro di test, pianificare orchestrazioni tramite un client di test e verificare i risultati. Non è necessaria alcuna simulazione per C# e JavaScript. Python usa un approccio basato su generatore con l'inserimento manuale dei risultati.

Ecco un test C# minimo per mostrare il modello:

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

Il resto di questo articolo illustra in dettaglio questo modello per C#, Python e JavaScript.

Prerequisiti

C#

• xUnit : framework di test
• Moq: framework di comportamento fittizio
• Familiarità con il modello .NET di lavoro isolato

Python

• Python unittest — framework di test
• unittest.mock — libreria di simulazione
• Familiarità con il modello di programmazione Python v2

JavaScript

I test unitari JavaScript per Durable Functions richiedono l'SDK Durable Task autonomo. Passare alla scheda DURABLE Task SDK nella parte superiore di questa pagina per gli esempi di test JavaScript con l'infrastruttura di test predefinita.

C#

• xUnit : framework di test
• Pacchetto NuGet Microsoft.DurableTask.InProcessTestHost (v1.0.0 o versione successiva)

Python

• pytest — framework di test (o unittest)
• Pacchetto durabletask PyPI

JavaScript

• Jest : framework di test
• Pacchetto @microsoft/durabletask-js npm

Testare le funzioni dell'orchestratore

Le funzioni dell'agente di orchestrazione coordinano le attività, i timer e gli eventi esterni. In genere contengono la logica di business più vantaggiosa e traggono il massimo vantaggio dagli unit test.

Simulare il contesto di orchestrazione per controllare i valori restituiti delle chiamate di attività. Chiamare quindi direttamente l'agente di orchestrazione e verificare l'output.

C#

Considerare questo orchestratore che chiama un'attività tre volte:

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

Usare Moq per simulare TaskOrchestrationContext e configurare i valori restituiti previsti per ogni chiamata di attività.

Note

Il It.Is<TaskName>(...) criterio è obbligatorio perché CallActivityAsync accetta uno TaskName struct, non una stringa normale. Moq richiede la corrispondenza esplicita del tipo.

[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

Si consideri questo agente di orchestrazione che concatena tre chiamate di attività:

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]

Simulare il contesto e usare orchestrator_generator_wrapper per gestire il generatore.

Note

Azure Durable Functions SDK fornisce orchestrator_generator_wrapper per simulare il meccanismo di riproduzione che alimenta i generatori di agenti di orchestrazione Python. Durable Task SDK autonomo non include questa utilità, quindi la scheda Durable Task SDK usa invece chiamate manuali 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

Per gli unit test JavaScript, passare alla scheda Durable Task SDK nella parte superiore di questa pagina.

C#

Usare DurableTaskTestHost per eseguire orchestrazioni in memoria. Registrare l'agente di orchestrazione di produzione e le classi di attività, pianificare un'orchestrazione e verificare il risultato.

Date queste classi di produzione:

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

Registrarli direttamente nell'host di test:

[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 esegue un motore di orchestrazione in memoria completo. Non sono necessari servizi esterni o processi sidecar.

Python

Il Python Durable Task SDK non fornisce ancora un'interfaccia di test incorporata come C# e JavaScript. Usare la simulazione standard per testare la logica dell'agente di orchestrazione. Simulare OrchestrationContext e guidare il generatore inviando i risultati delle attività simulate per ogni yield.

Note

A differenza di Azure Durable Functions SDK (che fornisce orchestrator_generator_wrapper), il SDK Durable Task autonomo richiede di avanzare manualmente il generatore con gen.send(). In questo modo è possibile controllare in modo esplicito i risultati dell'attività simulata.

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

Questo approccio testa il flusso di controllo dell'agente di orchestrazione senza dipendere dagli elementi interni dell'SDK.

JavaScript

Usare TestOrchestrationWorker e TestOrchestrationClient per eseguire orchestrazioni in memoria:

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

L'infrastruttura di test esegue internamente il motore di orchestrazione completo. Non sono necessari servizi sidecar o esterni.

Funzioni di attività di test

Le funzioni di attività contengono il lavoro effettivo, ovvero la chiamata di API, l'elaborazione dei dati o l'interazione con sistemi esterni. Sono il tipo di funzione più semplice da testare perché non hanno un comportamento di riproduzione specifico del framework.

C#

Le funzioni di attività in Funzioni di Azure ricevono un input e facoltativamente un elemento FunctionContext. Testarli come qualsiasi altra funzione:

[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

Le funzioni di attività in Funzioni di Azure sono funzioni Python regolari. Provali direttamente. Per altre informazioni, vedere Funzioni di Azure Python unit test.

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

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

JavaScript

Per i test di attività JavaScript, passare alla scheda DURABLE Task SDK nella parte superiore di questa pagina.

Le funzioni di attività ricevono un oggetto contesto e un input. Il contesto fornisce metadati come l'ID di orchestrazione e l'ID attività, ma la maggior parte dei test non ne ha bisogno.

C#

Usando la SayHelloActivity classe dell'esempio dell'agente di orchestrazione, chiamare RunAsync direttamente con un contesto fittizio:

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

Quando si usa DurableTaskTestHost, le attività vengono eseguite anche come parte del test di orchestrazione. Non sono necessari test di attività separati, a meno che l'attività non abbia una logica complessa.

Python

Testare direttamente la funzione di attività: non è necessaria alcuna configurazione speciale:

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

Testare direttamente la funzione di attività:

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

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

Testare le funzioni client

Le funzioni client ,dette anche funzioni trigger, avviano orchestrazioni e gestiscono le istanze. Usano l'associazione client durevole per interagire con il motore di orchestrazione.

C#

Si consideri questo trigger HTTP che avvia un'orchestrazione:

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

Simulare DurableTaskClient per restituire un ID istanza noto:

[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

Si consideri questo trigger HTTP che avvia un'orchestrazione:

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

Simulare il client durevole:

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

Per i test del client JavaScript, passare alla scheda SDK per attività permanenti nella parte superiore di questa pagina.

Testare le operazioni client

Con gli SDK di Durable Task autonomi, le operazioni client (pianificazione di orchestrazioni, query di stato, generazione di eventi) usano gli stessi TestOrchestrationClient già visualizzati nei test dell'agente di orchestrazione. Non esiste alcuna funzione client separata: è possibile chiamare direttamente l'API client.

C#

DurableTaskTestHostespone host.Client, che è un DurableTaskClient completamente funzionale. Utilizzalo per testare operazioni lato client, come la pianificazione, l'esecuzione di query o il terminare delle orchestrazioni.

[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

Il Python Durable Task SDK non fornisce ancora un client di test predefinito. Testare la logica a livello di client (pianificazione, esecuzione di query) simulando 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 supporta le stesse operazioni del client di produzione. Usarlo per testare la pianificazione, le query sullo stato e la generazione di eventi:

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

Contenuti correlati

• Introduzione a Durable Functions
• Vincoli di codice della funzione di Orchestrator
• Eseguire la migrazione da in-process a un ruolo di lavoro isolato (.NET)
• Serializzazione e persistenza in Durable Functions

• Che cos'è Durable Task?
• Vincoli di codice della funzione di Orchestrator
• Sviluppare con il Durable Task Scheduler