producer_test.py 文件源码

def test_messages_not_duplicated(self, message, producer_instance):
        with capture_new_messages(
            message.topic
        ) as get_messages, producer_instance as producer:
            producer.publish(message)
            producer.flush()

        assert len(multiprocessing.active_children()) == 0
        assert len(get_messages()) == 1