Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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

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

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.