test_activities.py 文件源码

def test_poll_task(self, mCreateSession):
        iSession = MockSession()
        mCreateSession.return_value = (iSession, '123456')
        client = iSession.client('stepfunctions')
        client.get_activity_task.return_value = {
            'taskToken': 'YYY',
            'input': '{}'
        }

        activity = ActivityMixin()
        activity.arn = 'XXX'

        token,  input_ = activity.poll_task('worker')

        self.assertEqual(token, 'YYY')
        self.assertEqual(input_, {})

        calls = [
            mock.call.get_activity_task(activityArn = 'XXX',
                                        workerName = 'worker')
        ]
        self.assertEqual(client.mock_calls, calls)