make test to wait until all events are processed

Change-Id: I7a72f635d7154e92f3dd36b478e69becb27f5716

Author: Vali (Vasile Baluta)

pom.xml

@@ -109,7 +109,16 @@
           <groupId>org.apache.qpid</groupId>
           <artifactId>qpid-broker</artifactId>
           <version>6.1.3</version>
+          <scope>test</scope>
       </dependency>
+
+      <dependency>
+      <groupId>de.flapdoodle.embed</groupId>
+      <artifactId>de.flapdoodle.embed.mongo</artifactId>
+      <version>2.0.0</version>
+      <scope>test</scope>
+   </dependency>
+
     </dependencies>
 
     <properties>

src/main/java/com/ericsson/ei/handlers/EventHandler.java

@@ -28,5 +28,11 @@ public void eventReceived(byte[] message) {
         String actualMessage = new String(message);
         log.info("Event received <" + actualMessage + ">");
         eventReceived(actualMessage);
+        if (System.getProperty("flow.test") == "true") {
+            String countStr = System.getProperty("eiffel.intelligence.processedEventsCount");
+            int count = Integer.parseInt(countStr);
+            count++;
+            System.setProperty("eiffel.intelligence.processedEventsCount", "" + count);
+        }
     }
 }

src/test/java/com/ericsson/ei/flowtests/FlowTest.java

@@ -1,7 +1,10 @@
 package com.ericsson.ei.flowtests;
 
+import static org.junit.Assert.assertEquals;
+
 import java.io.File;
 import java.io.IOException;
+import java.net.UnknownHostException;
 import java.util.ArrayList;
 
 import org.apache.commons.io.FileUtils;
@@ -14,11 +17,14 @@
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import com.ericsson.ei.handlers.EventHandler;
+import com.ericsson.ei.handlers.ObjectHandler;
 import com.ericsson.ei.rmq.consumer.RmqConsumer;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -29,18 +35,30 @@
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 
+import de.flapdoodle.embed.mongo.MongodExecutable;
+import de.flapdoodle.embed.mongo.MongodProcess;
+import de.flapdoodle.embed.mongo.MongodStarter;
+import de.flapdoodle.embed.mongo.config.IMongodConfig;
+import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
+import de.flapdoodle.embed.mongo.config.Net;
+import de.flapdoodle.embed.mongo.distribution.Version;
+import de.flapdoodle.embed.process.runtime.Network;
+
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringBootTest
 public class FlowTest {
 
-        public static File qpidConfig = null;
-        static AMQPBrokerManager amqpBrocker;
+    private static Logger log = LoggerFactory.getLogger(FlowTest.class);
+
+    public static File qpidConfig = null;
+    static AMQPBrokerManager amqpBrocker;
+    static MongodExecutable mongodExecutable = null;
 
-        @Autowired
-        RmqConsumer rmqConsumer;
+    @Autowired
+    RmqConsumer rmqConsumer;
 
-        @Autowired
-        EventHandler eventHandler;
+    @Autowired
+    ObjectHandler objectHandler;
 
     public static class AMQPBrokerManager {
         private String path;
@@ -71,13 +89,22 @@ public void stopBroker() {
     private static String jsonFileContent;
     private static JsonNode parsedJason;
     private static String jsonFilePath = "src/test/resources/test_events.json";
+    static private final String inputFilePath = "src/test/resources/AggregatedDocument.json";
 
     @BeforeClass
     public static void setup() throws Exception {
+        System.setProperty("flow.test", "true");
+        System.setProperty("eiffel.intelligence.processedEventsCount", "0");
+        setUpMessageBus();
+        setUpEmbeddedMongo();
+    }
+
+    public static void setUpMessageBus() throws Exception {
         System.setProperty("rabbitmq.port", "8672");
         System.setProperty("rabbitmq.user", "guest");
         System.setProperty("rabbitmq.password", "guest");
 
+
         String config = "src/test/resources/configs/qpidConfig.json";
         jsonFileContent = FileUtils.readFileToString(new File(jsonFilePath));
         ObjectMapper objectmapper = new ObjectMapper();
@@ -90,20 +117,44 @@ public static void setup() throws Exception {
         cf.setPassword("guest");
         cf.setPort(8672);
         conn = cf.newConnection();
+    }
+
+    public static void setUpEmbeddedMongo() throws Exception {
+        int port = 12349;
+        System.setProperty("mongodb.port", ""+port);
+
+        MongodStarter starter = MongodStarter.getDefaultInstance();
+
+        String bindIp = "localhost";
+
+        IMongodConfig mongodConfig = new MongodConfigBuilder()
+            .version(Version.Main.PRODUCTION)
+            .net(new Net(bindIp, port, Network.localhostIsIPv6()))
+            .build();
 
+
+        try {
+            mongodExecutable = starter.prepare(mongodConfig);
+            MongodProcess mongod = mongodExecutable.start();
+        } catch (Exception e) {
+            log.info(e.getMessage(),e);
+        }
     }
 
     @AfterClass
     public static void tearDown() throws Exception {
         amqpBrocker.stopBroker();
+
+        if (mongodExecutable != null)
+            mongodExecutable.stop();
+
         try {
             conn.close();
         } catch (Exception e) {
             //We try to close the connection but if
             //the connection is closed we just receive the
             //exception and go on
         }
-
     }
 
     @Test
@@ -113,25 +164,38 @@ public void test() {
             String queueName = rmqConsumer.getQueueName();
             String exchange = "ei-poc-4";
 
-            long count = channel.consumerCount(queueName);
             ArrayList<String> eventNames = getEventNamesToSend();
-
+            int eventsCount = eventNames.size();
             for(String eventName : eventNames) {
                 JsonNode eventJson = parsedJason.get(eventName);
                 String event = eventJson.toString();
                 channel.basicPublish(exchange, queueName,  null, event.getBytes());
             }
+
+            // wait for all events to be processed
+            int processedEvents = 0;
+            while (processedEvents < eventsCount) {
+                String countStr = System.getProperty("eiffel.intelligence.processedEventsCount");
+                processedEvents = Integer.parseInt(countStr);
+            }
+
+            String document = objectHandler.findObjectById("6acc3c87-75e0-4b6d-88f5-b1a5d4e62b43");
+            String expectedDocument = FileUtils.readFileToString(new File(inputFilePath));
+            ObjectMapper objectmapper = new ObjectMapper();
+            JsonNode expectedJson = objectmapper.readTree(expectedDocument);
+            JsonNode actualJson = objectmapper.readTree(document);
+            String breakString = "breakHere";
+            assertEquals(expectedJson, actualJson);
         } catch (Exception e) {
-            // TODO Auto-generated catch block
-            e.printStackTrace();
+            log.info(e.getMessage(),e);
         }
     }
 
     private ArrayList<String> getEventNamesToSend() {
          ArrayList<String> eventNames = new ArrayList<>();
          eventNames.add("event_EiffelArtifactCreatedEvent_3");
-//         eventNames.add("event_EiffelArtifactPublishedEvent_3");
-//         eventNames.add("event_EiffelConfidenceLevelModifiedEvent_3_2");
+         eventNames.add("event_EiffelArtifactPublishedEvent_3");
+         eventNames.add("event_EiffelConfidenceLevelModifiedEvent_3_2");
          eventNames.add("event_EiffelTestCaseStartedEvent_3");
          eventNames.add("event_EiffelTestCaseFinishedEvent_3");
 

src/test/resources/AggregatedDocument.json

@@ -0,0 +1 @@
+{ "_id" : "6acc3c87-75e0-4b6d-88f5-b1a5d4e62b43" , "aggregatedObject" : "{\"fileInformation\":[{\"extension\":\"jar\",\"classifier\":\"\"}],\"buildCommand\":null,\"confidenceLevels\":[{\"eventId\":\"f37d59a3-069e-4f4c-8cc5-a52e73501a75\",\"name\":\"readyForDelivery\",\"time\":1481875944272,\"value\":\"SUCCESS\"}],\"testCaseExecutions\":[{\"testCaseFinishEventId\":\"11109351-41e0-474a-bc1c-f6e81e58a1c9\",\"testCaseStartedTime\":1481875925916,\"testCaseStartedEventId\":\"cb9d64b0-a6e9-4419-8b5d-a650c27c59ca\",\"testCaseFinishedTime\":1481875935919,\"testCase\":{\"conclusion\":\"SUCCESSFUL\",\"verdict\":\"PASSED\",\"tracker\":\"My Other Test Management System\",\"id\":\"TC5\",\"uri\":\"https:\\/\\/other-tm.company.com\\/testCase\\/TC5\"}}],\"id\":\"6acc3c87-75e0-4b6d-88f5-b1a5d4e62b43\",\"time\":1481875891763,\"type\":\"EiffelArtifactCreatedEvent\",\"gav\":{\"groupId\":\"com.mycompany.myproduct\",\"artifactId\":\"sub-system\",\"version\":\"1.1.0\"},\"publications\":[{\"eventId\":\"33d05e6f-9bd9-4138-83b6-e20cc74680a3\",\"locations\":[{\"type\":\"PLAIN\",\"uri\":\"https:\\/\\/myrepository.com\\/mySubSystemArtifact\"}],\"time\":1481875921763}]}"}