test_pipeline.py 文件源码

def test_wrap_consumer(self, m_retry_type, m_logging_type):
        consumer = mock.sentinel.consumer
        retry_handler = mock.sentinel.retry_handler
        logging_handler = mock.sentinel.logging_handler
        m_retry_type.return_value = retry_handler
        m_logging_type.return_value = logging_handler
        thread_group = mock.sentinel.thread_group

        with mock.patch.object(h_dis.EventPipeline, '__init__'):
            pipeline = h_pipeline.ControllerPipeline(thread_group)
            ret = pipeline._wrap_consumer(consumer)

        self.assertEqual(logging_handler, ret)
        m_logging_type.assert_called_with(retry_handler)
        m_retry_type.assert_called_with(consumer, exceptions=mock.ANY)