Muszę wyśmiać RabbitMQ w mojej jednostce Test

Question

Używam RabbitMQ w moim projekcie.

Mam w moim kliencie kod części klienta rabbitMQ, a połączenie wymaga Tls1.1 do połączenia z prawdziwym MQ.

Chcę przetestować ten kod w moim teście JUnit i wyśmiać dostarczanie wiadomości do mojego konsumenta.

Widzę w Google kilka przykładów z różnymi narzędziami jak królik wielbłąda lub activeMQ, ale to narzędzie działa z amqp 1.0 i rabbitMQ działa tylko w amqp 0.9.

Ktoś miał ten problem?

Dzięki!

UPDATE

Jest to kod badania, aby otrzymać json z kolejki.

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

Więc tutaj jest, jak to zrobiłem, niektóre smary mogą być tu i tam w trakcie ukrywania niezbędne szczegóły implementacji klasy, ale co można uzyskać podpowiedź! :)

• założenie dla testów jednostkowych:
  • RMQ działa poprawnie, a dane przesyła byłoby pchnął w kolejce
  • Jedyną rzeczą być badane jest, czy dane generowane jest poprawne, czy nie
  • i czy trwa rozmowa z RMQ send(), czy nie!

---

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

wiem, że jest to stary pytanie, jak jeszcze nie ma odpowiedzi do tej pory. To, co bardzo mi pomogło w tym samym pytaniu, to następujący wpis na blogu: https://tamasgyorfi.net/2016/04/21/writing-integration-tests-for-rabbitmq-based-components/. Używa Apache QPID (nie ActiveMQ zgodnie z sugestią w OP) i ma obsługę AMQP 0.9.1.