KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use respond_activity_task_canceled method in localstack

Best Python code snippet using localstack_python

layer2.py

Source:layer2.py

...140    def cancel(self, task_token=None, details=None):141        """RespondActivityTaskCanceled."""142        if task_token is None:143            task_token = self.last_tasktoken144        return self._swf.respond_activity_task_canceled(task_token, details)145    @wraps(Layer1.respond_activity_task_completed)146    def complete(self, task_token=None, result=None):147        """RespondActivityTaskCompleted."""148        if task_token is None:149            task_token = self.last_tasktoken150        return self._swf.respond_activity_task_completed(task_token, result)151    @wraps(Layer1.respond_activity_task_failed)152    def fail(self, task_token=None, details=None, reason=None):153        """RespondActivityTaskFailed."""154        if task_token is None:155            task_token = self.last_tasktoken156        return self._swf.respond_activity_task_failed(task_token, details,157                                                      reason)158    @wraps(Layer1.record_activity_task_heartbeat)...

test_layer2_actors.py

Source:test_layer2_actors.py

1import boto.swf.layer22from boto.swf.layer2 import Decider, ActivityWorker3from tests.unit import unittest4from mock import Mock5class TestActors(unittest.TestCase):6    def setUp(self):7        boto.swf.layer2.Layer1 = Mock()8        self.worker = ActivityWorker(name='test-worker', domain='test', task_list='test_list')9        self.decider = Decider(name='test-worker', domain='test', task_list='test_list')10        self.worker._swf = Mock()11        self.decider._swf = Mock()12    13    def test_decider_pass_tasktoken(self):14        self.decider._swf.poll_for_decision_task.return_value = {15            'events': [{'eventId': 1,16             'eventTimestamp': 1379019427.953,17             'eventType': 'WorkflowExecutionStarted',18             'workflowExecutionStartedEventAttributes': {19                'childPolicy': 'TERMINATE',20                'executionStartToCloseTimeout': '3600',21                'parentInitiatedEventId': 0,22                'taskList': {'name': 'test_list'},23                'taskStartToCloseTimeout': '123',24                'workflowType': {'name': 'test_workflow_name',25                'version': 'v1'}}},26            {'decisionTaskScheduledEventAttributes': 27                {'startToCloseTimeout': '123',28                'taskList': {'name': 'test_list'}},29             'eventId': 2,30             'eventTimestamp': 1379019427.953,31             'eventType': 'DecisionTaskScheduled'},32            {'decisionTaskStartedEventAttributes': {'scheduledEventId': 2},33             'eventId': 3, 'eventTimestamp': 1379019495.585,34             'eventType': 'DecisionTaskStarted'}],35             'previousStartedEventId': 0, 'startedEventId': 3,36             'taskToken': 'my_specific_task_token',37             'workflowExecution': {'runId': 'fwr243dsa324132jmflkfu0943tr09=',38                       'workflowId': 'test_workflow_name-v1-1379019427'},39             'workflowType': {'name': 'test_workflow_name', 'version': 'v1'}}40        self.decider.poll()41        self.decider.complete()42        self.decider._swf.respond_decision_task_completed.assert_called_with('my_specific_task_token', None)43        self.assertEqual('my_specific_task_token', self.decider.last_tasktoken)44    def test_worker_pass_tasktoken(self):45        task_token = 'worker_task_token'46        self.worker._swf.poll_for_activity_task.return_value = {47            'activityId': 'SomeActivity-1379020713',48            'activityType': {'name': 'SomeActivity', 'version': '1.0'},49            'startedEventId': 6,50            'taskToken': task_token,51            'workflowExecution': {'runId': '12T026NzGK5c4eMti06N9O3GHFuTDaNyA+8LFtoDkAwfE=',52               'workflowId': 'MyWorkflow-1.0-1379020705'}}53        self.worker.poll()54        self.worker.cancel(details='Cancelling!')55        self.worker.complete(result='Done!')56        self.worker.fail(reason='Failure!')57        self.worker.heartbeat() 58        self.worker._swf.respond_activity_task_canceled.assert_called_with(task_token, 'Cancelling!')59        self.worker._swf.respond_activity_task_completed.assert_called_with(task_token, 'Done!')60        self.worker._swf.respond_activity_task_failed.assert_called_with(task_token, None, 'Failure!')61        self.worker._swf.record_activity_task_heartbeat.assert_called_with(task_token, None)62    def test_actor_poll_without_tasklist_override(self):63        self.worker.poll()64        self.decider.poll()65        self.worker._swf.poll_for_activity_task.assert_called_with('test', 'test_list')66        self.decider._swf.poll_for_decision_task.assert_called_with('test', 'test_list')67    def test_worker_override_tasklist(self):68        self.worker.poll(task_list='some_other_tasklist')69        self.worker._swf.poll_for_activity_task.assert_called_with('test', 'some_other_tasklist')70    def test_decider_override_tasklist(self):71        self.decider.poll(task_list='some_other_tasklist')72        self.decider._swf.poll_for_decision_task.assert_called_with('test', 'some_other_tasklist')73if __name__ == '__main__':...

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.