How to use poll_for_decision_task method in localstack

Best Python code snippet using localstack_python

test_decision_tasks.py

Source:test_decision_tasks.py Github

copy

Full Screen

...12 "test-domain", conn.run_id, "uid-abcd1234"13 )14 types = [evt["eventType"] for evt in resp["events"]]15 types.should.equal(["WorkflowExecutionStarted", "DecisionTaskScheduled"])16 resp = conn.poll_for_decision_task("test-domain", "queue", identity="srv01")17 types = [evt["eventType"] for evt in resp["events"]]18 types.should.equal(19 ["WorkflowExecutionStarted", "DecisionTaskScheduled", "DecisionTaskStarted"]20 )21 resp["events"][-1]["decisionTaskStartedEventAttributes"]["identity"].should.equal(22 "srv01"23 )24@mock_swf_deprecated25def test_poll_for_decision_task_previous_started_event_id():26 conn = setup_workflow()27 resp = conn.poll_for_decision_task("test-domain", "queue")28 assert resp["workflowExecution"]["runId"] == conn.run_id29 assert "previousStartedEventId" not in resp30 # Require a failing decision, in this case a non-existant activity type31 attrs = {32 "activityId": "spam",33 "activityType": {"name": "test-activity", "version": "v1.42"},34 "taskList": "eggs",35 }36 decision = {37 "decisionType": "ScheduleActivityTask",38 "scheduleActivityTaskDecisionAttributes": attrs,39 }40 conn.respond_decision_task_completed(resp["taskToken"], decisions=[decision])41 resp = conn.poll_for_decision_task("test-domain", "queue")42 assert resp["workflowExecution"]["runId"] == conn.run_id43 assert resp["previousStartedEventId"] == 344@mock_swf_deprecated45def test_poll_for_decision_task_when_none():46 conn = setup_workflow()47 conn.poll_for_decision_task("test-domain", "queue")48 resp = conn.poll_for_decision_task("test-domain", "queue")49 # this is the DecisionTask representation you get from the real SWF50 # after waiting 60s when there's no decision to be taken51 resp.should.equal({"previousStartedEventId": 0, "startedEventId": 0})52@mock_swf_deprecated53def test_poll_for_decision_task_on_non_existent_queue():54 conn = setup_workflow()55 resp = conn.poll_for_decision_task("test-domain", "non-existent-queue")56 resp.should.equal({"previousStartedEventId": 0, "startedEventId": 0})57@mock_swf_deprecated58def test_poll_for_decision_task_with_reverse_order():59 conn = setup_workflow()60 resp = conn.poll_for_decision_task("test-domain", "queue", reverse_order=True)61 types = [evt["eventType"] for evt in resp["events"]]62 types.should.equal(63 ["DecisionTaskStarted", "DecisionTaskScheduled", "WorkflowExecutionStarted"]64 )65# CountPendingDecisionTasks endpoint66@mock_swf_deprecated67def test_count_pending_decision_tasks():68 conn = setup_workflow()69 conn.poll_for_decision_task("test-domain", "queue")70 resp = conn.count_pending_decision_tasks("test-domain", "queue")71 resp.should.equal({"count": 1, "truncated": False})72@mock_swf_deprecated73def test_count_pending_decision_tasks_on_non_existent_task_list():74 conn = setup_workflow()75 resp = conn.count_pending_decision_tasks("test-domain", "non-existent")76 resp.should.equal({"count": 0, "truncated": False})77@mock_swf_deprecated78def test_count_pending_decision_tasks_after_decision_completes():79 conn = setup_workflow()80 resp = conn.poll_for_decision_task("test-domain", "queue")81 conn.respond_decision_task_completed(resp["taskToken"])82 resp = conn.count_pending_decision_tasks("test-domain", "queue")83 resp.should.equal({"count": 0, "truncated": False})84# RespondDecisionTaskCompleted endpoint85@mock_swf_deprecated86def test_respond_decision_task_completed_with_no_decision():87 conn = setup_workflow()88 resp = conn.poll_for_decision_task("test-domain", "queue")89 task_token = resp["taskToken"]90 resp = conn.respond_decision_task_completed(91 task_token, execution_context="free-form context"92 )93 resp.should.be.none94 resp = conn.get_workflow_execution_history(95 "test-domain", conn.run_id, "uid-abcd1234"96 )97 types = [evt["eventType"] for evt in resp["events"]]98 types.should.equal(99 [100 "WorkflowExecutionStarted",101 "DecisionTaskScheduled",102 "DecisionTaskStarted",103 "DecisionTaskCompleted",104 ]105 )106 evt = resp["events"][-1]107 evt["decisionTaskCompletedEventAttributes"].should.equal(108 {109 "executionContext": "free-form context",110 "scheduledEventId": 2,111 "startedEventId": 3,112 }113 )114 resp = conn.describe_workflow_execution("test-domain", conn.run_id, "uid-abcd1234")115 resp["latestExecutionContext"].should.equal("free-form context")116@mock_swf_deprecated117def test_respond_decision_task_completed_with_wrong_token():118 conn = setup_workflow()119 conn.poll_for_decision_task("test-domain", "queue")120 conn.respond_decision_task_completed.when.called_with(121 "not-a-correct-token"122 ).should.throw(SWFResponseError)123@mock_swf_deprecated124def test_respond_decision_task_completed_on_close_workflow_execution():125 conn = setup_workflow()126 resp = conn.poll_for_decision_task("test-domain", "queue")127 task_token = resp["taskToken"]128 # bad: we're closing workflow execution manually, but endpoints are not129 # coded for now..130 wfe = swf_backend.domains[0].workflow_executions[-1]131 wfe.execution_status = "CLOSED"132 # /bad133 conn.respond_decision_task_completed.when.called_with(task_token).should.throw(134 SWFResponseError135 )136@mock_swf_deprecated137def test_respond_decision_task_completed_with_task_already_completed():138 conn = setup_workflow()139 resp = conn.poll_for_decision_task("test-domain", "queue")140 task_token = resp["taskToken"]141 conn.respond_decision_task_completed(task_token)142 conn.respond_decision_task_completed.when.called_with(task_token).should.throw(143 SWFResponseError144 )145@mock_swf_deprecated146def test_respond_decision_task_completed_with_complete_workflow_execution():147 conn = setup_workflow()148 resp = conn.poll_for_decision_task("test-domain", "queue")149 task_token = resp["taskToken"]150 decisions = [151 {152 "decisionType": "CompleteWorkflowExecution",153 "completeWorkflowExecutionDecisionAttributes": {"result": "foo bar"},154 }155 ]156 resp = conn.respond_decision_task_completed(task_token, decisions=decisions)157 resp.should.be.none158 resp = conn.get_workflow_execution_history(159 "test-domain", conn.run_id, "uid-abcd1234"160 )161 types = [evt["eventType"] for evt in resp["events"]]162 types.should.equal(163 [164 "WorkflowExecutionStarted",165 "DecisionTaskScheduled",166 "DecisionTaskStarted",167 "DecisionTaskCompleted",168 "WorkflowExecutionCompleted",169 ]170 )171 resp["events"][-1]["workflowExecutionCompletedEventAttributes"][172 "result"173 ].should.equal("foo bar")174@mock_swf_deprecated175def test_respond_decision_task_completed_with_close_decision_not_last():176 conn = setup_workflow()177 resp = conn.poll_for_decision_task("test-domain", "queue")178 task_token = resp["taskToken"]179 decisions = [180 {"decisionType": "CompleteWorkflowExecution"},181 {"decisionType": "WeDontCare"},182 ]183 conn.respond_decision_task_completed.when.called_with(184 task_token, decisions=decisions185 ).should.throw(SWFResponseError, r"Close must be last decision in list")186@mock_swf_deprecated187def test_respond_decision_task_completed_with_invalid_decision_type():188 conn = setup_workflow()189 resp = conn.poll_for_decision_task("test-domain", "queue")190 task_token = resp["taskToken"]191 decisions = [192 {"decisionType": "BadDecisionType"},193 {"decisionType": "CompleteWorkflowExecution"},194 ]195 conn.respond_decision_task_completed.when.called_with(196 task_token, decisions=decisions197 ).should.throw(198 SWFResponseError,199 r"Value 'BadDecisionType' at 'decisions.1.member.decisionType'",200 )201@mock_swf_deprecated202def test_respond_decision_task_completed_with_missing_attributes():203 conn = setup_workflow()204 resp = conn.poll_for_decision_task("test-domain", "queue")205 task_token = resp["taskToken"]206 decisions = [207 {208 "decisionType": "should trigger even with incorrect decision type",209 "startTimerDecisionAttributes": {},210 }211 ]212 conn.respond_decision_task_completed.when.called_with(213 task_token, decisions=decisions214 ).should.throw(215 SWFResponseError,216 r"Value null at 'decisions.1.member.startTimerDecisionAttributes.timerId' "217 r"failed to satisfy constraint: Member must not be null",218 )219@mock_swf_deprecated220def test_respond_decision_task_completed_with_missing_attributes_totally():221 conn = setup_workflow()222 resp = conn.poll_for_decision_task("test-domain", "queue")223 task_token = resp["taskToken"]224 decisions = [{"decisionType": "StartTimer"}]225 conn.respond_decision_task_completed.when.called_with(226 task_token, decisions=decisions227 ).should.throw(228 SWFResponseError,229 r"Value null at 'decisions.1.member.startTimerDecisionAttributes.timerId' "230 r"failed to satisfy constraint: Member must not be null",231 )232@mock_swf_deprecated233def test_respond_decision_task_completed_with_fail_workflow_execution():234 conn = setup_workflow()235 resp = conn.poll_for_decision_task("test-domain", "queue")236 task_token = resp["taskToken"]237 decisions = [238 {239 "decisionType": "FailWorkflowExecution",240 "failWorkflowExecutionDecisionAttributes": {241 "reason": "my rules",242 "details": "foo",243 },244 }245 ]246 resp = conn.respond_decision_task_completed(task_token, decisions=decisions)247 resp.should.be.none248 resp = conn.get_workflow_execution_history(249 "test-domain", conn.run_id, "uid-abcd1234"250 )251 types = [evt["eventType"] for evt in resp["events"]]252 types.should.equal(253 [254 "WorkflowExecutionStarted",255 "DecisionTaskScheduled",256 "DecisionTaskStarted",257 "DecisionTaskCompleted",258 "WorkflowExecutionFailed",259 ]260 )261 attrs = resp["events"][-1]["workflowExecutionFailedEventAttributes"]262 attrs["reason"].should.equal("my rules")263 attrs["details"].should.equal("foo")264@mock_swf_deprecated265@freeze_time("2015-01-01 12:00:00")266def test_respond_decision_task_completed_with_schedule_activity_task():267 conn = setup_workflow()268 resp = conn.poll_for_decision_task("test-domain", "queue")269 task_token = resp["taskToken"]270 decisions = [271 {272 "decisionType": "ScheduleActivityTask",273 "scheduleActivityTaskDecisionAttributes": {274 "activityId": "my-activity-001",275 "activityType": {"name": "test-activity", "version": "v1.1"},276 "heartbeatTimeout": "60",277 "input": "123",278 "taskList": {"name": "my-task-list"},279 },280 }281 ]282 resp = conn.respond_decision_task_completed(task_token, decisions=decisions)...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful