From 0263d7269bccc1831050969aba0112ee37351d77 Mon Sep 17 00:00:00 2001
From: Robert Balaban <rbalaban@mdsol.com>
Date: Mon, 31 Oct 2022 14:45:02 +0200
Subject: [PATCH] test: add embedded avro test case

---
 .../flinkrunner/model/FlinkConfig.scala       |  7 +---
 ...KafkaRecordDeserializationSchemaTest.scala |  2 ++
 .../flinkrunner/util/AvroUtilsTest.scala      | 34 +++++++++++++++----
 3 files changed, 31 insertions(+), 12 deletions(-)

diff --git a/src/main/scala/io/epiphanous/flinkrunner/model/FlinkConfig.scala b/src/main/scala/io/epiphanous/flinkrunner/model/FlinkConfig.scala
index 652b35fe..10f5b70e 100644
--- a/src/main/scala/io/epiphanous/flinkrunner/model/FlinkConfig.scala
+++ b/src/main/scala/io/epiphanous/flinkrunner/model/FlinkConfig.scala
@@ -1,11 +1,6 @@
 package io.epiphanous.flinkrunner.model
 
-import com.typesafe.config.{
-  Config,
-  ConfigFactory,
-  ConfigObject,
-  ConfigOriginFactory
-}
+import com.typesafe.config.{Config, ConfigFactory, ConfigObject, ConfigOriginFactory}
 import com.typesafe.scalalogging.LazyLogging
 import io.epiphanous.flinkrunner.util.ConfigToProps.RichConfigObject
 import io.epiphanous.flinkrunner.util.FileUtils.getResourceOrFile
diff --git a/src/test/scala/io/epiphanous/flinkrunner/serde/ConfluentAvroRegistryKafkaRecordDeserializationSchemaTest.scala b/src/test/scala/io/epiphanous/flinkrunner/serde/ConfluentAvroRegistryKafkaRecordDeserializationSchemaTest.scala
index de65bc1f..9c4179f7 100644
--- a/src/test/scala/io/epiphanous/flinkrunner/serde/ConfluentAvroRegistryKafkaRecordDeserializationSchemaTest.scala
+++ b/src/test/scala/io/epiphanous/flinkrunner/serde/ConfluentAvroRegistryKafkaRecordDeserializationSchemaTest.scala
@@ -43,4 +43,6 @@ class ConfluentAvroRegistryKafkaRecordDeserializationSchemaTest
     collected.head shouldEqual aWrapper
   }
 
+
+
 }
diff --git a/src/test/scala/io/epiphanous/flinkrunner/util/AvroUtilsTest.scala b/src/test/scala/io/epiphanous/flinkrunner/util/AvroUtilsTest.scala
index 88b3d166..dabebcaa 100644
--- a/src/test/scala/io/epiphanous/flinkrunner/util/AvroUtilsTest.scala
+++ b/src/test/scala/io/epiphanous/flinkrunner/util/AvroUtilsTest.scala
@@ -1,12 +1,7 @@
 package io.epiphanous.flinkrunner.util
 
 import io.epiphanous.flinkrunner.PropSpec
-import io.epiphanous.flinkrunner.model.{
-  BRecord,
-  BWrapper,
-  FlinkConfig,
-  MyAvroADT
-}
+import io.epiphanous.flinkrunner.model.{BRecord, BWrapper, CRecord, FlinkConfig, MyAvroADT}
 import io.epiphanous.flinkrunner.util.AvroUtils.GenericToSpecific
 import org.apache.avro.SchemaBuilder
 import org.apache.avro.generic.{GenericRecord, GenericRecordBuilder}
@@ -23,6 +18,15 @@ class AvroUtilsTest extends PropSpec {
       .set("b3", spec.b3.toEpochMilli)
       .build()
 
+  def fromSpecCRecord(spec: CRecord): GenericRecord =
+    new GenericRecordBuilder(CRecord.SCHEMA$)
+      .set("id", spec.id)
+      .set("cOptInt", spec.cOptInt.getOrElse(null))
+      .set("cOptDouble", spec.cOptDouble.getOrElse(null))
+      .set("bRecord", spec.bRecord.getOrElse(null))
+      .set("ts", spec.ts.toEpochMilli)
+      .build()
+
   property("isGeneric property") {
     AvroUtils.isGeneric(classOf[GenericRecord]) shouldEqual true
     AvroUtils.isGeneric(classOf[BRecord]) shouldEqual false
@@ -41,6 +45,24 @@ class AvroUtilsTest extends PropSpec {
     }
   }
 
+  property("GenericToSpecific embedded Avro") {
+    forAll { c: CRecord =>
+      val v = fromSpecCRecord(c)
+      val s = v.toSpecific(new CRecord())
+      print(s)
+      s shouldEqual c
+    }
+  }
+
+  property("instanceOf property embedded avro") {
+    val x = AvroUtils.instanceOf(classOf[CRecord])
+    val y = new CRecord()
+    x.id shouldEqual y.id
+    x.cOptInt shouldEqual y.cOptInt
+    x.cOptDouble shouldEqual y.cOptDouble
+    x.bRecord shouldEqual y.bRecord
+  }
+
   property("instanceOf property") {
     val x = AvroUtils.instanceOf(classOf[BRecord])
     val y = new BRecord()