Wie stompst send, und verbinden Methoden

Question

Ich schreibe einen Protokollhandler für ActiveMQ. Ich habe eine Klasse, Messenger, um eine Nachricht an ActiveMQ zu veröffentlichen. Außerdem habe ich Klasse Handler hinzugefügt, um dies zu handhaben und get_logger, um diesen Logger zu bekommen.

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 

Ich habe ein UT dafür geschrieben.

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) 

Aber immer noch geht es um die ursprüngliche send-Methode und versucht, eine Verbindung mit dem Server herzustellen.

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 

Wie kann ich das verspotten?

Answer 1

Das Grundprinzip besteht darin, dass Sie ein Patch suchen, wo ein Objekt nachgeschlagen wird, das nicht notwendigerweise derselbe Ort ist, an dem es definiert ist.

Sie fügen den Patch an der falschen Stelle hinzu. Das Klassenmodul Messenger importiert Stomp von stompest.sync, bevor die Tests ausgeführt werden. Daher hat Ihr Patch keine Auswirkungen. Sie sollten den Verweis auf Stomp in Ihrem Modul patchen.

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 hilft, wenn Sie auf einer weiteren Klärung bedürfen, warum dies geschieht.