test_swf.py 文件源码

def test_register_type_does_not_raise(self, mocker):
        error_response = {'Error':{'Code':'TypeAlreadyExistsFault'}}
        client_error = botocore.exceptions.ClientError(error_response=error_response,
                operation_name="op_name")

        mock_function = Mock(side_effect=client_error)
        client_mock = type("ClientMock", (object,), {"register_workflow_type":mock_function})
        mocker.patch('floto.api.Swf.client', new_callable=PropertyMock, return_value=client_mock())

        swf = floto.api.Swf()
        args = {'domain': 'test_domain',
                'name': 'my_workflow_type',
                'version': 'v1'}
        workflow_type = floto.api.WorkflowType(**args)
        properties = workflow_type.swf_attributes
        swf.register_type(workflow_type)
        swf.client.register_workflow_type.assert_called_once_with(**properties)