“Python模拟：修补Python Pika的“basic_publish”函数”

Question

考虑测试中的代码：

import pika

class MQ_Client():
    connection = None
    channel = None
    exchange_name="my_exchange"

    def connect(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host="mq_server")

    def publish(self, message):

        properties = pika.BasicProperties(content_type='text/json', delivery_mode=1)

        if self.channel.basic_publish(exchange=self.exchange_name, routing_key='', properties=properties, body=message):
            log.info("Successfully published this message to exchange \"" + self.exchange_name + "\": " + message)
        else:
            log.error("Failed to publish message \"" + message + "\" to the exchange \"" + self.exchange_name + "\"!")
            raise Exception("Failed to publish message to the queue.")

我想要的是修补Pika的basic_publish方法来引发一个异常，并可能需要一些帮助来找出如何做到这一点。这是我最后一次尝试：

^{pr2}$

任何关于如何让这个工作的建议将不胜感激！在

Answer 1

我成功了。模拟实例方法与模拟类方法完全不同。在阅读了this excellent article之后，我终于明白了如何做到这一点，结果发现我所需要的修补basic_publish方法只需稍微修改一下我的测试，如下所示：

@mock.patch('mq_client.pika.BlockingConnection', spec=pika.BlockingConnection)
def test_that_mq_publish_problems_return_error(self, mocked_connection):
    with self.client as client:
       mocked_connection.return_value.channel.return_value.basic_publish.return_value = False