test_consumer.py 文件源码

def test_set_process_name(self, config):
        consumer = KafkaConsumerBase(
            'my_very_extraordinarily_elongated_topic_name',
            config, ['1', '2', '3', '4', '5'])
        with mock.patch(
            'yelp_kafka.consumer.setproctitle',
        ) as mock_setproctitle:
            consumer.set_process_name()
            expected_name = \
                '{procname}-my_very_extraordinarily_elongated_topic_name' \
                '-{messages}'.format(
                    procname=getproctitle(),
                    messages=['1', '2', '3', '4', '5'],
                )
            mock_setproctitle.assert_called_with(expected_name)