Teste de unidade de Durable Functions e SDKs de Tarefas Duráveis

O teste de unidade de orquestrações duráveis ajuda você a verificar a lógica de negócios e detectar erros antecipadamente. As orquestrações coordenam várias atividades e podem ficar complexas rapidamente, portanto, os testes protegem contra regressões à medida que o fluxo de trabalho evolui.

Selecione a guia que corresponde ao seu projeto: Durable Functions se você usar Azure Functions, ou SDKs de Tarefas Duráveis se utilizar o SDK autônomo sem Azure Functions.

Com Durable Functions, você testa orquestradores, atividades e funções de cliente (gatilho) simulando os objetos de contexto fornecidos pela estrutura e chamando suas funções diretamente. Essa abordagem isola sua lógica de negócios do runtime do Azure Functions.

Aqui está um teste mínimo do orquestrador em C# para mostrar o padrão:

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

O restante deste artigo aborda esse padrão em detalhes para C# e Python.

Os SDKs de Tarefa Duráveis autônomos fornecem uma infraestrutura de teste interna que executa orquestrações na memória sem dependências externas. Registre orquestradores e atividades com um trabalho de teste, agende orquestrações por meio de um cliente de teste e afirme sobre os resultados. Nenhum mock é necessário para C# e JavaScript. Python usa uma abordagem baseada em gerador com injeção de resultado manual.

Aqui está um teste de C# mínimo para mostrar o padrão:

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

O restante deste artigo aborda esse padrão em detalhes para C#, Python e JavaScript.

Pré-requisitos

xUnit — estrutura de teste
Moq - estrutura de simulação
Familiaridade com o modelo de trabalho isolado .NET

xUnit — estrutura de teste
O pacote NuGet Microsoft.DurableTask.InProcessTestHost (v1.0.0 ou posterior)

Testar funções de um orquestrador

As funções de orquestrador coordenam atividades, temporizadores e eventos externos. Normalmente, elas contêm mais lógica de negócios e se beneficiam mais do teste de unidade.

Simular o contexto de orquestração para controlar os valores retornados das chamadas de atividade. Em seguida, chame o orquestrador diretamente e verifique a saída.

Considere este orquestrador que chama uma atividade três vezes:

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

Use o Moq para simular TaskOrchestrationContext e configurar valores retornados esperados para cada chamada de atividade.

Note

O It.Is<TaskName>(...) padrão é necessário porque CallActivityAsync aceita um TaskName struct, não uma cadeia de caracteres simples. Moq precisa da correspondência explícita de 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]);
}

Considere este orquestrador que encadeia três chamadas de atividade:

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]

Simular o contexto e usar orchestrator_generator_wrapper para processar o gerador.

Note

O SDK do Azure Durable Functions fornece orchestrator_generator_wrapper para simular o mecanismo de reprodução que conduz geradores de orquestradores Python. O SDK autônomo de Tarefa Durável não inclui esta ferramenta, portanto, a guia SDKs de Tarefa Durável utiliza chamadas manuais 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!"])

Use DurableTaskTestHost para executar orquestrações na memória. Registre seu orquestrador de produção e as classes de atividade, agende uma orquestração e verifique o resultado.

Considerando estas classes de produção:

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

Registre-os diretamente no host de teste:

[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 executa um mecanismo de orquestração completo em memória. Nenhum serviço externo ou processos do tipo sidecar são necessários.

O SDK Python para Tarefas Duráveis ainda não fornece uma estrutura de teste embutida, como C# e JavaScript. Use a simulação padrão para testar a lógica do orquestrador. Simular o OrchestrationContext e operar o gerador enviando cada resultado de atividade simulada de volta para cada yield.

Note

Ao contrário do SDK Azure Durable Functions (que fornece orchestrator_generator_wrapper), o SDK de Tarefa Durável autônomo exige que você avance manualmente o gerador com gen.send(). Isso fornece controle explícito sobre os resultados da atividade simulada.

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

Essa abordagem testa o fluxo de controle do orquestrador sem depender do funcionamento interno do SDK.

Use TestOrchestrationWorker e TestOrchestrationClient para executar orquestrações na memória.

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

A infraestrutura de teste executa o motor de orquestração completo no mesmo processo. Nenhum sidecar ou serviços externos são necessários.

Funções de atividade de teste

As funções de atividade contêm o trabalho real : chamar APIs, processar dados ou interagir com sistemas externos. Eles são o tipo de função mais simples a ser testado porque não têm nenhum comportamento de reprodução específico da estrutura.

As funções de atividade em Azure Functions recebem uma entrada e, opcionalmente, um FunctionContext. Teste-os como qualquer outra função:

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

As funções de atividade em Azure Functions são funções de Python regulares. Teste-os diretamente. Para obter mais informações, consulte teste de unidade do 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!"

As funções de atividade recebem um objeto de contexto e uma entrada. O contexto fornece metadados como a ID de orquestração e a ID da tarefa, mas a maioria dos testes não precisa dele.

Usando a SayHelloActivity classe do exemplo de orquestrador, chame RunAsync diretamente com um contexto fictício:

[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 você usa DurableTaskTestHost, as atividades também são executadas como parte do teste de orquestração. Você não precisa de testes de atividade separados, a menos que a atividade tenha uma lógica complexa.

Teste a função de atividade diretamente — nenhuma configuração especial é necessária:

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

Teste a função de atividade diretamente:

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

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

Testar funções de cliente

As funções de cliente (também chamadas de funções de gatilho) iniciam orquestrações e gerenciam instâncias. Eles usam a associação de cliente durável para interagir com o mecanismo de orquestração.

Considere este gatilho HTTP que inicia uma orquestração:

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

Simular DurableTaskClient para retornar uma ID de instância conhecida:

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

Considere este gatilho HTTP que inicia uma orquestração:

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

Simular o cliente durável:

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

Testar operações do cliente

Com os SDKs de Tarefa Duráveis autônomos, as operações do cliente (agendando orquestrações, consultando status, gerando eventos) usam o mesmo TestOrchestrationClient já mostrado nos testes do orquestrador. Não existe nenhuma função de cliente separada – você chama a API do cliente diretamente.

DurableTaskTestHost expõe host.Client, que é um DurableTaskClient totalmente funcional. Use-o para testar operações no nível do cliente, como agendamento, consulta ou encerramento de orquestrações.

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

O SDK Python Durable Task ainda não fornece um cliente de teste embutido. Teste a lógica no nível do cliente (agendamento, consulta) zombando do 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 dá suporte às mesmas operações que o cliente de produção. Use-o para testar agendamento, consultas de status e aumento de eventos:

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

Comentários

Esta página foi útil?

Last updated on 2026-06-03

Teste de unidade de Durable Functions e SDKs de Tarefas Duráveis

Pré-requisitos

Testar funções de um orquestrador

Funções de atividade de teste

Testar funções de cliente

Testar operações do cliente

Conteúdo relacionado

Comentários

Recursos adicionais