Tác giả: D Surya Sai và Sahithi Ginjupalli
Ngày phát hành: 22 MAR 2026
Chuyên mục: AWS Step Functions, Compute
AWS Step Functions gần đây đã công bố những cải tiến mới cho khả năng kiểm thử cục bộ của Step Functions, giới thiệu kiểm thử dựa trên API mà các nhà phát triển có thể sử dụng để xác thực quy trình làm việc trước khi triển khai lên AWS. Như đã trình bày chi tiết trong bài đăng blog của chúng tôi, API TestState biến đổi quá trình phát triển Step Functions bằng cách cho phép kiểm thử từng trạng thái một cách độc lập hoặc dưới dạng các quy trình làm việc hoàn chỉnh. Điều này hỗ trợ các phản hồi giả lập (mocked responses) và tích hợp dịch vụ AWS thực tế, đồng thời cung cấp các khả năng nâng cao. Các khả năng này bao gồm các trạng thái Map/Parallel, mô phỏng lỗi với cơ chế thử lại, xác thực đối tượng ngữ cảnh (context object), và siêu dữ liệu kiểm tra chi tiết để kiểm thử cục bộ toàn diện ứng dụng serverless của bạn.
API TestState có thể được truy cập thông qua nhiều giao diện như AWS Command Line Interface (AWS CLI), AWS SDK, LocalStack. Theo mặc định, API TestState trong AWS CLI và SDK chạy trên điểm cuối AWS từ xa, cung cấp xác thực đối với cơ sở hạ tầng dịch vụ Step Functions thực tế. Chúng tôi đã hợp tác với LocalStack để cung cấp một điểm cuối kiểm thử bổ sung cho API TestState. Các nhà phát triển có thể sử dụng LocalStack để kiểm thử đơn vị (unit testing) các quy trình làm việc của họ bằng cách thay đổi cấu hình điểm cuối của client AWS SDK để trỏ đến LocalStack: `http://localhost.localstack.cloud:4566/` thay vì điểm cuối AWS. Cách tiếp cận này cung cấp sự cô lập mạng hoàn toàn khi cần thiết. Để có trải nghiệm phát triển hợp lý, bạn cũng có thể sử dụng tiện ích mở rộng LocalStack VSCode để tự động cấu hình môi trường của bạn trỏ đến điểm cuối LocalStack. Cách tiếp cận này được trình bày chi tiết trong bài đăng blog của AWS.
Bài đăng blog này trình bày cách xây dựng các bộ kiểm thử để kiểm thử đơn vị các quy trình làm việc của Step Functions bằng AWS SDK for Python sử dụng framework pytest. Việc triển khai hoàn chỉnh có sẵn trong kho lưu trữ GitHub.
Xây dựng các trường hợp kiểm thử bằng API TestState
Quy trình làm việc ví dụ này triển khai một hệ thống xử lý đơn hàng thương mại điện tử thực tế sử dụng JSONata để chuyển đổi dữ liệu nâng cao. Nó kết hợp các mẫu Step Functions phức tạp bao gồm các trạng thái Map phân tán, thực thi Parallel và cơ chế callback waitForTaskToken. Quy trình này xác thực các đơn hàng thông qua các hàm AWS Lambda, phân phối xử lý các mục đơn hàng với khả năng chịu lỗi có thể cấu hình, chạy cập nhật thanh toán và kho hàng song song, xử lý các quy trình phê duyệt của con người bằng cách sử dụng task token, sau đó lưu trữ các đơn hàng trong Amazon DynamoDB với việc gửi thông báo. Quy trình làm việc này trình bày cách xử lý lỗi nâng cao với nhiều Catchers và Retriers, exponential backoff cho việc điều tiết Lambda và giới hạn DynamoDB, và các chuyển đổi trạng thái tinh vi mà trước đây rất khó kiểm thử cục bộ. Điều này làm cho nó trở thành lựa chọn được khuyến nghị để trình bày việc sử dụng các tính năng kiểm thử cục bộ của API TestState được cải tiến.
Quy trình làm việc hoàn chỉnh có sẵn trong kho lưu trữ GitHub, nơi bạn có thể kiểm tra định nghĩa máy trạng thái đầy đủ và xem cách các biểu thức JSONata xử lý chuyển đổi dữ liệu trong suốt luồng thực thi.
Hình 1: Quy trình làm việc của máy trạng thái minh họa một hệ thống xử lý đơn hàng thương mại điện tử thực tế.
Kiểm thử Step Functions hiệu quả đòi hỏi một cách tiếp cận có hệ thống để tích hợp API TestState, cung cấp khả năng xác thực trạng thái, mô phỏng lỗi và xác nhận. Framework kiểm thử được xây dựng bằng framework pytest của Python, sử dụng fixtures để tự động cung cấp các phiên bản runner được cấu hình sẵn xử lý khởi tạo client API TestState và tải định nghĩa máy trạng thái. Điều này loại bỏ mã thiết lập lặp đi lặp lại và cung cấp môi trường kiểm thử nhất quán. API TestState được cải tiến hỗ trợ cả tích hợp giả lập (mock integrations) và tích hợp thực tế với các dịch vụ AWS, cung cấp sự linh hoạt trong các chiến lược kiểm thử. Đối với bản trình diễn này, bạn sử dụng các tích hợp giả lập để trình bày cách có thể đạt được kiểm thử cục bộ hoàn chỉnh mà không cần triển khai bất kỳ tài nguyên nào vào tài khoản AWS.
Framework này được xây dựng cho mục đích trình diễn, và bạn có thể xây dựng các framework kiểm thử của riêng mình tương tự bằng các ngôn ngữ lập trình khác như Java, Node.js. Framework kiểm thử sử dụng các mẫu method chaining để tạo các trường hợp kiểm thử dễ đọc với các phương thức xác nhận toàn diện, tự động output chaining giữa các lần thực thi trạng thái, và mô phỏng lỗi để kiểm thử cơ chế thử lại, khoảng thời gian backoff và các khối catch trên các điều kiện lỗi của dịch vụ AWS.
Các triển khai kiểm thử sau đây trình bày các khả năng kiểm thử có thể đạt được với API TestState được cải tiến trong môi trường phát triển cục bộ. Các trường hợp kiểm thử được chạy trên Statemachine đã nói ở trên.
Trường hợp kiểm thử 1: Kiểm thử cơ chế điều tiết Lambda và thử lại
Tích hợp dịch vụ với Statemachines như AWS Lambda, Amazon DynamoDB có thể gặp phải tình trạng điều tiết (throttling) tùy thuộc vào mức độ sử dụng của chúng. Một khả năng chính của API TestState được cải tiến là khả năng mô phỏng cơ chế thử lại với quyền kiểm soát số lần thử lại và khoảng thời gian backoff. Kiểm thử này trình bày các khả năng kiểm thử thử lại của API TestState được cải tiến thông qua tham số stateConfiguration.retrierRetryCount và các trường phản hồi inspectionData.errorDetails. Trường phản hồi này cung cấp retryBackoffIntervalSeconds để xác thực các phép tính exponential backoff, retryIndex để theo dõi chuỗi các lần thử lại, và catchIndex để xác định trình xử lý lỗi nào đã xử lý ngoại lệ. Các khả năng kiểm tra nâng cao này cho phép xác thực logic thử lại, chiến lược backoff, và các mẫu lan truyền lỗi trên các quy trình làm việc của máy trạng thái phức tạp.
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 calculationresponse = 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"))
Trường hợp kiểm thử 2: Kiểm thử trạng thái Map với ngưỡng dung sai
Các trạng thái Map phân tán đặt ra những thách thức kiểm thử độc đáo do tính chất xử lý song song và khả năng chịu lỗi của chúng. API TestState được cải tiến cung cấp các tùy chọn cấu hình chuyên biệt để kiểm thử các kịch bản phức tạp này.
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 executionmap_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 scenariotolerance_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"))
Kiểm thử này trình bày các khả năng kiểm thử trạng thái Map của API TestState được cải tiến thông qua tham số stateConfiguration.mapIterationFailureCount để mô phỏng các lỗi lặp. API cung cấp dữ liệu kiểm tra toàn diện bao gồm inspectionData.afterItemSelector để xác thực các chuyển đổi ItemSelector, inspectionData.afterItemBatcher để xác thực xử lý hàng loạt, inspectionData.toleratedFailureCount và inspectionData.toleratedFailurePercentage để xác minh ngưỡng. Khi số lượng lỗi được chỉ định vượt quá ngưỡng dung sai đã cấu hình, API sẽ trả về chính xác States.ExceedToleratedFailureThreshold, cho phép kiểm thử các mẫu khả năng phục hồi của trạng thái Map.
Trường hợp kiểm thử 3: Kiểm thử mẫu WaitForCallback
Tích hợp waitForCallback yêu cầu xây dựng đối tượng ngữ cảnh (context object) để mô phỏng môi trường thực thi thực tế, đặc biệt đối với các quy trình phê duyệt của con người.
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 correctlyresponse = runner.get_response()after_args = json.loads(response['inspectionData']['afterArguments'])assert after_args['Payload']['taskToken'] == context_data['Task']['Token']
Kiểm thử này trình bày sự hỗ trợ của API TestState được cải tiến cho các tích hợp waitForCallback thông qua tham số context để mô phỏng đối tượng Context thực tế. API cho phép kiểm thử toàn diện các biểu thức JSONata tham chiếu $states.context.Task.Token, $states.context.Execution.Id và các trường ngữ cảnh khác. Trường phản hồi inspectionData.afterArguments xác thực rằng các biểu thức JSONata đã xử lý dữ liệu ngữ cảnh một cách chính xác, trong khi API tự động xử lý sự phức tạp của việc nhúng task token vào các payload tích hợp dịch vụ cho các kịch bản kiểm thử waitForCallback.
Trường hợp kiểm thử 4: Kiểm thử luồng thành công (Happy path) – xác thực quy trình làm việc hoàn chỉnh
Kiểm thử luồng thành công (Happy path testing) xác thực rằng các quy trình làm việc thực thi chính xác trong điều kiện hoạt động bình thường. API TestState được cải tiến cho phép bạn xâu chuỗi các lần thực thi trạng thái lại với nhau, tự động chuyển đầu ra giữa các trạng thái để mô phỏng một lần thực thi quy trình làm việc hoàn chỉnh.
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"))
Kiểm thử này trình bày cách API TestState duy trì ngữ cảnh trạng thái giữa các lần thực thi, cho phép mô phỏng quy trình làm việc thực tế. Phương thức get_output() truy xuất đầu ra đã xử lý từ một trạng thái để sử dụng làm đầu vào cho trạng thái tiếp theo, mô phỏng hành vi thực thi Step Functions thực tế.
Lưu ý: Đoạn mã trên chỉ hiển thị hai trạng thái đầu tiên của kiểm thử quy trình làm việc hoàn chỉnh để ngắn gọn. Mã kiểm thử đầy đủ với tất cả các trạng thái (ProcessOrderItems, ParallelProcessing, WaitForApproval, CheckApproval, SaveOrderDetails, và SendNotification) có thể được xem trong kho lưu trữ GitHub hoàn chỉnh, trình bày xác thực quy trình làm việc từ đầu đến cuối bằng cách sử dụng cùng một mẫu method chaining.
Tích hợp với các quy trình CI/CD hiện đại
Trong phần này, chúng ta sẽ khám phá cách tích hợp các kiểm thử đơn vị trước đó vào một quy trình CI/CD để cho phép kiểm thử cục bộ.
Kho lưu trữ mẫu bao gồm một quy trình GitHub Actions minh họa cách kiểm thử API TestState tích hợp vào các quy trình tích hợp liên tục và phân phối liên tục (CI/CD). Quy trình làm việc (.github/workflows/test-and-deploy.yml) cung cấp một quy trình hai bước xác thực trước khi bất kỳ tài nguyên AWS nào được triển khai bằng AWS Serverless Application Model (AWS SAM).
Quy trình CI/CD tuân theo mẫu sau:
- Unit Tests: Thực thi bộ kiểm thử API TestState hoàn chỉnh bằng cách sử dụng
pytest tests/unit_test.py -v - SAM Deploy: Triển khai tài nguyên AWS bằng cách sử dụng sam build và sam deploy
Để cho phép quy trình GitHub Actions triển khai tài nguyên vào tài khoản AWS của bạn, hãy cấu hình các thông tin xác thực AWS này trong cài đặt kho lưu trữ GitHub của bạn. Để biết hướng dẫn thiết lập chi tiết, hãy xem bài đăng blog của AWS.
Sau đây là các secret cần thiết để được cấu hình trong cài đặt kho lưu trữ GitHub:
AWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEYAWS_REGION
Trong môi trường sản xuất, bạn thường có thể mở rộng quy trình cơ bản này để bao gồm các giai đoạn bổ sung. Quy trình nâng cao thường bắt đầu bằng việc triển khai vào một tài khoản phát triển trước, sau đó là kiểm thử tích hợp đối với các tài nguyên đã triển khai. Giai đoạn cuối cùng bao gồm việc chuyển sang sản xuất với các cổng phê duyệt và kiểm tra tuân thủ quét bảo mật phù hợp.
Kết luận
API TestState được cải tiến cho phép kiểm thử các quy trình làm việc của Step Functions cục bộ mà không yêu cầu triển khai AWS, giúp tăng tốc chu kỳ phát triển và giảm thời gian kiểm thử. Bài đăng này trình bày cách triển khai kiểm thử cho các loại trạng thái bao gồm các trạng thái Map với ngưỡng dung sai, cơ chế thử lại với exponential backoff, và các mẫu waitForTaskToken với mô phỏng đối tượng ngữ cảnh sử dụng các tích hợp giả lập để kiểm thử cô lập.
Bằng cách tích hợp kiểm thử API TestState vào các quy trình CI/CD, bạn có thể xác thực logic quy trình làm việc trước khi triển khai, giảm thiểu rủi ro các vấn đề trong sản xuất. Ví dụ về quy trình GitHub Actions minh họa một triển khai chạy các kiểm thử và triển khai tài nguyên theo một trình tự được kiểm soát. Các ví dụ mã hoàn chỉnh và framework kiểm thử có sẵn trong kho lưu trữ GitHub để triển khai các phương pháp kiểm thử tương tự cho các quy trình làm việc của Step Functions.
AWS Study Group 