From 59bde8ad442d77933b32109ca89c32441935ffa8 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 30 Jan 2014 17:02:08 +0200 Subject: [PATCH] INT-3282: Add `BrokerRunning` Rule JIRA: https://jira.springsource.org/browse/INT-3282 Remove `@Ignore` from integration tests and some polishing --- .../amqp/channel/ChannelTests.java | 7 +- ...utboundGatewayIntegrationTests-context.xml | 15 +- .../OutboundGatewayIntegrationTests.java | 36 +++-- .../integration/amqp/rule/BrokerRunning.java | 133 ++++++++++++++++++ 4 files changed, 171 insertions(+), 20 deletions(-) create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java index 10025c1c2b..23415437d4 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java @@ -21,13 +21,14 @@ import java.util.Collection; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; -import org.junit.Ignore; +import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.integration.amqp.rule.BrokerRunning; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; @@ -45,6 +46,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @RunWith(SpringJUnit4ClassRunner.class) public class ChannelTests { + @ClassRule + public static final BrokerRunning brokerIsRunning = BrokerRunning.isRunning(); + @Autowired private PublishSubscribeAmqpChannel channel; @@ -55,7 +59,6 @@ public class ChannelTests { private ConfigurableApplicationContext context; @Test - @Ignore public void pubSubLostConnectionTest() throws Exception { final CyclicBarrier latch = new CyclicBarrier(2); channel.subscribe(new MessageHandler() { diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml index e99a23eb52..26c77fe2d6 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml @@ -16,24 +16,21 @@ - + - - - - - - + + + + + amqp-template="amqpTemplate"/> (payload)); + Message receive = this.fromRabbit.receive(1000); + assertNotNull(receive); + assertEquals(payload.toUpperCase(), receive.getPayload()); } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java new file mode 100644 index 0000000000..410f911960 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java @@ -0,0 +1,133 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.amqp.rule; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Assume; +import org.junit.rules.TestWatcher; +import org.junit.runner.Description; +import org.junit.runners.model.Statement; + +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.util.Assert; + +/** + *

+ * A rule that prevents integration tests from failing if the Rabbit broker application is not running or not + * accessible. If the Rabbit broker is not running in the background all the tests here will simply be skipped because + * of a violated assumption (showing as successful). + *

+ * The rule can be declared as static so that it only has to check once for all tests in the enclosing test case, but + * there isn't a lot of overhead in making it non-static. + * + * @author Dave Syer + * @author Artem Bilan + * + * @since 4.0 + */ +public class BrokerRunning extends TestWatcher { + + public static final int PORT = 5672; + + private static final Log logger = LogFactory.getLog(BrokerRunning.class); + + private static final Queue DEFAULT_QUEUE_NAME = new Queue(BrokerRunning.class.getName()); + + private final Queue[] queues; + + /** + * Ensure the broker is running and has an empty queue (which can be addressed via the default exchange). + * + * @return a new rule that assumes an existing running broker + */ + public static BrokerRunning isRunningWithEmptyQueues(Queue... queues) { + Assert.notNull(queues); + Assert.noNullElements(queues); + return new BrokerRunning(queues); + } + + /** + * Ensure the broker is running and has an empty queue (which can be addressed via the default exchange). + * + * @return a new rule that assumes an existing running broker + */ + public static BrokerRunning isRunningWithEmptyQueues(String... queues) { + Assert.notNull(queues); + Assert.noNullElements(queues); + return new BrokerRunning(queues); + } + + /** + * @return a new rule that assumes an existing running broker + */ + public static BrokerRunning isRunning() { + return new BrokerRunning(DEFAULT_QUEUE_NAME); + } + + + private BrokerRunning(Queue... queues) { + this.queues = queues; + } + + private BrokerRunning(String... queues) { + List queueList = new ArrayList(queues.length); + for (String queue : queues) { + queueList.add(new Queue(queue)); + } + this.queues = queueList.toArray(new Queue[queues.length]); + } + + @Override + public Statement apply(Statement base, Description description) { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + + try { + + connectionFactory.setPort(PORT); + + RabbitAdmin admin = new RabbitAdmin(connectionFactory); + + for (Queue queue : queues) { + String queueName = queue.getName(); + logger.info("Deleting queue: " + queueName); + // Delete completely - gets rid of consumers and bindings as well + admin.deleteQueue(queueName); + + if (!DEFAULT_QUEUE_NAME.getName().equals(queueName)) { + admin.declareQueue(queue); + } + } + + + } + catch (final Exception e) { + logger.warn("Not executing tests because basic connectivity test failed", e); + Assume.assumeNoException(e); + } + finally { + connectionFactory.destroy(); + } + + return super.apply(base, description); + } + +}