Stratio#107: Add tests for SSL connection

jniebuhr · Aug 18, 2017 · 40c6b5c · 40c6b5c
1 parent 9bbaf8c
commit 40c6b5c
Showing 1 changed file with 17 additions and 3 deletions.
diff --git a/src/test/scala/org/apache/spark/streaming/rabbitmq/RabbitMQConsumerIT.scala b/src/test/scala/org/apache/spark/streaming/rabbitmq/RabbitMQConsumerIT.scala
@@ -27,16 +27,30 @@ class RabbitMQConsumerIT extends TemporalDataSuite {
 
   override val exchangeName = s"$configExchangeName-${this.getClass().getName()}-${UUID.randomUUID().toString}"
 
-  test("RabbitMQ Receiver should read all the records") {
+  test("RabbitMQ Receiver should read all the records without SSL") {
+    testReadRecords(5672, null)
+  }
+
+  test("RabbitMQ Receiver should read all the records with SSL") {
+    testReadRecords(5671, "tlsv1.2")
+  }
+
+  private def hostsWithPort(port: Int): String = hosts
+    .split(",")
+    .map(h => if(h.contains(":")) h.split(":")(0) else h)
+    .map(_ + ":" + port)
+    .mkString(",")
 
+  private def testReadRecords(port: Int, ssl: String): Unit = {
     val receiverStream = RabbitMQUtils.createStream(ssc, Map(
-      "hosts" -> hosts,
+      "hosts" -> hostsWithPort(port),
       "queueName" -> queueName,
       "exchangeName" -> exchangeName,
       "exchangeType" -> exchangeType,
       "vHost" -> vHost,
       "userName" -> userName,
-      "password" -> password
+      "password" -> password,
+      "sslProtocol" -> ssl
     ))
     val totalEvents = ssc.sparkContext.longAccumulator("My Accumulator")