@RabbitListener method testing in SpringBoot app

Question

Code:

RabbitMQListener:

@Component
public class ServerThroughRabbitMQ implements ServerThroughAMQPBroker {
    private static final AtomicLong ID_COUNTER=new AtomicLong();
    private final long instanceId=ID_COUNTER.incrementAndGet();


    @Autowired
    public ServerThroughRabbitMQ( UserService userService,LoginService loginService....){
....
    }

    @Override
    @RabbitListener(queues = "#{registerQueue.name}")
    public String registerUserAndLogin(String json) {
       .....
    }

ServerConfig:

@Configuration
public class ServerConfig {
    @Value("${amqp.broker.exchange-name}")
    private String exchangeName;
    @Value("${amqp.broker.host}")
    private String ampqBrokerHost;
    @Value("${amqp.broker.quidco.queue.postfix}")
    private String quidcoQueuePostfix;
    @Value("${amqp.broker.quidco.queue.durability:true}")
    private boolean quidcoQueueDurability;
    @Value("${amqp.broker.quidco.queue.autodelete:false}")
    private boolean quidcoQueueAutodelete;

    private String registerAndLoginQuequName;


    @PostConstruct
    public void init() {
        registerAndLoginQuequName = REGISTER_AND_LOGIN_ROUTING_KEY + quidcoQueuePostfix;
    public String getRegisterAndLoginQueueName() {
        return registerAndLoginQuequName;
    }

    public String getLoginAndCheckBonusQueueName() {
        return loginAndCheckBonusQuequName;
    }



    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(ampqBrokerHost);
        return connectionFactory;
    }

    @Bean
    public AmqpAdmin amqpAdmin() {
        return new RabbitAdmin(connectionFactory());
    }

    @Bean
    public TopicExchange topic() {
        return new TopicExchange(exchangeName);
    }

    @Bean(name = "registerQueue")
    public Queue registerQueue() {
        return new Queue(registerAndLoginQuequName, quidcoQueueDurability, false, quidcoQueueAutodelete);
    }


    @Bean
    public Binding bindingRegisterAndLogin() {
        return BindingBuilder.bind(registerQueue()).to(topic()).with(REGISTER_AND_LOGIN_ROUTING_KEY);
    }

   }

TestConfig:

@EnableRabbit
@TestPropertySource("classpath:test.properties")
public class ServerThroughAMQPBrokerRabbitMQIntegrationTestConfig {
    private final ExecutorService=Executors.newCachedThreadPool();
    private LoginService loginServiceMock=mock(LoginService.class);
    private UserService userServiceMock =mock(UserService.class);

    @Bean
    public ExecutorService executor() {
        return executorService;
    }

    @Bean
    public LoginService getLoginServiceMock() {
        return loginServiceMock;
    }

    @Bean
    public UserService getUserService() {
        return userServiceMock;
    }

    @Bean
    @Autowired
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMaxConcurrentConsumers(5);
        return factory;
    }

    @Bean
    @Autowired
    public RabbitTemplate getRabbitTemplate(ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

    @Bean
    public ServerThroughRabbitMQ getServerThroughRabbitMQ() {
        return new ServerThroughRabbitMQ(userServiceMock, loginServiceMock,...);
    }

}

Integration tests:

@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes ={ServerConfig.class,ServerThroughAMQPBrokerRabbitMQIntegrationTestConfig.class})
@Category({IntegrationTest.class})
@TestPropertySource("classpath:test.properties")
public class ServerThroughAMQPBrokerRabbitMQIntegrationTest {
    final private ObjectMapper jackson = new ObjectMapper();
    @Autowired
    private ExecutorService executor;

    @Autowired
    private ServerThroughRabbitMQ serverThroughRabbitMQ;

    @Autowired
    private RabbitTemplate template;

    @Autowired
    private TopicExchange exchange;

    @Autowired
    UserService userService;

    @Autowired
    LoginService loginService;

    @Autowired
    private AmqpAdmin amqpAdmin;

    @Autowired
    private ServerConfig serverConfig;

    final String username = "username";
    final String email = "[email protected]";
    final Integer tcVersion=1;
    final int quidcoUserId = 1;
    final String jwt = ProcessLauncherForJwtPhpBuilderUnitWithCxtTest.EXPECTED_JWT;


    @Before
    public void cleanAfterOthersForMyself() {
        cleanTestQueues();
    }

    @After
    public void cleanAfterMyselfForOthers() {
        cleanTestQueues();
    }

    private void cleanTestQueues() {
        amqpAdmin.purgeQueue(serverConfig.getRegisterAndLoginQueueName(), false);
    }

    @Test
    @Category({SlowTest.class,IntegrationTest.class})
    public void testRegistrationAndLogin() throws TimeoutException {
        final Waiter waiter = new Waiter();

        when(userService.register(anyString(), anyString(), anyString())).thenReturn(...);
        when(loginService....()).thenReturn(...);


        executor.submit(() -> {
            final RegistrationRequest request = new RegistrationRequest(username, email,tcVersion);
            final String response;
            try {
                //@todo: converter to convert RegistrationRequest inside next method to json
                response = (String) template.convertSendAndReceive(exchange.getName(), REGISTER_AND_LOGIN_ROUTING_KEY.toString(), jackson.writeValueAsString(request));
                waiter.assertThat(response, not(isEmptyString()));

                final RegistrationResponse registrationResponse = jackson.readValue(response, RegistrationResponse.class);
                waiter.assertThat(...);
                waiter.assertThat(...);

            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            waiter.resume();
        });

        waiter.await(5, TimeUnit.SECONDS);
    }

}

When I run that test separetly , everything works fine, but when I run it with other tests the mocked ServerThroughRabbitMQ isn't being used, so some spring caches force to use old rabbit listener.

I tried to debug it and I can see, that correct bean is being autowired to the test, but for some reason old listener is using(old bean field instanceId=1 new mocked bean instanceId=3) and test failing(Not sure how it's possible, so if in case of existing old bean I assume to get an autowire exception).

I tried to use @DirtiesContext BEFORE_CLASS, but faced anoter problem(see here)

Answer 1

RabbitMQ and Integration Testing can be hard, since Rabbit MQ keeps some kind of state: - messages from previous tests in queues - listeners from previous tests still listening on queues

There are several approaches:

• Purge all queues before you start the test (that might be what you mean by cleanTestQueues())
• Delete all queues (or use temporary queues) and recreate them before each test
• Using the Rabbit Admin Rest API killing listeners or connections of previous tests
• delete the vhost and recreating the infrasture for each test (which is the most brutal way)