简短的描述
pytest-stepfunctions的Python项目详细描述
pytest阶跃函数
pytest fixture使您能够在AWS StepFunctions本地测试期间模拟Lambda代码。在
目录
概述
AWS提供了本地Step函数作为JAR和Docker映像，以便在没有部署的情况下进行快速测试。他们还描述了如何在this article中执行这样的任务。一开始我很兴奋，但很快就因为无法模拟Lambda函数的外部依赖性而沮丧。然后我想：如果用一个假的Lambda服务启动一个Python线程，然后使用这个伪服务来执行Lambda函数呢？幸运的是，它起作用了！在
安装
使用pip安装：
$ pip install pytest-stepfunctions
入门
假设有一个状态机简单地收集所有的EMR集群唯一标识符。以下是状态机定义：
^{pr2}$以及Lambda代码my/pkg/emr.py：
importboto3deflist_ids(*args,**kwargs):emr_client=boto3.client("emr")response=emr_client.list_clusters()return[item["Id"]foriteminresponse["Clusters"]]
创建状态机
在测试文件tests/test_foo.py中，创建一个Step函数客户机，其端点URL指向我们的Step函数服务，并使用该客户机通过上面的定义创建一个状态机资源
fromstringimportTemplateimportboto3deftest_bar(aws_stepfunctions_endpoint_url):definition_template=Template(""" { "StartAt": "ListIds", "States": { "ListIds": { "Type": "Task", "Resource": "${ListIdsLambdaArn}", "ResultPath": "$.cluster_ids", "End": true } } } """)list_ids_lambda_arn="arn:aws:lambda:us-east-1:123456789012:function:my.pkg.emr.list_ids"definition=definition_template.safe_substitute(ListIdsLambdaArn=list_ids_lambda_arn)sfn_client=boto3.client("stepfunctions",endpoint_url=aws_stepfunctions_endpoint_url)state_machine_arn=sfn_client.create_state_machine(name="list-ids",definition=definition,roleArn="arn:aws:iam::012345678901:role/DummyRole")["stateMachineArn"]
注意pytest stepfunctions中的内部假Lambda服务将解析Lambda ARNs以识别调用什么。在
在Lambda代码
中模拟EMR客户端这里使用pytest-mock夹具临时修补Lambda代码中的boto3模块。botocore.stub.Stubber还用于确保模拟请求参数和响应内容都有效：
frombotocore.stubimportStubberdeftest_bar(aws_stepfunctions_endpoint_url,mocker):...emr_client=boto3.client("emr")mocker.patch("my.pkg.emr.boto3",autospec=True).client.return_value=emr_clientstubber=Stubber(emr_client)stubber.add_response("list_clusters",service_response={"Clusters":[{"Id":"j-00001"},{"Id":"j-00002"}]})
开始执行并验证结果
启动并等待执行状态不是RUNNING：
importjsonimporttimedeftest_bar(aws_stepfunctions_endpoint_url,mocker):...execution_arn=sfn_client.start_execution(stateMachineArn=state_machine_arn,name="list-ids-exec",input="{}")["executionArn"]withstubber:whileTrue:response=sfn_client.describe_execution(executionArn=execution_arn)ifresponse["status"]!="RUNNING":breaktime.sleep(0.5)stubber.assert_no_pending_responses()assert"SUCCEEDED"==response["status"]assert["j-00001","j-00002"]==json.loads(response["output"])["cluster_ids"]
使用步骤函数JAR
运行测试这个罐子是可用的here。先下载并执行：
$ java -jar /path/to/StepFunctionsLocal.jar \ --lambda-endpoint http://localhost:13000 \ --step-functions-endpoint http://localhost:8083 \ --wait-time-scale 0 Step Functions Local Version: 1.4.0 Build: 2019-09-18 2020-07-06 18:40:28.284: Configure [Lambda Endpoint] to [http://localhost:13000]2020-07-06 18:40:28.285: Configure [Step Functions Endpoint] to [http://localhost:8083]2020-07-06 18:40:28.323: Loaded credentials from profile: default 2020-07-06 18:40:28.324: Starting server on port 8083 with account 123456789012, region us-east-1
然后使用以下命令运行测试：
$ python -m pytest -v \ --pytest-stepfunctions-endpoint-url=http://0.0.0.0:8083 \ --pytest-stepfunctions-lambda-address=0.0.0.0 \ --pytest-stepfunctions-lambda-port=13000\ ./tests ===============================test session starts================================ platform linux -- Python 3.7.3, pytest-5.4.3, py-1.9.0, pluggy-0.13.1 -- /tmp/gg/venv/bin/python cachedir: .pytest_cache rootdir: /tmp/gg plugins: mock-3.1.1, stepfunctions-0.1a2 collected 1 item tests/test_foo.py::test_bar PASSED [100%]================================1 passed in 1.01s =================================
使用步骤函数Docker Image
我个人推荐这种方法，因为它更容易复制测试环境。在
这是Dockerfile
FROMpython:3.7WORKDIR /appCOPY ./my ./my COPY ./tests ./tests RUN pip install pytest pytest-stepfunctions pytest-mock boto3
以及Docker的docker-compose.yml组合：
version:"3.2"services:tester:build:context:.dockerfile:./Dockerfileenvironment:AWS_DEFAULT_REGION:us-east-1AWS_ACCESS_KEY_ID:xxxAWS_SECRET_ACCESS_KEY:xxxcommand:>bash -c "python -m pytest -v--pytest-stepfunctions-endpoint-url=http://sfn-endpoint:8083--pytest-stepfunctions-lambda-address=0.0.0.0--pytest-stepfunctions-lambda-port=13000./tests"sfn-endpoint:image:amazon/aws-stepfunctions-local:1.5.1environment:AWS_DEFAULT_REGION:us-east-1AWS_ACCESS_KEY_ID:xxxAWS_SECRET_ACCESS_KEY:xxxWAIT_TIME_SCALE:0STEP_FUNCTIONS_ENDPOINT:http://sfn-endpoint:8083LAMBDA_ENDPOINT:http://tester:13000
然后运行以下命令运行测试：
$ docker-compose up --build --exit-code-from tester
已知问题
- 嵌套工作流非常慢：如果一个状态机包含许多嵌套的状态机，那么即使
WAIT_TIME_SCALE设置为0，执行也会非常缓慢。这是官方JAR中已知的性能问题。在 - 还不支持Lambda以外的AWS服务集成。像EMR这样的服务在官方JAR中甚至没有端点选项。对于某些情况，一种可能的解决方法是通过调用Lambda函数来调用它们。在
- 项目
标签：
欢迎加入QQ群-->： 979659372
