[!437] - feature: Remove duplicated ways of running tests with local spark

# New features and improvements

Only one way to run spark local tests.

# Breaking changes

None.

# Migration

None.

# Bug fixes

None.

# How this feature was tested

Existing tests.

# Related issue

Closes #561

Author: agilelab-tmnd1991

build.sbt

@@ -1,5 +1,7 @@
 import Flavor.EMR613
 
+val dependCompileOnCompileAndTestOnTest = "compile->compile;test->test"
+
 lazy val flavor = {
   val f = Flavor.currentFlavor()
   System.err.println(Utils.printWithBorders(s"Building for flavor: ${f}", "*"))
@@ -111,7 +113,7 @@ lazy val plugin_hbase_spark = Project("wasp-plugin-hbase-spark", file("plugin-hb
 lazy val plugin_plain_hbase_writer_spark =
   Project("wasp-plugin-plain-hbase-writer-spark", file("plugin-plain-hbase-writer-spark"))
     .settings(settings.commonSettings: _*)
-    .dependsOn(consumers_spark)
+    .dependsOn(consumers_spark % dependCompileOnCompileAndTestOnTest)
     .settings(libraryDependencies ++= dependencies.pluginPlainHbaseWriterSparkDependencies)
 
 lazy val plugin_jdbc_spark = Project("wasp-plugin-jdbc-spark", file("plugin-jdbc-spark"))
@@ -130,7 +132,7 @@ lazy val plugin_kafka_spark = Project("wasp-plugin-kafka-spark", file("plugin-ka
     Test / unmanagedSourceDirectories += sourceDirectory.value / "test"
       / s"scala${if (flavor == EMR613) "-emr613" else "-legacy"}"
   )
-  .dependsOn(consumers_spark % "compile->compile;test->test")
+  .dependsOn(consumers_spark % dependCompileOnCompileAndTestOnTest)
   .settings(libraryDependencies ++= dependencies.pluginKafkaSparkDependencies)
 
 lazy val plugin_kafka_spark_old = Project("wasp-plugin-kafka-spark-old", file("plugin-kafka-spark-old"))
@@ -141,13 +143,13 @@ lazy val plugin_kafka_spark_old = Project("wasp-plugin-kafka-spark-old", file("p
     Test / unmanagedSourceDirectories += baseDirectory.value / ".." / "plugin-kafka-spark" / "src" / "test"
       / s"scala${if (flavor == EMR613) "-emr613" else "-legacy"}"
   )
-  .dependsOn(consumers_spark % "compile->compile;test->test")
+  .dependsOn(consumers_spark % dependCompileOnCompileAndTestOnTest)
   .settings(libraryDependencies ++= dependencies.pluginKafkaSparkOldDependencies)
 
 lazy val plugin_raw_spark = Project("wasp-plugin-raw-spark", file("plugin-raw-spark"))
   .settings(settings.commonSettings: _*)
   .settings(libraryDependencies ++= dependencies.sparkPluginBasicDependencies)
-  .dependsOn(consumers_spark % "compile->compile;test->test")
+  .dependsOn(consumers_spark % dependCompileOnCompileAndTestOnTest)
 
 lazy val plugin_solr_spark = Project("wasp-plugin-solr-spark", file("plugin-solr-spark"))
   .settings(settings.commonSettings: _*)
@@ -170,30 +172,26 @@ lazy val plugin_http_spark = Project("wasp-plugin-http-spark", file("plugin-http
       / s"scala${if (flavor == EMR613) "-emr613" else "-legacy"}"
   )
   .settings(settings.commonSettings: _*)
-  .dependsOn(consumers_spark % "compile->compile;test->test")
+  .dependsOn(consumers_spark % dependCompileOnCompileAndTestOnTest)
   .settings(libraryDependencies ++= dependencies.pluginHttpSparkDependencies)
 
 lazy val plugin_cdc_spark = Project("wasp-plugin-cdc-spark", file("plugin-cdc-spark"))
   .settings(settings.commonSettings: _*)
-  .dependsOn(consumers_spark % "compile->compile;test->test")
+  .dependsOn(consumers_spark % dependCompileOnCompileAndTestOnTest)
   .settings(libraryDependencies ++= dependencies.pluginCdcSparkDependencies)
 
 lazy val microservice_catalog = Project("wasp-microservice-catalog", file("microservice-catalog"))
   .settings(settings.commonSettings: _*)
   .settings(libraryDependencies ++= dependencies.microserviceCatalogDependencies)
-  .dependsOn(consumers_spark % "compile->compile;test->test")
+  .dependsOn(consumers_spark % dependCompileOnCompileAndTestOnTest)
 
 lazy val plugin_parallel_write_spark = Project("wasp-plugin-parallel-write-spark", file("plugin-parallel-write-spark"))
-  .settings(
-    Test / unmanagedSourceDirectories += sourceDirectory.value / "test"
-      / s"scala${if (flavor == EMR613) "-emr613" else "-legacy"}"
-  )
   .settings(Defaults.itSettings)
   .settings(settings.commonSettings: _*)
   .settings(settings.disableParallelTests)
   .settings(libraryDependencies ++= dependencies.pluginParallelWriteSparkDependencies)
-  .dependsOn(microservice_catalog % "compile->compile;test->test")
-  .dependsOn(aws_auth_temporary_credentials % "compile->compile;test->test")
+  .dependsOn(microservice_catalog % dependCompileOnCompileAndTestOnTest)
+  .dependsOn(aws_auth_temporary_credentials % dependCompileOnCompileAndTestOnTest)
 
 /* Yarn  */
 

consumers-spark/src/test/scala/it/agilelab/bigdata/wasp/consumers/spark/eventengine/InnerEventStrategySpec.scala

@@ -1,60 +1,71 @@
 package it.agilelab.bigdata.wasp.consumers.spark.eventengine
 
-import java.io.InputStreamReader
-import java.time.{Clock, Instant, ZoneId}
-
 import com.typesafe.config.ConfigFactory
+import it.agilelab.bigdata.wasp.consumers.spark.utils.SparkSuite
 import it.agilelab.bigdata.wasp.core.eventengine.Event
 import it.agilelab.bigdata.wasp.core.logging.Logging
 import org.scalatest.{BeforeAndAfter, Matchers, WordSpec}
 
+import java.io.InputStreamReader
+import java.time.{Clock, Instant, ZoneId}
 import scala.util.Random
 
+class InnerEventStrategySpec extends WordSpec with Matchers with BeforeAndAfter with SparkSuite with Logging {
 
-class InnerEventStrategySpec extends WordSpec
-  with Matchers with BeforeAndAfter with SparkSetup with Logging {
-
-  private val randomSeed = 0
-  private val rand = new Random(randomSeed)
+  private val randomSeed        = 0
+  private val rand              = new Random(randomSeed)
   private val fixedClock: Clock = Clock.fixed(Instant.EPOCH, ZoneId.of("UTC"))
 
   // This seq is made of random messages which may or may not trigger events
   private val randomSeq =
     Range(0, 1000)
-      .map(i => FakeData(s"sensor_$i", rand.nextFloat() * 200, System.currentTimeMillis(), if (i % 2 == 0) "pari" else "dispari", rand.nextInt(101)))
+      .map(i =>
+        FakeData(
+          s"sensor_$i",
+          rand.nextFloat() * 200,
+          System.currentTimeMillis(),
+          if (i % 2 == 0) "pari" else "dispari",
+          rand.nextInt(101)
+        )
+      )
 
   // This seq is made of pre-defined messages which have to trigger a specific list of events
   private val mixedSeq = Seq(
-    FakeData(            // Triggers both events
+    FakeData( // Triggers both events
       name = "sensor_0",
       temperature = 200,
-      someLong=0,
-      someStuff="dispari",
-      someNumber = 99),
-    FakeData(          //  Trigger no events
+      someLong = 0,
+      someStuff = "dispari",
+      someNumber = 99
+    ),
+    FakeData( //  Trigger no events
       name = "sensor_1",
       temperature = 10,
-      someLong=0,
-      someStuff="pari",
-      someNumber = 100),
-    FakeData(        //   Triggers HighTemperature event in WARN severity
+      someLong = 0,
+      someStuff = "pari",
+      someNumber = 100
+    ),
+    FakeData( //   Triggers HighTemperature event in WARN severity
       name = "sensor_2",
       temperature = 120,
-      someLong=0,
-      someStuff="dispari",
-      someNumber = 50),
-    FakeData(       //   Triggers HighTemperature event with prova2=false as payload
+      someLong = 0,
+      someStuff = "dispari",
+      someNumber = 50
+    ),
+    FakeData( //   Triggers HighTemperature event with prova2=false as payload
       name = "sensor_3",
       temperature = 620,
-      someLong=0,
-      someStuff="dispari",
-      someNumber = 5),
-    FakeData(      //   Trigger OddHighNumbers in LOW_TEMP severity
+      someLong = 0,
+      someStuff = "dispari",
+      someNumber = 5
+    ),
+    FakeData( //   Trigger OddHighNumbers in LOW_TEMP severity
       name = "sensor_4",
       temperature = 45,
-      someLong=0,
-      someStuff="dispari",
-      someNumber = 101)
+      someLong = 0,
+      someStuff = "dispari",
+      someNumber = 101
+    )
   )
   // Control Seq of events generated from mixedSeq (not taking into account timestamp and id)
   private val mixedSeqEvents = Seq(
@@ -66,7 +77,8 @@ class InnerEventStrategySpec extends WordSpec
       severity = "CRITICAL",
       sourceId = Some("sensor_0"),
       eventId = "0",
-      timestamp = 0),
+      timestamp = 0
+    ),
     Event(
       eventRuleName = "HighTemperature",
       source = "streamingSource1",
@@ -85,7 +97,8 @@ class InnerEventStrategySpec extends WordSpec
       severity = "CRITICAL",
       sourceId = Some("sensor_3"),
       eventId = "0",
-      timestamp = 0),
+      timestamp = 0
+    ),
     Event(
       eventRuleName = "OddHighNumbers",
       source = "streamingSource2",
@@ -108,97 +121,97 @@ class InnerEventStrategySpec extends WordSpec
     )
   )
 
-  private val highTempEventsQuantity = 3
+  private val highTempEventsQuantity       = 3
   private val oddHighNumbersEventsQuantity = 2
-  private val totalEventsQuantity = highTempEventsQuantity + oddHighNumbersEventsQuantity
+  private val totalEventsQuantity          = highTempEventsQuantity + oddHighNumbersEventsQuantity
 
   // Fake configuration objects to fetch event rules
-  private val reader =  new InputStreamReader(getClass.getResourceAsStream("inner_event_strategy.conf"))
-  private val fakeConfig = try {
-    ConfigFactory.parseReader(reader)
-  } finally {
-    reader.close()
-  }
+  private val reader = new InputStreamReader(getClass.getResourceAsStream("inner_event_strategy.conf"))
+  private val fakeConfig =
+    try {
+      ConfigFactory.parseReader(reader)
+    } finally {
+      reader.close()
+    }
 
   // This seq doesn't trigger any event
   private val fruitlessSeq = Range(0, 1000)
-    .map(i => FakeData(s"sensor_$i", rand.nextFloat() * 50, System.currentTimeMillis(), if (i % 2 == 0) "pari" else "dispari", rand.nextInt(50)))
+    .map(i =>
+      FakeData(
+        s"sensor_$i",
+        rand.nextFloat() * 50,
+        System.currentTimeMillis(),
+        if (i % 2 == 0) "pari" else "dispari",
+        rand.nextInt(50)
+      )
+    )
 
   // This seq doesn't contain any message
   private val emptySeq: Seq[FakeData] = Seq.empty
 
-
   "When dealing with multiple event rules, InnerEventStrategy" should {
 
     val target = new InnerEventStrategy(fakeConfig.getConfig("multipleRules"), fixedClock, FixedIdGenerator)
 
     //TODO: this should not be a unit test but actually a bench
-    s"Process a big random sequence" in withSparkSession { ss => {
-      import ss.implicits._
-      val events: Array[Event] = target.transform(ss.sparkContext.parallelize(randomSeq).toDF).as[Event].collect()
+    s"Process a big random sequence" in {
+      import spark.implicits._
+      val events: Array[Event] = target.transform(spark.sparkContext.parallelize(randomSeq).toDF).as[Event].collect()
       println(events.length)
-    }}
+    }
 
-    s"Find $totalEventsQuantity test events in mixedSeq which match the control event seq" in withSparkSession { ss => {
-      import ss.implicits._
-      val events: Array[Event] = target.transform(ss.sparkContext.parallelize(mixedSeq).toDF).as[Event].collect()
+    s"Find $totalEventsQuantity test events in mixedSeq which match the control event seq" in {
+      import spark.implicits._
+      val events: Array[Event] = target.transform(spark.sparkContext.parallelize(mixedSeq).toDF).as[Event].collect()
 
-      events.length should be (totalEventsQuantity)
+      events.length should be(totalEventsQuantity)
 
       val (highTempEvents, oddNumberEvents) = events.partition(e => e.eventRuleName.equals("HighTemperature"))
-      highTempEvents.length should be (highTempEventsQuantity)
-      oddNumberEvents.length should be (oddHighNumbersEventsQuantity)
-
-      mixedSeqEvents should contain theSameElementsAs  events
-
-    }}
-
-    "Find no events in fruitlessSeq" in withSparkSession { ss => {
-      import ss.implicits._
-      val events: Array[Event] = target.transform(ss.sparkContext.parallelize(fruitlessSeq).toDF).as[Event].collect()
-      events.length should be (0)
-    }}
-
-    "Find no events in emptySeq" in withSparkSession { ss => {
-      import ss.implicits._
-      val events: Array[Event] = target.transform(ss.sparkContext.parallelize(emptySeq).toDF).as[Event].collect()
-      events.length should be (0)
-    }}
-
+      highTempEvents.length should be(highTempEventsQuantity)
+      oddNumberEvents.length should be(oddHighNumbersEventsQuantity)
+
+      mixedSeqEvents should contain theSameElementsAs events
+    }
+    "Find no events in fruitlessSeq" in {
+      import spark.implicits._
+      val events: Array[Event] = target.transform(spark.sparkContext.parallelize(fruitlessSeq).toDF).as[Event].collect()
+      events.length should be(0)
+    }
+    "Find no events in emptySeq" in {
+      import spark.implicits._
+      val events: Array[Event] = target.transform(spark.sparkContext.parallelize(emptySeq).toDF).as[Event].collect()
+      events.length should be(0)
+    }
   }
 
   "When dealing with a single event rule, InnerEventStrategy" should {
 
     val target = new InnerEventStrategy(fakeConfig.getConfig("singleRule"), fixedClock, FixedIdGenerator)
 
-    s"Find exactly $highTempEventsQuantity in mixedSeq" in withSparkSession { ss => {
-      import ss.implicits._
-      val events: Array[Event] = target.transform(ss.sparkContext.parallelize(mixedSeq).toDF).as[Event].collect()
+    s"Find exactly $highTempEventsQuantity in mixedSeq" in {
+      import spark.implicits._
+      val events: Array[Event] = target.transform(spark.sparkContext.parallelize(mixedSeq).toDF).as[Event].collect()
 
-      events.length should be (highTempEventsQuantity)
+      events.length should be(highTempEventsQuantity)
 
       val (highTempEvents, oddNumberEvents) = events.partition(e => e.eventRuleName.equals("HighTemperature"))
-      highTempEvents.length should be (highTempEventsQuantity)
-      oddNumberEvents.length should be (0)
-    }}
-
-    "Find no events in noEventSeq" in withSparkSession { ss => {
-      import ss.implicits._
-      val events: Array[Event] = target.transform(ss.sparkContext.parallelize(fruitlessSeq).toDF).as[Event].collect()
-      events.length should be (0)
-    }}
-
-    "Find no events in emptySeq" in withSparkSession { ss => {
-      import ss.implicits._
-      val events: Array[Event] = target.transform(ss.sparkContext.parallelize(emptySeq).toDF).as[Event].collect()
-      events.length should be (0)
-    }}
-
+      highTempEvents.length should be(highTempEventsQuantity)
+      oddNumberEvents.length should be(0)
+    }
+    "Find no events in noEventSeq" in {
+      import spark.implicits._
+      val events: Array[Event] = target.transform(spark.sparkContext.parallelize(fruitlessSeq).toDF).as[Event].collect()
+      events.length should be(0)
+    }
+
+    "Find no events in emptySeq" in {
+      import spark.implicits._
+      val events: Array[Event] = target.transform(spark.sparkContext.parallelize(emptySeq).toDF).as[Event].collect()
+      events.length should be(0)
+    }
   }
-
-
 }
 
 case object FixedIdGenerator extends IDGenerator {
   override def generate(): String = "0"
-}
+}