test_activities.py 文件源码

def test_run_exception(self, mCreateSession):
        iSession = MockSession()
        mCreateSession.return_value = (iSession, '123456')
        client = iSession.client('stepfunctions')

        target = mock.MagicMock()
        target.side_effect = BossError('cause')

        task = TaskMixin(process = target)
        task.handle_task('token', None)

        self.assertEqual(task.token, None)
        call = mock.call.send_task_failure(taskToken = 'token',
                                           error = 'BossError',
                                           cause = 'cause')
        self.assertEqual(client.mock_calls, [call])