Improve coverage

ging · May 8, 2019 · 27ef826 · 27ef826
1 parent 5b8c266
commit 27ef826
Show file tree

Hide file tree

Showing 5 changed files with 26 additions and 18 deletions.
diff --git a/src/main/scala/org/fiware/cosmos/orion/flink/connector/OrionHttpHandler.scala b/src/main/scala/org/fiware/cosmos/orion/flink/connector/OrionHttpHandler.scala
@@ -53,13 +53,16 @@ class OrionHttpHandler(
   override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef): Unit = {
     msg match {
       case req : FullHttpRequest =>
+
         if (req.method() != HttpMethod.POST) {
           throw new Exception("Only POST requests are allowed")
         }
         val ngsiEvent = parseMessage(req)
         if (sc != null && ngsiEvent != null) {
           logger.info(write(ngsiEvent))
           sc.collect(ngsiEvent)
+        } else {
+          exceptionCaught(ctx, new Exception("Request body is not an NgsiEvent"))
         }
         if (HttpUtil.is100ContinueExpected(req)) {
           ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE))
@@ -110,12 +113,12 @@ class OrionHttpHandler(
           //Convert attributes to Attribute objects
           .transform((k,v) => MapToAttributeConverter
           .unapply(v.asInstanceOf[Map[String,Any]]))
-        new Entity(entityId,entityType,attrs)
+        Entity(entityId, entityType, attrs)
       })
       // Generate timestamp
       val creationTime = System.currentTimeMillis
       // Generate NgsiEvent
-      val ngsiEvent = new NgsiEvent(creationTime, service, servicePath, entities, subscriptionId)
+      val ngsiEvent = NgsiEvent(creationTime, service, servicePath, entities, subscriptionId)
       ngsiEvent
     } catch {
       case e: Exception => null

diff --git a/src/main/scala/org/fiware/cosmos/orion/flink/connector/OrionSink.scala b/src/main/scala/org/fiware/cosmos/orion/flink/connector/OrionSink.scala
@@ -78,7 +78,7 @@ object OrionSink {
 
        try {
          val response = client.execute(httpEntity)
-          logger.info("POST to " + msg.url)
+          logger.info("Sent to " + msg.url)
        } catch {
          case e: Exception => {
            logger.error(e.toString)

diff --git a/src/test/scala/org.fiware.cosmos.orion.flink.connector.tests/FlinkJobTest.scala b/src/test/scala/org.fiware.cosmos.orion.flink.connector.tests/FlinkJobTest.scala
@@ -1,10 +1,11 @@
 package org.fiware.cosmos.orion.flink.connector.test
 
 import org.apache.flink.streaming.api.scala._
-
-import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment}
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
 import org.apache.flink.streaming.api.windowing.time.Time
-import org.fiware.cosmos.orion.flink.connector.OrionSource
+import org.fiware.cosmos.orion.flink.connector._
+import org.json4s.DefaultFormats
+import org.json4s.jackson.Serialization.write
 
 
 object Constants {
@@ -18,6 +19,7 @@ object Constants {
   * @author @sonsoleslp
   */
 object FlinkJobTest{
+  implicit val formats = DefaultFormats
 
   def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
@@ -29,15 +31,18 @@ object FlinkJobTest{
       .map(entity => {
         val temp = entity.attrs("temperature").value.asInstanceOf[Number].floatValue()
         val pres = entity.attrs("pressure").value.asInstanceOf[Number].floatValue()
-        new EntityNode( entity.id, temp, pres)
+        EntityNode(entity.id, temp, pres)
       })
       .keyBy("id")
       .timeWindow(Time.seconds(Constants.MaxWindow), Time.seconds(Constants.MinWindow))
 
-    processedDataStream.max("temperature").map(max=> {
-      simulatedNotification.maxTempVal = max.temperature})
-    processedDataStream .max("pressure").map(max=> {
-      simulatedNotification.maxPresVal = max.pressure})
+      processedDataStream.max("temperature").map(max=> {
+          simulatedNotification.maxTempVal = max.temperature})
+      val sinkStream = processedDataStream .max("pressure").map(max=> {
+        simulatedNotification.maxPresVal = max.pressure
+        OrionSinkObject(write(max),"http://localhost:3000",ContentType.JSON,HTTPMethod.POST)
+      })
+    OrionSink.addSink(sinkStream)
 
     env.execute("Socket Window NgsiEvent")
   }

diff --git a/src/test/scala/org.fiware.cosmos.orion.flink.connector.tests/OrionConnectorTest.scala b/src/test/scala/org.fiware.cosmos.orion.flink.connector.tests/OrionConnectorTest.scala
@@ -102,7 +102,7 @@ class OrionConnectorTest extends  BaseTest{
   }
 
   @Test def buildHttpPostSinkEntity : Unit = {
-    val os = new OrionSinkObject(Utils.Content, Utils.OrionAddress,  ContentType.Plain, HTTPMethod.POST)
+    val os = OrionSinkObject(Utils.Content, Utils.OrionAddress, ContentType.Plain, HTTPMethod.POST)
     val httpMsg = OrionSink.createHttpMsg(os)
     val content = scala.io.Source.fromInputStream(httpMsg.getEntity.getContent).mkString
 
@@ -112,7 +112,7 @@ class OrionConnectorTest extends  BaseTest{
   }
 
   @Test def buildHttpPutSinkEntity : Unit = {
-    val os = new OrionSinkObject(Utils.Content, Utils.OrionAddress,  ContentType.JSON, HTTPMethod.PUT)
+    val os = OrionSinkObject(Utils.Content, Utils.OrionAddress, ContentType.JSON, HTTPMethod.PUT)
     val httpMsg = OrionSink.createHttpMsg(os)
     val content = scala.io.Source.fromInputStream(httpMsg.getEntity.getContent).mkString
 
@@ -122,7 +122,7 @@ class OrionConnectorTest extends  BaseTest{
   }
 
   @Test def buildHttpPatchSinkEntity : Unit = {
-    val os = new OrionSinkObject(Utils.Content, Utils.OrionAddress,  ContentType.JSON, HTTPMethod.PATCH)
+    val os = OrionSinkObject(Utils.Content, Utils.OrionAddress, ContentType.JSON, HTTPMethod.PATCH)
     val httpMsg = OrionSink.createHttpMsg(os)
     val content = scala.io.Source.fromInputStream(httpMsg.getEntity.getContent).mkString
 
@@ -181,6 +181,5 @@ class OrionConnectorTest extends  BaseTest{
     Thread.sleep(Utils.SleepTimeShort)
     Assert.assertEquals(simulatedNotification.maxTempVal,originalValue,0)
 
-
   }
 }
diff --git a/src/test/scala/org.fiware.cosmos.orion.flink.connector.tests/SimulatedNotification.scala b/src/test/scala/org.fiware.cosmos.orion.flink.connector.tests/SimulatedNotification.scala
@@ -15,9 +15,10 @@ final val notificationVal =
        ],
        "subscriptionId": "57458eb60962ef754e7c0998"
      }""".stripMargin
-final val defaultTemperature = 20;
-final val defaultPressure = 40;
-def notification(temperature: Float = defaultTemperature, pressure: Float = defaultPressure) : String = {
+final val defaultTemperature = 20
+  final val defaultPressure = 40
+
+  def notification(temperature: Float = defaultTemperature, pressure: Float = defaultPressure) : String = {
   s"""{
         "data": [
            {