Как имитировать методы отправки и подключения stompest

Я пишу обработчик журнала для 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

Как я могу смеяться над этим?

1 ответ

  1. Основной принцип заключается в том, что вы исправляете, где объект ищется, что не обязательно то же самое место, где он определен.

    Вы добавляете патч не в то место. Модуль Messengerкласса импортируется Stompиз stompest.syncдо запуска тестов, и поэтому исправление не имеет эффекта. Вы должны исправить ссылку на 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"
    
            ...
    

    Это поможет, если вам нужно дальнейшее разъяснение о том, почему это происходит.