[NU-1557] Test on generated data for Table Source

components-api/src/main/scala/pl/touk/nussknacker/engine/api/test/TestRecordParser.scala

@@ -2,6 +2,6 @@ package pl.touk.nussknacker.engine.api.test
 
 trait TestRecordParser[+T] {
 
-  def parse(testRecord: TestRecord): T
+  def parse(testRecords: List[TestRecord]): List[T]
 
 }

docs/Changelog.md

@@ -28,6 +28,8 @@
 * [#6415](https://github.com/TouK/nussknacker/pull/6415) Added "Component group" field to fragment properties, which allows selection of the group of components in the Creator Panel in which the fragment will be available
 * [#6462](https://github.com/TouK/nussknacker/pull/6462) Improvement of Component's API: `canHaveManyInputs` property is now 
   determined automatically, developer doesn't need to provide it by his/her own
+* [#6195](https://github.com/TouK/nussknacker/pull/6195) Added randomized test data generation for Table Source scenarios
+* [#6340](https://github.com/TouK/nussknacker/pull/6340) Added running tests on generated data for Table Source scenarios
 
 1.16.2 (18 July 2024)
 -------------------------

docs/MigrationGuide.md

@@ -17,6 +17,11 @@ To see the biggest differences please consult the [changelog](Changelog.md).
     Now method `returnType` from `EagerServiceWithStaticParameters` requires implicit nodeId param
   * [#6462](https://github.com/TouK/nussknacker/pull/6462) `CustomStreamTransformer.canHaveManyInputs` field was
     removed. You don't need to implement any other method in replacement, just remove this method.
+* [#6418](https://github.com/TouK/nussknacker/pull/6418) Improvement: Pass implicit nodeId to `EagerServiceWithStaticParameters.returnType`
+    * Now method `returnType` from `EagerServiceWithStaticParameters` requires implicit nodeId param
+* [#6340](https://github.com/TouK/nussknacker/pull/6340) `TestRecordParser` trait used in `SourceTestSupport` trait
+  changed to work on lists instead of single records - its `parse` method now takes `List[TestRecord]` instead of a
+  single `TestRecord` and returns a list of results instead of a single result.
 
 ### REST API changes
 

e2e-tests/src/test/scala/pl/touk/nussknacker/BatchDataGenerationSpec.scala

@@ -27,11 +27,13 @@ class BatchDataGenerationSpec
 
   private val designerServiceUrl = "http://localhost:8080"
 
+  override def beforeAll(): Unit = {
+    createBatchScenario(simpleBatchTableScenario.name.value)
+    super.beforeAll()
+  }
+
   "Generate file endpoint should generate records with randomized values for scenario with table source" in {
     given()
-      .applicationState(
-        createBatchScenario(simpleBatchTableScenario.name.value)
-      )
       .when()
       .request()
       .basicAuthAdmin()
@@ -55,6 +57,143 @@ class BatchDataGenerationSpec
       )
   }
 
+  "Test on generated data endpoint should return results and counts for scenario with table source" in {
+    given()
+      .when()
+      .request()
+      .basicAuthAdmin()
+      .jsonBody(toScenarioGraph(simpleBatchTableScenario).asJson.spaces2)
+      .post(
+        s"$designerServiceUrl/api/processManagement/generateAndTest/${simpleBatchTableScenario.name.value}/1"
+      )
+      .Then()
+      .statusCode(200)
+      .body(
+        matchJsonWithRegexValues(s"""{
+             |  "results": {
+             |    "nodeResults": {
+             |      "sourceId": [
+             |        {
+             |          "id": "SumTransactions-sourceId-0-0",
+             |          "variables": {
+             |            "input": {
+             |              "pretty": {
+             |                 "datetime": "${regexes.localDateTimeRegex}",
+             |                 "client_id": "[a-z\\\\d]{100}",
+             |                 "amount": "${regexes.decimalRegex}"
+             |              }
+             |            }
+             |          }
+             |        }
+             |      ],
+             |      "end": [
+             |        {
+             |          "id": "SumTransactions-sourceId-0-0",
+             |          "variables": {
+             |            "input": {
+             |              "pretty": {
+             |                 "datetime": "${regexes.localDateTimeRegex}",
+             |                 "client_id": "[a-z\\\\d]{100}",
+             |                 "amount": "${regexes.decimalRegex}"
+             |              }
+             |            }
+             |          }
+             |        }
+             |      ]
+             |    },
+             |    "invocationResults": {},
+             |    "externalInvocationResults": {},
+             |    "exceptions": []
+             |  },
+             |  "counts": {
+             |      "sourceId": {
+             |        "all": 1,
+             |        "errors": 0,
+             |        "fragmentCounts": {}
+             |      },
+             |      "end": {
+             |        "all": 1,
+             |        "errors": 0,
+             |        "fragmentCounts": {}
+             |      }
+             |  }
+             |}""".stripMargin)
+      )
+  }
+
+  "Test from file endpoint should return results and counts for scenario with table source" in {
+    given()
+      .when()
+      .request()
+      .basicAuthAdmin()
+      .multiPart(
+        "scenarioGraph",
+        toScenarioGraph(simpleBatchTableScenario).asJson.spaces2,
+        "application/json"
+      )
+      .multiPart(
+        "testData",
+        """{"sourceId":"sourceId","record":{"datetime":"2024-07-19 08:56:08.485","client_id":"aClientId","amount":123123.12}}""",
+        "text/ plain"
+      )
+      .post(
+        s"$designerServiceUrl/api/processManagement/test/${simpleBatchTableScenario.name.value}"
+      )
+      .Then()
+      .statusCode(200)
+      .body(
+        matchJsonWithRegexValues(s"""{
+             |  "results": {
+             |    "nodeResults": {
+             |      "sourceId": [
+             |        {
+             |          "id": "SumTransactions-sourceId-0-0",
+             |          "variables": {
+             |            "input": {
+             |              "pretty": {
+             |                 "datetime": "2024-07-19T08:56:08.485",
+             |                 "client_id": "aClientId",
+             |                 "amount": "123123.12"
+             |              }
+             |            }
+             |          }
+             |        }
+             |      ],
+             |      "end": [
+             |        {
+             |          "id": "SumTransactions-sourceId-0-0",
+             |          "variables": {
+             |            "input": {
+             |              "pretty": {
+             |                 "datetime": "2024-07-19T08:56:08.485",
+             |                 "client_id": "aClientId",
+             |                 "amount": "123123.12"
+             |              }
+             |            }
+             |          }
+             |        }
+             |      ]
+             |    },
+             |    "invocationResults": {},
+             |    "externalInvocationResults": {},
+             |    "exceptions": []
+             |  },
+             |  "counts": {
+             |      "sourceId": {
+             |        "all": 1,
+             |        "errors": 0,
+             |        "fragmentCounts": {}
+             |      },
+             |      "end": {
+             |        "all": 1,
+             |        "errors": 0,
+             |        "fragmentCounts": {}
+             |      }
+             |  }
+             |}""".stripMargin)
+      )
+  }
+
   private def createBatchScenario(scenarioName: String): Unit = {
     given()
       .when()

engine/flink/components/base-tests/src/test/scala/pl/touk/nussknacker/engine/flink/table/source/TableSourceDataGenerationTest.scala

@@ -7,25 +7,36 @@ import pl.touk.nussknacker.engine.flink.table.extractor.SqlTestData.SimpleTypesT
 
 class TableSourceDataGenerationTest extends AnyFunSuite with Matchers {
 
+  private val tableSource = new TableSource(
+    tableDefinition = SimpleTypesTestCase.tableDefinition,
+    sqlStatements = SqlStatementReader.readSql(SimpleTypesTestCase.sqlStatement),
+    enableFlinkBatchExecutionMode = true
+  )
+
   /*
   Note: Testing features like data generation or scenario testing (like ad hoc test) requires a full e2e test where
         Designer is deployed separately (like in a docker container). This is because these features rely on Flink
         Minicluster which relies on proper classloader setup, which is hard to do in simple tests. These tests below
         are useful for checking the output of these methods, but if they pass it doesn't mean that it works e2e.
    */
   test("table source should generate random records with given schema") {
-    val tableSource = new TableSource(
-      tableDefinition = SimpleTypesTestCase.tableDefinition,
-      sqlStatements = SqlStatementReader.readSql(SimpleTypesTestCase.sqlStatement),
-      enableFlinkBatchExecutionMode = true
-    )
-    val records = tableSource.generateTestData(10)
-
-    records.testRecords.size shouldBe 10
+    val records = tableSource.generateTestData(1)
+
+    records.testRecords.size shouldBe 1
     val mapRecord = records.testRecords.head.json.asObject.get.toMap
+
     mapRecord("someString").isString shouldBe true
     mapRecord("someVarChar").isString shouldBe true
     mapRecord("someInt").isNumber shouldBe true
   }
 
+  test("table source should parse json records") {
+    val testData = tableSource.generateTestData(1).testRecords
+    val result   = tableSource.testRecordParser.parse(testData).head
+
+    result.get("someString") shouldBe a[String]
+    result.get("someVarChar") shouldBe a[String]
+    result.get("someInt") shouldBe a[Number]
+  }
+
 }