我需要在我的單元中模擬RabbitMQ測試

Question

我在我的項目中使用了RabbitMQ。

我在我的消費者rabbitMQ的客戶端部分的代碼和連接需要一個tls1.1來連接真正的MQ。

我想在我的JUnit測試中測試此代碼並嘲笑向我的客戶發送消息。

我在谷歌看到幾個不同的工具的例子如何駱駝兔或activeMQ，但這個工具與amqp 1.0和rabbitMQ工程只適用於amqp 0.9。

有人有這個問題嗎？

謝謝！

UPDATE

這是測試從代碼隊列中接收json的代碼。

package com.foo.foo.queue; 

import java.io.File; 
import java.io.FileInputStream; 
import java.io.IOException; 
import java.net.URL; 
import java.security.*; 
import java.security.cert.CertificateException; 
import javax.net.ssl.*; 

import org.apache.commons.lang3.StringUtils; 
import org.apache.log4j.LogManager; 
import org.apache.log4j.Logger; 
import org.json.JSONObject; 

import com.foo.foo.Constants.Constants; 
import com.foo.foo.core.ConfigurationContainer; 
import com.foo.foo.policyfinders.PolicyFinder; 
import com.rabbitmq.client.Channel; 
import com.rabbitmq.client.Connection; 
import com.rabbitmq.client.ConnectionFactory; 
import com.rabbitmq.client.QueueingConsumer; 

public class BrokerThreadHLConsumer extends Thread { 

private static BrokerThreadHLConsumer instance; 

private static final Logger log = LogManager.getLogger(BrokerThreadHLConsumer.class); 

private Channel channel; 
private String queueName; 
private PolicyFinder PolicyFinder; 
private Connection connection; 
private QueueingConsumer consumer; 

private boolean loop; 

private BrokerThreadHLConsumer() throws IOException { 
    ConnectionFactory factory = new ConnectionFactory(); 
    char[] keyPassphrase = "clientrabbit".toCharArray(); 
    KeyStore keyStoreCacerts; 
    ConfigurationContainer configurationContainer = ConfigurationContainer.getInstance(); 
    String exchangeName = configurationContainer.getProperty(Constants.EXCHANGE_NAME); 
    String rabbitHost = configurationContainer.getProperty(Constants.RABBITMQ_SERVER_HOST_VALUE); 
    try { 
     /* Public key cacerts to connect to message queue*/ 
     keyStoreCacerts = KeyStore.getInstance("PKCS12"); 
     URL resourcePublicKey = this.getClass().getClassLoader().getResource("certs/client.keycert.p12"); 
     File filePublicKey = new File(resourcePublicKey.toURI()); 
     keyStoreCacerts.load(new FileInputStream(filePublicKey), keyPassphrase); 
     KeyManagerFactory keyManager; 

     keyManager = KeyManagerFactory.getInstance("SunX509"); 
     keyManager.init(keyStoreCacerts, keyPassphrase); 

     char[] trustPassphrase = "changeit".toCharArray(); 
     KeyStore tks; 

     tks = KeyStore.getInstance("JCEKS"); 

     URL resourceCacerts = this.getClass().getClassLoader().getResource("certs/cacerts"); 
     File fileCacerts = new File(resourceCacerts.toURI()); 

     tks.load(new FileInputStream(fileCacerts), trustPassphrase); 

     TrustManagerFactory tmf; 
     tmf = TrustManagerFactory.getInstance("SunX509"); 
     tmf.init(tks); 

     SSLContext c = SSLContext.getInstance("TLSv1.1"); 
     c.init(keyManager.getKeyManagers(), tmf.getTrustManagers(), null); 

     factory.setUri(rabbitHost); 
     factory.useSslProtocol(c); 
     connection = factory.newConnection(); 
     channel = connection.createChannel(); 
     channel.exchangeDeclare(exchangeName, "fanout"); 
     queueName = channel.queueDeclare().getQueue(); 
     channel.queueBind(queueName, exchangeName, ""); 

    } catch (NoSuchAlgorithmException e) { 
     e.printStackTrace(); 
    } catch (CertificateException e) { 
     e.printStackTrace(); 
    } catch (KeyStoreException e) { 
     e.printStackTrace(); 
    } catch (UnrecoverableKeyException e) { 
     e.printStackTrace(); 
    } catch (KeyManagementException e1) { 
     e1.printStackTrace(); 
    } catch (Exception e) { 
     log.error("Couldn't instantiate a channel with the broker installed in " + rabbitHost); 
     log.error(e.getStackTrace()); 
     e.printStackTrace(); 
    } 
} 

public static BrokerThreadHLConsumer getInstance() throws CertificateException, UnrecoverableKeyException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException, IOException { 
    if (instance == null) 
     instance = new BrokerThreadHLConsumer(); 
    return instance; 
} 

public void run() { 
    if (PolicyFinder != null) { 
     try { 
      consumer = new QueueingConsumer(channel); 
      channel.basicConsume(queueName, true, consumer); 
      log.info("Consumer broker started and waiting for messages"); 
      loop = true; 
      while (loop) { 
       try { 
        QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 
        String message = new String(delivery.getBody()); 
        JSONObject obj = new JSONObject(message); 
        log.info("Message received from broker " + obj); 
        if (StringUtils.isNotEmpty(message) && !PolicyFinder.managePolicySet(obj)) { 
         log.error("PolicySet error: error upgrading the policySet"); 
        } 
       } catch (Exception e) { 
        log.error("Receiving message error"); 
        log.error(e); 
       } 
      } 
     } catch (IOException e) { 
      log.error("Consumer couldn't start"); 
      log.error(e.getStackTrace()); 
     } 
    } else { 
     log.error("Consumer couldn't start cause of PolicyFinder is null"); 
    } 
} 

public void close() { 
    loop = false; 
    try { 
     consumer.getChannel().basicCancel(consumer.getConsumerTag()); 
    } catch (IOException e) { 
     e.printStackTrace(); 
    } 
    try { 
     channel.close(); 
    } catch (IOException e) { 
     e.printStackTrace(); 
    } 
    try { 
     connection.close(); 
    } catch (IOException e) { 
     e.printStackTrace(); 
    } 
} 

public void setLuxPolicyFinder(PolicyFinder PolicyFinder) { 
    this.PolicyFinder = PolicyFinder; 
} 
} 

Answer 1

所以這裏是我是如何做到的，有些東西可能在這裏和那裏隱藏必要的類實現細節，但你會得到一個提示！ :)

• 擔任單元測試：
  • RMQ工作正常，併發送數據到它會在隊列推
  • 唯一要測試的是，如果產生的數據正確與否
  • 以及是否正在呼叫RMQs send()！

---

public class SomeClassTest { 
     private Config config; 
     private RmqConfig rmqConfig; 
     private static final ObjectMapper mapper = new ObjectMapper(); 
     private JasperServerClient jasperServerClient; 
    // @Mock 
     @InjectMocks 
     private RabbitMQProducer rabbitMQProducer; 
     private Connection phoenixConnection; 
     private String targetNotificationMessage; 
     SomeClass someClassObject; 

     @Before 
     public void setUp() { 

      // Mock basic stuffs 
      config = mock(Config.class); 
      Connection = mock(Connection.class); 
      rabbitMQProducer = mock(RabbitMQProducer.class); // Imp 


      jasperServerClient = mock(JasperServerClient.class); 

      rmqConfig = RmqConfig.builder() 
        .host("localhost") 
        .port(5672) 
        .userName("guest") 
        .password("guest") 
        .queueName("somequeue_name") 
        .prefetch(1) 
        .build(); 
      final String randomMessage = "This is a waste message"; 
      Message mockMsg = Message.forSending(randomMessage.getBytes(), null, rmqConfig.getQueueName(), rmqConfig.getQueueName(), "text/plain", "UTF-8", true); // prepare a mock message 


      // Prepare service configs 
      ConnectionConfig connectionConfig = RmqConfigUtil.getConfig(rmqConfig); 
      ProducerConfig producerConfig = new ProducerConfigBuilder() 
        .exchange(rmqConfig.getQueueName()) 
        .contentType("text/pain") 
        .contentEncoding("UTF-8") 
        .connection(connectionConfig).build(); 
      rabbitMQProducer.open(croducerConfig.asMap()); 

      // build the major stuff where the code resides 
      someClassObject = SomeClass.builder() 
        .phoenixConnection(phoenixConnection) 
        .userExchangeName(rmqConfig.getQueueName()) 
        .userRabbitMQProducer(rabbitMQProducer) 
        .ftpConfig(config.getFtpConfig()) 
        .jasperServerClient(jasperServerClient) 
        .objectMapper(new ObjectMapper()) 
        .build(); 

      MockitoAnnotations.initMocks(this); 
     } 


     @Test 
     public void testNotificationPub() throws Exception { 

      // Prepare expected Values 
      targetNotificationMessage = <<some message>> 

      // Reflection - my target functions were private 
      Class cls = Class.forName("com.some.path.to.class"); 
      Object[] objForGetMessage = {<<stuffs>>, <<stuffs>>}; 

      Method getNotificationMessage = cls.getDeclaredMethod("private_fn_1", <<some class>>.class, <<some class>>.class); 
      Method pubNotification = cls.getDeclaredMethod("private_fn_2", <<some class>>.class, RabbitMQProducer.class, String.class); 

      getNotificationMessage.setAccessible(true); 
      pubNotification.setAccessible(true); 

      // Test Case #1 
      final <<some class>> notificationMessage = (<<some class>>)getNotificationMessage.invoke(someClassObject, objForGetMessage); 
      assertEquals(notificationMessage.getMessage(), targetNotificationMessage); 

      // Test Case #2 - this does RMQ call 
      Object[] objPubMessage = {notificationMessage, rabbitMQProducer, rmqConfig.getQueueName()}; 
      final Object publishNotification = pubNotification.invoke(someClassObject, objPubMessage); 
      assertEquals(publishNotificationResp, publishNotification); //viola 


      //Important, since RabbitMQProducer is mocked, we need to checkup if function call is made to "send" function which send data to RMQ 
      verify(rabbitMQProducer,times(1)).send(any()); 

     } 


     @Test 
     public void testMockCreation(){ 
      assertNotNull(rmqConfig); 
      assertNotNull(config); 
     } 

Answer 2

我知道，這是一個老問題，仍沒有答案爲止。同樣的問題對我有很大的幫助，以下是博文：https://tamasgyorfi.net/2016/04/21/writing-integration-tests-for-rabbitmq-based-components/。 它使用Apache QPID（不是OP中建議的ActiveMQ），它支持AMQP 0.9.1。