tests

Author: Piotr Fałdrowicz

engine/flink/components/base-tests/src/test/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/JavaCollectionsSerializationTest.scala

@@ -14,47 +14,56 @@ import pl.touk.nussknacker.engine.flink.test.FlinkSpec
 import pl.touk.nussknacker.engine.flink.test.ScalatestMiniClusterJobStatusCheckingOps.miniClusterWithServicesToOps
 import pl.touk.nussknacker.engine.flink.util.source.CollectionSource
 import pl.touk.nussknacker.engine.flink.util.transformer.FlinkBaseComponentProvider
+import pl.touk.nussknacker.engine.graph.expression.Expression
 import pl.touk.nussknacker.engine.process.helpers.ConfigCreatorWithCollectingListener
 import pl.touk.nussknacker.engine.process.runner.FlinkScenarioUnitTestJob
 import pl.touk.nussknacker.engine.spel.SpelExtension._
 import pl.touk.nussknacker.engine.testing.LocalModelData
 import pl.touk.nussknacker.engine.testmode.{ResultsCollectingListener, ResultsCollectingListenerHolder}
 
+import java.util
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
+import scala.util.Random
 
 class JavaCollectionsSerializationTest extends AnyFunSuite with FlinkSpec with Matchers with Inside {
 
   private val processId = "aggregateFilterProcess"
 
-  private val process: CanonicalProcess =
-    ScenarioBuilder
+  private def process(expressionOption: Option[Expression] = None): CanonicalProcess = {
+    val scenario = ScenarioBuilder
       .streaming(processId)
       .parallelism(1)
       .source("start", "start")
+
+    expressionOption
+      .map(expression => scenario.buildSimpleVariable("mapVariable", "mapVariable", expression))
+      .getOrElse(scenario)
       .customNodeNoOutput(
         "delay",
         "delay",
         "key"   -> "#input.id".spel,
         "delay" -> "T(java.time.Duration).parse('PT30M')".spel
       )
       .emptySink("end", "dead-end")
+  }
+
+  private val record = Record(
+    id = "2",
+    map = mutable.Map(1 -> "a").asJava,
+    list = mutable.ListBuffer("abc").asJava,
+    set = mutable.Set("def").asJava
+  )
 
   // In Scala 2.13 all java collections class wrappers were rewritten from case class to regular class. Now kryo does not
   // serialize them properly, so JavaWrapperScala2_13Registrar class was added to fix this issue. This test verifies
   // if we can serialize and deserialize records properly.
   test("should serialize record with java map, list and set") {
-    val record = Record(
-      id = "2",
-      map = mutable.Map(1 -> "a").asJava,
-      list = mutable.ListBuffer("abc").asJava,
-      set = mutable.Set("def").asJava
-    )
 
     ResultsCollectingListenerHolder.withListener { collectingListener =>
       val model = modelData(collectingListener, List(record))
 
-      runScenario(model, process)
+      runScenario(model, process())
 
       val result = collectingListener.results
         .nodeResults("end")
@@ -64,6 +73,40 @@ class JavaCollectionsSerializationTest extends AnyFunSuite with FlinkSpec with M
     }
   }
 
+  test("should serialize java map without changing fields order") {
+
+    ResultsCollectingListenerHolder.withListener { collectingListener =>
+      val model = modelData(collectingListener, List(record))
+
+      val sampleMap =
+        ('a' to 'z')
+          .map(x => x -> Random.nextInt())
+          .sortBy(_._2)
+
+      runScenario(
+        model,
+        process(
+          Some(
+            sampleMap
+              .map { case (c, i) => s""""$c" : $i""" }
+              .mkString("{", ",", "}")
+              .spel
+          )
+        )
+      )
+
+      val linkedHashMap = new util.LinkedHashMap[String, AnyRef]()
+      sampleMap.foreach { case (char, int) => linkedHashMap.put(s"$char", Integer.valueOf(int)) }
+
+      val result = collectingListener.results
+        .nodeResults("end")
+        .map(_.variableTyped[Map[_, _]]("mapVariable"))
+
+      result shouldBe List(Some(linkedHashMap))
+
+    }
+  }
+
   def modelData(collectingListener: ResultsCollectingListener[Any], list: List[Record] = List()): LocalModelData = {
     val sourceComponent = SourceFactory.noParamUnboundedStreamFactory[Record](
       CollectionSource[Record](list, None, Typed.fromDetailedType[List[Record]])

utils/utils/src/test/scala/pl/touk/nussknacker/engine/util/json/ToJsonEncoderSpec.scala

@@ -10,6 +10,7 @@ import java.time._
 import java.util
 import java.util.UUID
 import scala.collection.immutable.{ListMap, ListSet}
+import scala.util.Random
 
 class ToJsonEncoderSpec extends AnyFunSpec with Matchers {
 
@@ -72,6 +73,19 @@ class ToJsonEncoderSpec extends AnyFunSpec with Matchers {
     encoder.encode(map) shouldEqual obj("key1" -> fromLong(1), "key2" -> fromString("value"))
   }
 
+  it("should encode maps as a json without changing fields order") {
+    val sampleMap =
+      ('a' to 'z')
+        .map(x => x -> Random.nextInt())
+        .sortBy(_._2)
+    val linkedHashMap = new util.LinkedHashMap[String, AnyRef]()
+    sampleMap.foreach { case (char, int) => linkedHashMap.put(s"$char", Integer.valueOf(int)) }
+
+    val encoded  = encoder.encode(linkedHashMap)
+    val expected = obj(sampleMap.map { case (c, i) => s"$c" -> fromInt(i) }: _*)
+    encoded.asObject.map(_.keys.toList) shouldEqual expected.asObject.map(_.keys.toList)
+  }
+
   it("should encode arrays as a json") {
     encoder.encode(Array(1, 2, 3)) shouldEqual arr(fromLong(1), fromLong(2), fromLong(3))
     encoder.encode(Seq(Array(1, 2, 3))) shouldEqual arr(arr(fromLong(1), fromLong(2), fromLong(3)))