Testando workflows do Step Functions: um guia para a API TestState aprimorada

Por D Surya Sai, Technical Account Manager na Amazon Web Services e Sahithi Ginjupalli, Cloud Support Engeneer na Amazon Web Services.

AWS Step Functions anunciou recentemente novos aprimoramentos nas capacidades de teste local, introduzindo testes baseados em API que os desenvolvedores podem usar para validar workflows antes de implantar na AWS. Conforme detalhado em nossa postagem de blog de anúncio, a API TestState transforma o desenvolvimento do Step Functions ao permitir o teste de estados individuais isoladamente ou como workflows completos. A API suporta respostas simuladas e integrações reais com serviços AWS, e fornece capacidades avançadas, como estados Map/Parallel, simulação de erros com mecanismos de repetição, validação de objeto de contexto e metadados de inspeção detalhados para testes locais abrangentes da sua aplicação Serverless.

A API TestState pode ser acessada por meio de múltiplas interfaces, como AWS Command Line Interface (AWS CLI), AWS SDK, LocalStack. Por padrão, a API TestState no AWS CLI e SDK é executada contra o endpoint AWS remoto, fornecendo validação contra a infraestrutura real do serviço Step Functions. Firmamos parceria com o LocalStack para oferecer um endpoint de teste adicional para a API TestState. Os desenvolvedores podem usar o LocalStack para testes unitários de seus workflows alterando a configuração do endpoint do cliente AWS SDK para apontar para o LocalStack: http://localhost.localstack.cloud:4566/ em vez do endpoint AWS. Essa abordagem oferece isolamento completo de rede quando necessário. Para uma experiência de desenvolvimento simplificada, você também pode usar a extensão VSCode do LocalStack para configurar automaticamente seu ambiente para apontar para o endpoint do LocalStack. Essa abordagem é descrita na postagem de blog da AWS.

Esta publicação demonstra como construir suítes de testes para testar unitariamente seus workflows do Step Functions usando o AWS SDK para Python com o framework pytest. A implementação completa está disponível no repositório do GitHub.

Construindo casos de teste usando a API TestState

Este exemplo de workflow implementa um sistema real de processamento de pedidos de e-commerce usando JSONata para transformações avançadas de dados. Ele incorpora padrões complexos do Step Functions incluindo estados Map distribuídos, execução Parallel e mecanismos de callback waitForTaskToken. O processo valida pedidos através de funções AWS Lambda, distribui o processamento de itens de pedidos com tolerância configurável a falhas, executa atualizações paralelas de pagamento e inventário, lida com workflows de aprovação humana usando tokens de tarefa e, em seguida, persiste pedidos no Amazon DynamoDB com entrega de notificação. Este workflow demonstra tratamento avançado de erros com múltiplos Catchers e Retriers, backoff exponencial para limitação do Lambda e limites do DynamoDB, e transições de estado sofisticadas que anteriormente eram desafiadoras de testar localmente. Isso o torna ideal para demonstrar o uso dos recursos de teste local da API TestState aprimorada.

O workflow completo está disponível no repositório do GitHub, onde você pode examinar a definição completa da máquina de estados e ver como as expressões JSONata lidam com a transformação de dados ao longo do fluxo de execução.

Figura 1: Workflow da máquina de estados que demonstra um sistema real de processamento de pedidos de e-commerce.Testes eficazes do Step Functions requerem uma abordagem sistemática de integração com a API TestState, que oferece validação de estado, simulação de erros e capacidades de asserção. O framework de teste é construído usando o framework pytest do Python, usando fixtures que fornecem automaticamente instâncias de runner pré-configuradas, responsáveis pela inicialização do cliente da API TestState e o carregamento da definição da máquina de estados. Isso elimina código de configuração repetitivo e garante consistência nos ambientes de teste. A API TestState aprimorada suporta tanto integrações simuladas quanto integrações reais com serviços AWS, fornecendo flexibilidade nas estratégias de teste. Nesta demonstração, são usadas integrações simuladas para mostrar como testes locais completos podem ser realizados sem nenhum recurso implantado em contas da AWS.

Este framework foi construído para fins de demonstração, e você pode criar seus próprios frameworks de teste de forma semelhante usando outras linguagens de programação como Java, Node.js. O framework de teste usa padrões de encadeamento de métodos para criar casos de teste legíveis, com métodos de asserção abrangentes, encadeamento automático de saída entre execuções de estado e simulação de erros para testar mecanismos de repetição, intervalos de backoff e blocos catch em condições de erro de serviços da AWS.

As implementações de teste a seguir demonstram as capacidades de teste alcançáveis com a API TestState aprimorada em ambientes de desenvolvimento local. Os casos de teste são executados com base na máquina de estados anterior.

Caso de Teste 1: Limitação do Lambda e teste de mecanismo de repetição

Integrações de serviços com máquinas de estados, como AWS Lambda e Amazon DynamoDB, podem enfrentar limitação dependendo do uso. Uma capacidade-chave da API TestState aprimorada é a possibilidade de simular mecanismos de repetição com controle sobre contagens de repetição e intervalos de backoff. Este teste demonstra as capacidades de teste de repetição da API TestState aprimorada através do parâmetro stateConfiguration.retrierRetryCount e campos de resposta inspectionData.errorDetails. Este campo de resposta fornece retryBackoffIntervalSeconds para validar os cálculos de backoff exponencial, retryIndex para rastrear a sequência de tentativas de repetição e catchIndex para identificar qual tratador de erro processou a exceção. Essas capacidades de inspeção aprimoradas permitem validar a lógica de repetição, estratégias de backoff e padrões de propagação de erros em workflows complexos de máquinas de estado.

def test_lambda_throttling_retry_mechanism(self, runner):
"""Test retry mechanism for Lambda.TooManyRequestsException"""
throttling_error = {
"Error": "Lambda.TooManyRequestsException",
"Cause": "Request rate exceeded"
}

# Test first retry attempt
(runner
.with_input({"orderId": "order-retry-test"})
.with_mock_error(throttling_error)
.with_retrier_retry_count(0)
.execute("ValidateOrder")
.assert_retriable()
.assert_error("Lambda.TooManyRequestsException"))

# Verify exponential backoff calculation
response = runner.get_response()
error_details = response['inspectionData']['errorDetails']
assert error_details['retryBackoffIntervalSeconds'] == 2

# Test retry exhaustion
(runner
.with_retrier_retry_count(3)
.execute("ValidateOrder")
.assert_caught_error()
.assert_next_state("ValidationFailed"))

Caso de Teste 2: Teste de estado Map com limites de tolerância

Estados Map distribuídos apresentam desafios únicos de teste devido à natureza de processamento paralelo e capacidades de tolerância a falhas. A API TestState aprimorada oferece opções de configuração especializadas para testar esses cenários complexos.

def test_map_state_tolerated_failure_threshold(self, runner):
"""Test Map state with tolerated failure threshold"""
test_input = {
"orderId": "order-map-test",
"orderItems": [
{"itemId": "item-1"}, {"itemId": "item-2"}, 
{"itemId": "item-3"}, {"itemId": "item-4"}
]
}

# Test normal Map state execution
map_success_result = [
{"itemId": "item-1", "processed": True},
{"itemId": "item-2", "processed": True}
]

(runner
.with_input(test_input)
.with_mock_result(map_success_result)
.execute("ProcessOrderItems")
.assert_succeeded()
.assert_next_state("ParallelProcessing"))

# Test tolerance threshold exceeded scenario
tolerance_error = {
"Error": "States.ExceedToleratedFailureThreshold",
"Cause": "Map state exceeded tolerated failure threshold"
}

(runner
.with_input(test_input)
.with_mock_error(tolerance_error)
.execute("ProcessOrderItems")
.assert_caught_error()
.assert_next_state("ValidationFailed"))

Este teste demonstra as capacidades de teste de estado Map da API TestState aprimorada através do parâmetro stateConfiguration.mapIterationFailureCount para simular falhas de iteração. A API fornece dados de inspeção abrangentes, incluindo inspectionData.afterItemSelector para validar transformações ItemSelector, inspectionData.afterItemBatcher para validação de processamento em lote, inspectionData.toleratedFailureCount e inspectionData.toleratedFailurePercentage para verificação de limites. Quando a contagem de falhas especificada excede a tolerância configurada, a API retorna corretamente States.ExceedToleratedFailureThreshold, permitindo testar padrões de resiliência de estado Map.

Caso de Teste 3: Teste de padrão WaitForCallback

A integração waitForCallback requer a construção de objeto de contexto para simular ambientes de execução realistas, especialmente para workflows de aprovação humana.

def test_context_object_usage_in_jsonata_expressions(self, runner):
"""Test Context object usage in waitForTaskToken scenarios"""
test_input = {
"orderId": "order-context-test",
"amount": 125.0
}

context_data = {
"Task": {"Token": "ahbdgftgehbdcndsjnwjkhas327yr4hendc73yehdb723y"},
"Execution": {
"Id": "arn:aws:states:us-east-1:123456789012:execution:test:exec-123"
},
"State": {
"Name": "WaitForApproval",
"EnteredTime": "2025-01-15T10:45:00Z"
}
}

mock_result = {
"approved": True,
"taskToken": "ahbdgftgehbdcndsjnwjkhas327yr4hendc73yehdb723y"
}

(runner
.with_input(test_input)
.with_context(context_data)
.with_mock_result(mock_result)
.execute("WaitForApproval")
.assert_succeeded()
.assert_next_state("CheckApproval"))

# Verify JSONata expressions processed context correctly
response = runner.get_response()
after_args = json.loads(response['inspectionData']['afterArguments'])
assert after_args['Payload']['taskToken'] == context_data['Task']['Token']

Este teste demonstra o suporte da API TestState aprimorada para integrações waitForCallback através do parâmetro context para simulação realista de objeto Context. A API permite testar de forma abrangente expressões JSONata que referenciam $states.context.Task.Token, $states.context.Execution.Id e outros campos de contexto. O campo de resposta inspectionData.afterArguments valida que as expressões JSONata processaram corretamente os dados de contexto, enquanto a API gerencia automaticamente a complexidade de incorporação de token de tarefa em payloads de integração de serviço para cenários de teste waitForCallback.

Caso de Teste 4: Teste de caminho feliz – validação completa de workflow

O teste de caminho feliz valida que os workflows são executados corretamente em condições normais de operação. A API TestState aprimorada permite encadear execuções de estado, passando automaticamente as saídas entre estados para simular uma execução completa de workflow.

def test_complete_order_processing_workflow(self, runner):
"""Integration test: Complete happy path workflow using method chaining"""
test_input = {
"orderId": "order-12345",
"amount": 150.75,
"customerEmail": "customer@example.com",
"orderItems": [
{"itemId": "item-1", "quantity": 2, "price": 50.25}
]
}

# Test ValidateOrder state
(runner
.with_input(test_input)
.with_mock_result({"statusCode": 200, "isValid": True})
.execute("ValidateOrder")
.assert_succeeded()
.assert_next_state("CheckValidation"))

# Test CheckValidation choice state (no mock needed)
validation_output = runner.get_output()
(runner
.with_input(validation_output)
.clear_mocks()
.execute("CheckValidation")
.assert_succeeded()
.assert_next_state("ProcessOrderItems"))

Este teste demonstra como a API TestState mantém o contexto de estado entre execuções, permitindo uma simulação realista de workflow. O método get_output() recupera a saída processada de um estado para usar como entrada do próximo, reproduzindo o comportamento real de execução do Step Functions.

Nota: O trecho de código acima mostra apenas os dois primeiros estados do teste completo de workflow, por brevidade. O código de teste completo com todos os estados (ProcessOrderItems, ParallelProcessing, WaitForApproval, CheckApproval, SaveOrderDetails e SendNotification) pode ser visualizado no repositório completo do GitHub, demonstrando a validação de workflow de ponta a ponta usando o mesmo padrão de encadeamento de métodos.

Integração com pipelines modernos de CI/CD

Nesta seção, vamos explorar como integrar os testes unitários anteriores em um pipeline de CI/CD para viabilizar testes locais.

O repositório de exemplo inclui um workflow do GitHub Actions que demonstra como o teste da API TestState se integra em pipelines de integração contínua e entrega contínua (CI/CD). O workflow (.github/workflows/test-and-deploy.yml) define um processo de duas etapas que valida antes que quaisquer recursos da AWS sejam implantados usando AWS Serverless Application Model (AWS SAM).

O pipeline de CI/CD segue o seguinte padrão:

Testes Unitários: Executa a suíte completa de testes da API TestState usando pytest tests/unit_test.py -v
SAM Deploy: Implanta recursos AWS usando sam build e sam deploy

Para que o workflow do GitHub Actions possa implantar recursos na sua conta da AWS, configure essas credenciais da AWS nas configurações do seu repositório do GitHub. Para instruções detalhadas de configuração, consulte a postagem de blog da AWS.

A seguir estão os segredos que devem ser configurados nas configurações do repositório do GitHub:

AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY
AWS_REGION

Em ambientes de produção, você pode estender este pipeline básico para incluir etapas adicionais. O pipeline aprimorado geralmente começa com a implantação em uma conta de desenvolvimento, seguida por testes de integração com os recursos implantados. A etapa final envolve a promoção para produção com portões de aprovação adequados e verificações de segurança e conformidade.

Conclusão

A API TestState aprimorada permite testar workflows do Step Functions localmente sem exigir implantações na AWS, acelerando os ciclos de desenvolvimento e reduzindo os tempos de teste. Esta publicação demonstra como implementar testes para tipos de estado, incluindo estados Map com limites de tolerância, mecanismos de repetição com backoff exponencial e padrões waitForTaskToken com simulação de objeto de contexto usando integrações simuladas para testes isolados.

Ao integrar testes da API TestState em pipelines de CI/CD, você pode validar a lógica do workflow antes da implantação, reduzindo o risco de problemas em produção. O exemplo de workflow do GitHub Actions demonstra uma implementação que executa testes e implanta recursos em sequência controlada. Os exemplos de código completos e o framework de teste estão disponíveis no repositório do GitHub para implementar práticas de teste semelhantes para workflows do Step Functions.

Este conteúdo é uma tradução da publicação original em inglês, que pode ser encontrada aqui.

Biografia do Autores

	D Surya Sai é um Technical Account Manager na Amazon Web Services, ajudando clientes a construir soluções na AWS.
	Sahithi Ginjupalli é uma Cloud Support Engeneer na Amazon Web Services.

Biografia do tradutores

Daniel Abib é Arquiteto de Soluções Sênior e Especialista em Amazon Bedrock na AWS, com mais de 25 anos trabalhando com gerenciamento de projetos, arquiteturas de soluções escaláveis, desenvolvimento de sistemas e CI/CD, microsserviços, arquitetura Serverless & Containers e especialização em Machine Learning. Ele trabalha apoiando Startups, ajudando-os em sua jornada para a nuvem.

https://www.linkedin.com/in/danielabib/

Rodrigo Peres é Arquiteto de Soluções na AWS, com mais de 20 anos de experiência trabalhando com arquitetura de soluções, desenvolvimento de sistemas e modernização de sistemas legados.

O blog da AWS