Add functional tests for reconnecting with message bus (#144)

* Add functional tests for reconnecting with message bus
Refactored RmqHandler

Author: e-pettersson-ericsson

src/functionaltests/java/com/ericsson/ei/rabbitmq/RabbitMQTestConnectionRunner.java

@@ -0,0 +1,15 @@
+package com.ericsson.ei.rabbitmq;
+
+import com.ericsson.ei.utils.BaseRunner;
+import org.junit.runner.RunWith;
+
+import cucumber.api.CucumberOptions;
+import cucumber.api.junit.Cucumber;
+
+@RunWith(Cucumber.class)
+@CucumberOptions(features = "src/functionaltests/resources/features/rabbitMQTestConnection.feature",
+        glue = {"com.ericsson.ei.rabbitmq"}, plugin = {
+        "html:target/cucumber-reports/RabbitMQTestConnectionRunner"})
+public class RabbitMQTestConnectionRunner extends BaseRunner {
+
+}

src/functionaltests/java/com/ericsson/ei/rabbitmq/RabbitMQTestConnectionSteps.java

@@ -0,0 +1,79 @@
+package com.ericsson.ei.rabbitmq;
+
+import com.ericsson.ei.utils.AMQPBrokerManager;
+import com.ericsson.ei.utils.FunctionalTestBase;
+import com.ericsson.ei.utils.TestContextInitializer;
+
+import cucumber.api.java.en.Given;
+import cucumber.api.java.en.Then;
+import cucumber.api.java.en.When;
+import org.junit.Ignore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+@Ignore
+public class RabbitMQTestConnectionSteps extends FunctionalTestBase {
+
+    @Value("${rabbitmq.port}")
+    private String rabbitMQPort;
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQTestConnectionSteps.class);
+    private static final String EIFFEL_EVENTS = "src/functionaltests/resources/eiffel_events_for_thread_testing.json";
+
+    private AMQPBrokerManager amqpBroker;
+
+    @Given("^We are connected to message bus$")
+    public void connect_to_message_bus() {
+        amqpBroker = TestContextInitializer.getBroker(Integer.parseInt(rabbitMQPort));
+        assertEquals("Expected message bus to be up",
+                true, amqpBroker.isRunning);
+    }
+
+    @When("^Message bus goes down$")
+    public void message_bus_goes_down() {
+        LOGGER.debug("Shutting down message bus");
+        amqpBroker.stopBroker();
+        assertEquals("Expected message bus to be down",
+                false, amqpBroker.isRunning);
+    }
+
+    @When("^Message bus is restarted$")
+    public void message_bus_is_restarted() throws Exception {
+        amqpBroker.startBroker();
+    }
+
+    @Then("^I can send events which are put in the waitlist$")
+    public void can_send_events_which_are_put_in_the_waitlist() throws Exception {
+        LOGGER.debug("Sending eiffel events");
+        int waitListSize = 0;
+        List<String> eventNames = getEventNamesToSend();
+        long maxTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30);
+
+        while (waitListSize != 4 && maxTime > System.currentTimeMillis()) {
+            eventManager.sendEiffelEvents(EIFFEL_EVENTS, eventNames);
+            TimeUnit.SECONDS.sleep(2);
+            waitListSize = dbManager.waitListSize();
+        }
+        assertEquals(4, waitListSize);
+    }
+
+    /**
+     * This method collects all the event names of events we will send to the message bus.
+     */
+    protected List<String> getEventNamesToSend() {
+        List<String> eventNames = new ArrayList<>();
+        eventNames.add("event_EiffelConfidenceLevelModifiedEvent_3_2");
+        eventNames.add("event_EiffelArtifactPublishedEvent_3");
+        eventNames.add("event_EiffelTestCaseTriggeredEvent_3");
+        eventNames.add("event_EiffelTestCaseStartedEvent_3");
+        return eventNames;
+    }
+
+}

src/functionaltests/java/com/ericsson/ei/utils/AMQPBrokerManager.java

@@ -6,28 +6,31 @@
 import org.apache.qpid.server.BrokerOptions;
 
 public class AMQPBrokerManager {
+    public boolean isRunning = false;
 
     private final Broker broker = new Broker();
     private String path;
-    private int port;
+    private String port;
 
-    public AMQPBrokerManager(String path, int port) {
+    public AMQPBrokerManager(String path, String port) {
         super();
         this.path = path;
         this.port = port;
     }
 
     public void startBroker() throws Exception {
         final BrokerOptions brokerOptions = new BrokerOptions();
-        brokerOptions.setConfigProperty("qpid.amqp_port", "" + port);
+        brokerOptions.setConfigProperty("qpid.amqp_port", port);
         brokerOptions.setConfigProperty("qpid.pass_file", "src/functionaltests/resources/configs/password.properties");
         brokerOptions.setConfigProperty("qpid.work_dir", Files.createTempDir().getAbsolutePath());
         brokerOptions.setInitialConfigurationLocation(path);
 
         broker.startup(brokerOptions);
+        isRunning = true;
     }
 
     public void stopBroker() {
         broker.shutdown();
+        isRunning = false;
     }
 }

src/functionaltests/java/com/ericsson/ei/utils/BaseRunner.java

@@ -0,0 +1,12 @@
+package com.ericsson.ei.utils;
+
+import org.junit.AfterClass;
+
+public class BaseRunner {
+
+    @AfterClass()
+    public static void shutdownAmqpBroker() {
+        String rabbitMQPort = System.getProperty("rabbitmq.port");
+        TestContextInitializer.removeBroker(rabbitMQPort);
+    }
+}

src/functionaltests/java/com/ericsson/ei/utils/TestConfigs.java

@@ -1,62 +1,55 @@
 package com.ericsson.ei.utils;
 
 import com.mongodb.MongoClient;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.apache.tomcat.util.codec.binary.Base64;
 import org.apache.tomcat.util.codec.binary.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.amqp.core.BindingBuilder;
-import org.springframework.amqp.core.Queue;
-import org.springframework.amqp.core.TopicExchange;
-import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.util.SocketUtils;
 
 import de.flapdoodle.embed.mongo.distribution.Version;
 import de.flapdoodle.embed.mongo.tests.MongodForTestsFactory;
 
 public class TestConfigs {
 
-    private static AMQPBrokerManager amqpBroker;
-    private static ConnectionFactory cf;
-    private Connection conn;
-    private MongodForTestsFactory testsFactory;
+    private final static Logger LOGGER = LoggerFactory.getLogger(TestConfigs.class);
+
     private MongoClient mongoClient = null;
 
-    final static Logger LOGGER = (Logger) LoggerFactory.getLogger(TestConfigs.class);
+    protected static Map<Integer, AMQPBrokerManager> amqpBrokerMap = new HashMap<>();
 
-    public void amqpBroker() throws Exception {
+    AMQPBrokerManager createAmqpBroker() throws Exception {
+        // Generates a random port for amqpBroker and starts up a new broker
         int port = SocketUtils.findAvailableTcpPort();
-        System.setProperty("rabbitmq.port", "" + port);
+
+        System.setProperty("rabbitmq.port", Integer.toString(port));
         System.setProperty("rabbitmq.user", "guest");
         System.setProperty("rabbitmq.password", "guest");
         System.setProperty("waitlist.initialDelayResend", "500");
         System.setProperty("waitlist.fixedRateResend", "3000");
 
         String config = "src/functionaltests/resources/configs/qpidConfig.json";
         File qpidConfig = new File(config);
-        amqpBroker = new AMQPBrokerManager(qpidConfig.getAbsolutePath(), port);
-        amqpBroker.startBroker();
-        cf = new ConnectionFactory();
-        cf.setUsername("guest");
-        cf.setPassword("guest");
-
-        cf.setPort(port);
-        cf.setHandshakeTimeout(600000);
-        cf.setConnectionTimeout(600000);
-        conn = cf.newConnection();
+        AMQPBrokerManager amqpBroker = new AMQPBrokerManager(qpidConfig.getAbsolutePath(), Integer.toString(port));
+
         LOGGER.debug("Started embedded message bus for tests on port: " + port);
+        amqpBroker.startBroker();
+
+        // add new amqp broker to pool
+        amqpBrokerMap.put(port, amqpBroker);
+
+        return amqpBroker;
     }
 
-    public void mongoClient() throws IOException {
+    void startUpMongoClient() throws IOException {
         try {
-            testsFactory = MongodForTestsFactory.with(Version.V3_4_1);
+            MongodForTestsFactory testsFactory = MongodForTestsFactory.with(Version.V3_4_1);
             mongoClient = testsFactory.newMongo();
             String port = "" + mongoClient.getAddress().getPort();
             System.setProperty("spring.data.mongodb.port", port);
@@ -66,7 +59,7 @@ public void mongoClient() throws IOException {
         }
     }
 
-    public void setAuthorization() {
+    void setAuthorization() {
         String password = StringUtils.newStringUtf8(Base64.encodeBase64("password".getBytes()));
         System.setProperty("ldap.enabled", "true");
         System.setProperty("ldap.url", "ldap://ldap.forumsys.com:389/dc=example,dc=com");
@@ -76,14 +69,13 @@ public void setAuthorization() {
         System.setProperty("ldap.user.filter", "uid={0}");
     }
 
-    public void createExchange(final String exchangeName, final String queueName) {
-        final CachingConnectionFactory ccf = new CachingConnectionFactory(cf);
-        RabbitAdmin admin = new RabbitAdmin(ccf);
-        Queue queue = new Queue(queueName, false);
-        admin.declareQueue(queue);
-        final TopicExchange exchange = new TopicExchange(exchangeName);
-        admin.declareExchange(exchange);
-        admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("#"));
-        ccf.destroy();
+    public static AMQPBrokerManager getBroker(int port) {
+        return amqpBrokerMap.get(port);
+    }
+
+    public static void removeBroker(String port) {
+        AMQPBrokerManager broker = amqpBrokerMap.get(Integer.parseInt(port));
+        broker.stopBroker();
+        amqpBrokerMap.remove(port);
     }
 }

src/functionaltests/java/com/ericsson/ei/utils/TestContextInitializer.java

@@ -1,23 +1,25 @@
 package com.ericsson.ei.utils;
 
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationContextInitializer;
 import org.springframework.context.ConfigurableApplicationContext;
 
+
 public class TestContextInitializer extends TestConfigs
         implements ApplicationContextInitializer<ConfigurableApplicationContext> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(TestContextInitializer.class);
 
     @Override
     public void initialize(ConfigurableApplicationContext ac) {
-        try {
-            amqpBroker();
-            mongoClient();
 
+        try {
+            createAmqpBroker();
+            startUpMongoClient();
         } catch (Exception e) {
-            LOGGER.error(e.getMessage(), e);
+            LOGGER.error("Failed to startup Mongo client or AMQP broker for test");
         }
     }
 }

src/functionaltests/resources/features/rabbitMQTestConnection.feature

@@ -0,0 +1,8 @@
+
+Feature: RabbitMQConnection
+
+  Scenario: Testing to automatically reconnect to Message Bus
+    Given We are connected to message bus
+    When Message bus goes down
+    And Message bus is restarted
+    Then I can send events which are put in the waitlist