如何模拟 stompest 发送和连接方法答案

【问题标题】：How to mock stompest send, and connect methods如何模拟 stompest 发送和连接方法
【发布时间】：2016-05-24 16:37:39
【问题描述】：

我正在为ActiveMQ 编写日志处理程序。我有一个班级Messenger，可以向ActiveMQ 发布消息。另外，我添加了Handler 类来处理这个问题，并添加了get_logger 来获取这个记录器。

import json
import logging

from stompest.config import StompConfig
from stompest.sync import Stomp

class Messanger(object):
    def __init__(self, uri):
        self.cfg = StompConfig(uri)

    def publish(self, queue, data):
        data = json.dumps(data)
        client = Stomp(self.cfg)
        client.connect()

        try:
            client.send(queue, data)
        except Exception, exc:
            print "Error: ", exc
        client.disconnect()

class Handler(logging.Handler):
    def __init__(self, amq_uri, out_queue):
        logging.Handler.__init__(self)
        self.queue = queue
        self.uri = uri

    def emit(self, record):
        msg = self.format(record)
        messanger = Messanger(self.uri)
        messanger.send(self.queue, msg)



def get_logger(uri, queue):
    logger = logging.getLogger('testlogger')
    logger.addHandler(Handler(uri, queue))

    return logger

我为此写了一个 UT。

from unittest import TestCase
from mock import patch, Mock

from my_package import get_logger 

class TestHandler(TestCase):
    @patch('stompest.sync.client.Stomp')
    def test_activemq_handler(self, mock_stomp):
        URI = "tcp://localhost:61613"
        QUEUE = "/queue/logger_queue"

        mock_stomp.connect = Mock(return_value=1)
        mock_stomp.send = Mock(return_value=2)

        data = "This is test logging"

        LOG = get_logger(URI, QUEUE)
        LOG.error(data)

但它仍然会转到原始发送方法并尝试连接到服务器。

Traceback (most recent call last):
  File ".my_package/venv/lib/python2.7/site-packages/mock/mock.py", line 1305, in patched
    return func(*args, **keywargs)
  File "./my_package/tests/test_activemq_handler.py", line 23, in test_activemq_handler
    LOG.error(data)
  File "/System/Library/Python.framework/Versions/2.7/lib/python2.7/logging/__init__.py", line 1191, in error
    self._log(ERROR, msg, args, **kwargs)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/logging/__init__.py", line 1284, in _log
    self.handle(record)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/logging/__init__.py", line 1294, in handle
    self.callHandlers(record)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/logging/__init__.py", line 1334, in callHandlers
    hdlr.handle(record)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/logging/__init__.py", line 757, in handle
    self.emit(record)
  File "./my_package/__init__.py", line 28, in emit
    messanger.send(self.queue, msg)
  File "./my_package/clients/stomp_messanger.py", line 41, in send
    client.connect()
  File "./my_package/venv/lib/python2.7/site-packages/stompest/sync/client.py", line 85, in connect
    for (broker, connectDelay) in self._failover:
  File "./my_package/venv/lib/python2.7/site-packages/stompest/protocol/failover.py", line 48, in __iter__
    yield broker, self._delay()
  File "./my_package/venv/lib/python2.7/site-packages/stompest/protocol/failover.py", line 66, in _delay
    raise StompConnectTimeout('Reconnect timeout: %d attempts' % self._maxReconnectAttempts)
StompConnectTimeout: Reconnect timeout: 0 attempts

如何模拟这个？

【问题讨论】：

标签： python unit-testing mocking stomp

【解决方案1】：

基本原则是你在查找对象的地方打补丁，不一定和定义它的地方是同一个地方。

您将补丁添加到错误的位置。 Messenger 类模块在测试运行之前从stompest.sync 导入Stomp，这就是你的补丁没有效果的原因。您应该修补模块中对Stomp 的引用。

from unittest import TestCase
from mock import patch, Mock

from my_package import get_logger 

class TestHandler(TestCase):
    @patch('path.to.your.Messenger_class_module.Stomp') #path to your module
    def test_activemq_handler(self, mock_stomp):
        URI = "tcp://localhost:61613"
        QUEUE = "/queue/logger_queue"

        ...

如果您需要进一步说明为什么会发生这种情况，This 会有所帮助。

【讨论】：