Merge d2eccc9 into 413d5ff

epiphanous · Dec 28, 2022 · 6139b92 · 6139b92
2 parents 413d5ff + d2eccc9
commit 6139b92
Show file tree

Hide file tree

Showing 8 changed files with 44 additions and 16 deletions.
diff --git a/src/main/scala/io/epiphanous/flinkrunner/model/BasePropGenerators.scala b/src/main/scala/io/epiphanous/flinkrunner/model/BasePropGenerators.scala
@@ -47,6 +47,19 @@ trait BasePropGenerators {
     durationGen()
   )
 
+  def simpleJsonStringGen(
+                           min: Int = 1,
+                           max: Int = 90
+                         ): Gen[String] =
+                   for {
+                     num <- Gen.choose(min, max)
+                     key: String <- nameGen("A", num)
+                     value: String <- uuidGen
+                   } yield s"{$key: $value}"
+                    implicit lazy val jsonStringArbitraryResult: Arbitrary[String] = Arbitrary(
+    simpleJsonStringGen()
+  )
+
   def uuidGen: Gen[String]                     = Gen.uuid.map(u => u.toString)
   implicit lazy val uuidArb: Arbitrary[String] = Arbitrary(uuidGen)
 

diff --git a/src/test/avro/BRecord.avsc b/src/test/avro/BRecord.avsc
@@ -6,6 +6,7 @@
     { "name": "b0", "type": "string" },
     { "name": "b1", "type": ["null", "int"] },
     { "name": "b2", "type": ["null", "double"] },
-    { "name": "b3", "type": {"type": "long", "logicalType": "timestamp-millis" }}
+    { "name": "b3", "type": {"type": "long", "logicalType": "timestamp-millis" }},
+    { "name": "b4", "type": "string" }
   ]
 }
diff --git a/src/test/resources/SampleB.csv b/src/test/resources/SampleB.csv
@@ -1,9 +1,9 @@
-gx21ge6,B-NGMZ-5351,200.0,200,2022-08-26T20:37:33.031Z
-zVqYPPA,B-GCWD-8429,268.51074674159236,,2022-08-26T20:37:37.854Z
-nF1kVdP,B-CEWP-2441,299.0,200,2022-08-26T20:37:10.113Z
-b0Q1VjB,B-TFWR-9685,200.0,226,2022-08-26T20:37:17.299Z
-bGlSUta,B-GSZQ-8036,200.0,245,2022-08-26T20:37:13.649Z
-dluBK7m,B-RYOT-2386,200.0,200,2022-08-26T20:37:39.150Z
-edopOkb,B-VSFJ-9246,284.84753169869873,299,2022-08-26T20:37:23.287Z
-w0x0NBB,B-VHHF-7895,299.0,200,2022-08-26T20:36:56.524Z
-i4t00SY,B-RLTY-8415,267.506351122217,223,2022-08-26T20:37:41.671Z
+gx21ge6,B-NGMZ-5351,200.0,200,2022-08-26T20:37:33.031Z,{"a": 5}
+zVqYPPA,B-GCWD-8429,268.51074674159236,,2022-08-26T20:37:37.854Z,{"ASMTDN": "Y"}
+nF1kVdP,B-CEWP-2441,299.0,200,2022-08-26T20:37:10.113Z,{}
+b0Q1VjB,B-TFWR-9685,200.0,226,2022-08-26T20:37:17.299Z,{}
+bGlSUta,B-GSZQ-8036,200.0,245,2022-08-26T20:37:13.649Z,{}
+dluBK7m,B-RYOT-2386,200.0,200,2022-08-26T20:37:39.150Z,{}
+edopOkb,B-VSFJ-9246,284.84753169869873,299,2022-08-26T20:37:23.287Z,{}
+w0x0NBB,B-VHHF-7895,299.0,200,2022-08-26T20:36:56.524Z,{}
+i4t00SY,B-RLTY-8415,267.506351122217,223,2022-08-26T20:37:41.671Z,{}
diff --git a/src/test/scala/io/epiphanous/flinkrunner/PropGenerators.scala b/src/test/scala/io/epiphanous/flinkrunner/PropGenerators.scala
@@ -18,7 +18,8 @@ trait PropGenerators extends BasePropGenerators {
     b1 <- Gen.option(Gen.chooseNum(1, 1000))
     b2 <- Gen.option(Gen.chooseNum(1d, 100d))
     b3 <- instantGen()
-  } yield BRecord(b0, b1, b2, b3)
+    b4 <- simpleJsonStringGen()
+  } yield BRecord(b0, b1, b2, b3, b4)
   implicit lazy val bRecordArb: Arbitrary[BRecord] = Arbitrary(bRecordGen)
 
   lazy val cRecordGen: Gen[CRecord]                = for {
@@ -71,7 +72,8 @@ trait PropGenerators extends BasePropGenerators {
     b1 <- Gen.chooseNum(200d, 299d)
     b2 <- Gen.option(Gen.chooseNum(200, 299))
     ts <- instantGen()
-  } yield SimpleB(id, b0, b1, b2, ts)
+    json <- simpleJsonStringGen()
+  } yield SimpleB(id, b0, b1, b2, ts, json)
   implicit lazy val simpleBArb: Arbitrary[SimpleB] = Arbitrary(simpleBGen)
 
   lazy val simpleCGen: Gen[SimpleC]                = for {

diff --git a/src/test/scala/io/epiphanous/flinkrunner/model/MySimpleADT.scala b/src/test/scala/io/epiphanous/flinkrunner/model/MySimpleADT.scala
@@ -33,6 +33,8 @@ case class SimpleA(id: String, a0: String, a1: Int, ts: Instant)
   *   an Option[Int]
   * @param ts
   *   an Instant
+  * @param json
+  *   a String of json
   */
 case class SimpleB(
     id: String,
@@ -41,7 +43,8 @@ case class SimpleB(
     @JsonDeserialize(contentAs = classOf[java.lang.Integer]) b2: Option[
       Int
     ],
-    ts: Instant)
+    ts: Instant,
+    json: String)
     extends MySimpleADT {
   override def $id: String = id
 

diff --git a/src/test/scala/io/epiphanous/flinkrunner/model/sink/JdbcSinkJobTest.scala b/src/test/scala/io/epiphanous/flinkrunner/model/sink/JdbcSinkJobTest.scala
@@ -57,6 +57,10 @@ class JdbcSinkJobTest extends SinkSpec {
          |          type = TIMESTAMP
          |          nullable = false
          |        }
+         |        {
+         |          name = json
+         |          type = JSON
+         |        }
          |      ]
          |    }
          |  }
@@ -79,7 +83,7 @@ class JdbcSinkJobTest extends SinkSpec {
         ) + "|" + rs.getDouble("b1") + "|" + rs.getInt("b2") + "|" + rs
           .getTimestamp(
             "ts"
-          )
+          ) + "|" + rs.getString("json")
       )
     }
     stmt.close()
@@ -132,6 +136,9 @@ class JdbcSinkJobTest extends SinkSpec {
            |          type = TIMESTAMP
            |          nullable = false
            |        }
+           |        {
+           |          name = b4
+           |          type = JSON
            |      ]
            |    }
            |  }
@@ -152,7 +159,8 @@ class JdbcSinkJobTest extends SinkSpec {
         val b1  = Option(rs.getInt("b1"))
         val b2  = Option(rs.getDouble("b2"))
         val b3  = Instant.ofEpochMilli(rs.getTimestamp("b3").getTime)
-        println(row -> BRecord(b0, b1, b2, b3))
+        val b4 =  rs.getString("b4").trim()
+        println(row -> BRecord(b0, b1, b2, b3, b4))
       }
       stmt.close()
       conn.close()

diff --git a/src/test/scala/io/epiphanous/flinkrunner/serde/DelimitedRowEncoderTest.scala b/src/test/scala/io/epiphanous/flinkrunner/serde/DelimitedRowEncoderTest.scala
@@ -30,7 +30,7 @@ class DelimitedRowEncoderTest
 
   property("encode non-nested avro property") {
     def csvLine(b: BRecord): String =
-      s"${b.b0},${b.b1.getOrElse("")},${b.b2.getOrElse("")},${b.b3.toEpochMilli}${System.lineSeparator()}"
+      s"${b.b0},${b.b1.getOrElse("")},${b.b2.getOrElse("")},${b.b3.toEpochMilli},${b.b4}${System.lineSeparator()}"
     val encoder                     = getTypedRowEncoder[BRecord]()
     forAll { test: BRecord =>
       encoder

diff --git a/src/test/scala/io/epiphanous/flinkrunner/util/AvroUtilsTest.scala b/src/test/scala/io/epiphanous/flinkrunner/util/AvroUtilsTest.scala
@@ -16,6 +16,7 @@ class AvroUtilsTest extends PropSpec {
       .set("b1", spec.b1.getOrElse(null)) // orNull won't work here
       .set("b2", spec.b2.getOrElse(null)) // orNull won't work here
       .set("b3", spec.b3.toEpochMilli)
+      .set("b4", spec.b4)
       .build()
 
   def fromSpecCRecord(spec: CRecord): GenericRecord =