[NU-2183] Tests cases support

common-api/src/main/scala/pl/touk/nussknacker/engine/graph/Test.scala

@@ -0,0 +1,24 @@
+package pl.touk.nussknacker.engine.graph
+
+import io.circe.generic.JsonCodec
+import pl.touk.nussknacker.engine.graph.Test.NodeName
+import pl.touk.nussknacker.engine.graph.expression.Expression
+
+case object Test {
+  type NodeName = String
+}
+
+@JsonCodec final case class Test(
+    id: String,
+    inputs: Map[NodeName, List[TestSourceInput]],
+    mocks: Map[NodeName, EnricherMock],
+    assertions: Map[NodeName, List[Assertion]],
+)
+
+@JsonCodec final case class TestSourceInput(expression: Expression)
+
+//todo: type
+@JsonCodec final case class EnricherMock(expression: Expression)
+
+//todo: type
+@JsonCodec final case class Assertion(expression: String)

components-api/src/main/scala/pl/touk/nussknacker/engine/api/context/ProcessCompilationError.scala

@@ -468,4 +468,20 @@ object ProcessCompilationError {
   object TrailingSpacesId                                                     extends IdErrorType
   final case class IllegalCharactersId(illegalCharacterHumanReadable: String) extends IdErrorType
 
+  // todo: it can be separated from ProcessCompilationError hierarchy (depends on if we finally put test compilation as part of process compilation)
+  sealed trait TestCompilationError extends ProcessCompilationError
+
+  sealed trait TestConfigurationPart
+  object InputData extends TestConfigurationPart
+  object Assertion extends TestConfigurationPart
+  object Mock      extends TestConfigurationPart
+
+  final case class TestConfigurationRefersToNotExistingNode(
+      nodeId: NodeId,
+      testId: String,
+      configurationPart: TestConfigurationPart
+  ) extends TestCompilationError {
+    override val nodeIds: Set[String] = Set.empty
+  }
+
 }

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

@@ -3,6 +3,7 @@ package pl.touk.nussknacker.engine.api.test
 import io.circe.Json
 import pl.touk.nussknacker.engine.api.NodeId
 import pl.touk.nussknacker.engine.api.parameter.ParameterName
+import pl.touk.nussknacker.engine.graph.Test
 import pl.touk.nussknacker.engine.graph.expression.Expression
 
 sealed trait ScenarioTestRecord {
@@ -21,15 +22,19 @@ object ScenarioTestJsonRecord {
 
 }
 
+sealed trait ScenarioTestData
+
 /**
  * Holds test records for a scenario. The difference to [[TestData]] is that records are assigned to the individual sources in the scenario.
  */
-case class ScenarioTestData(testRecords: List[ScenarioTestRecord])
+//todo: merge together
+case class SimpleScenarioTestData(testRecords: List[ScenarioTestRecord]) extends ScenarioTestData
+case class TestCaseScenarioTestData(test: Test)                          extends ScenarioTestData
 
 object ScenarioTestData {
 
   def apply(sourceId: String, parameterExpressions: Map[ParameterName, Expression]): ScenarioTestData = {
-    ScenarioTestData(List(ScenarioTestParametersRecord(NodeId(sourceId), parameterExpressions)))
+    SimpleScenarioTestData(List(ScenarioTestParametersRecord(NodeId(sourceId), parameterExpressions)))
   }
 
 }

designer/restmodel/src/main/scala/pl/touk/nussknacker/restmodel/validation/PrettyValidationErrors.scala

@@ -282,6 +282,11 @@ object PrettyValidationErrors {
             s"Incompatible change to the parameter's definition detected. $parameterEditor editor doesn't support '$language' language",
           paramName = Some(paramName)
         )
+      case TestConfigurationRefersToNotExistingNode(nodeId, testId, _) =>
+        node(
+          message = s"Test with id $testId refers to not existing node $nodeId",
+          description = s"Test configuration can't refer to not existing nodes",
+        )
     }
   }
 

designer/server/src/main/scala/pl/touk/nussknacker/ui/process/test/ScenarioTestService.scala

@@ -7,14 +7,15 @@ import com.typesafe.scalalogging.LazyLogging
 import pl.touk.nussknacker.engine.api.{MetaData, ProcessVersion}
 import pl.touk.nussknacker.engine.api.definition.Parameter
 import pl.touk.nussknacker.engine.api.graph.ScenarioGraph
-import pl.touk.nussknacker.engine.api.test.ScenarioTestData
+import pl.touk.nussknacker.engine.api.test.{ScenarioTestData, TestCaseScenarioTestData}
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.definition.test.{TestInfoProvider, TestingCapabilities}
 import pl.touk.nussknacker.engine.definition.test.TestInfoProvider.{
   ParametersDefinitionError,
   SourceTestDataGenerationError,
   TestingCapabilitiesError
 }
+import pl.touk.nussknacker.engine.graph.Test
 import pl.touk.nussknacker.engine.graph.node.SourceNodeData
 import pl.touk.nussknacker.engine.testmode.TestProcess.TestResults
 import pl.touk.nussknacker.restmodel.definition.UISourceParameters
@@ -117,6 +118,30 @@ class ScenarioTestService(
     } yield rawTestData
   }
 
+  def performTest(
+      scenarioGraph: ScenarioGraph,
+      processVersion: ProcessVersion,
+      isFragment: Boolean,
+      test: Test,
+  )(implicit ec: ExecutionContext, user: LoggedUser): Future[Either[PerformTestError, ResultsWithCounts]] = {
+    val canonical = toCanonicalProcess(
+      scenarioGraph,
+      processVersion,
+      isFragment
+    )
+    val scenarioTestData = TestCaseScenarioTestData(test)
+    (for {
+      testResults <- EitherT.liftF(
+        testExecutorService.testProcess(
+          processVersion,
+          canonical,
+          scenarioTestData,
+        )
+      )
+      _ <- EitherT.fromEither[Future](validateTestResultsAreNotTooBig(testResults))
+    } yield ResultsWithCounts(testResults, computeCounts(canonical, isFragment, testResults))).value
+  }
+
   def performTest(
       scenarioGraph: ScenarioGraph,
       processVersion: ProcessVersion,

designer/server/src/test/scala/pl/touk/nussknacker/ui/api/ManagementApiHttpServiceBusinessSpec.scala

@@ -46,6 +46,7 @@ class ManagementApiHttpServiceBusinessSpec
                 invocationResults = Map.empty,
                 externalInvocationResults = Map.empty,
                 exceptions = List.empty,
+                assertionsResults = Map.empty
               )
             )
           )

engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/TestFlinkProcessCompilerDataFactory.scala

@@ -7,7 +7,7 @@ import pl.touk.nussknacker.engine.{ModelConfig, ModelData, RuntimeMode}
 import pl.touk.nussknacker.engine.api._
 import pl.touk.nussknacker.engine.api.component.NodesDeploymentData
 import pl.touk.nussknacker.engine.api.process._
-import pl.touk.nussknacker.engine.api.test.ScenarioTestData
+import pl.touk.nussknacker.engine.api.test.{ScenarioTestData, SimpleScenarioTestData, TestCaseScenarioTestData}
 import pl.touk.nussknacker.engine.api.typed.typing.TypingResult
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.compile.nodecompilation.EvaluableLazyParameterCreator
@@ -159,8 +159,12 @@ class StubbedSourcePreparer(
   }
 
   private def collectSamples(originalSource: Source, nodeId: NodeId): List[Object] = {
-    val testRecordsForSource = scenarioTestData.testRecords.filter(_.sourceId == nodeId)
-    testDataPreparer.prepareRecordsForTest(originalSource, testRecordsForSource)
+    scenarioTestData match {
+      case SimpleScenarioTestData(testRecords) =>
+        testDataPreparer.prepareRecordsForTest(originalSource, testRecords.filter(_.sourceId == nodeId))
+      case TestCaseScenarioTestData(test) =>
+        testDataPreparer.prepareRecordsForTestCase(test.inputs.getOrElse(nodeId.id, List.empty))(nodeId)
+    }
   }
 
 }

engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/compiler/StubbedFlinkProcessCompilerDataFactoryTest.scala

@@ -102,7 +102,7 @@ class StubbedFlinkProcessCompilerDataFactoryTest extends AnyFunSuite with Matche
 
   test("stubbing for test purpose should work for one source") {
     val scenarioTestData =
-      ScenarioTestData(List(1, 2, 3).map(v => ScenarioTestJsonRecord("left-source", Json.fromLong(v))))
+      SimpleScenarioTestData(List(1, 2, 3).map(v => ScenarioTestJsonRecord("left-source", Json.fromLong(v))))
     val compiledProcess = testCompile(scenarioWithSingleSource, scenarioTestData)
     val sources = compiledProcess.sources.collect { case source: SourcePart =>
       source.obj
@@ -112,7 +112,7 @@ class StubbedFlinkProcessCompilerDataFactoryTest extends AnyFunSuite with Matche
   }
 
   test("stubbing for test purpose should work for multiple sources") {
-    val scenarioTestData = ScenarioTestData(
+    val scenarioTestData = SimpleScenarioTestData(
       List(
         ScenarioTestJsonRecord("left-source", Json.fromLong(11)),
         ScenarioTestJsonRecord("right-source", Json.fromLong(21)),
@@ -137,7 +137,7 @@ class StubbedFlinkProcessCompilerDataFactoryTest extends AnyFunSuite with Matche
   }
 
   test("stubbing for test purpose should work for one source using parameter record") {
-    val scenarioTestData = ScenarioTestData(
+    val scenarioTestData = SimpleScenarioTestData(
       List(1, 2, 3).map(v =>
         ScenarioTestParametersRecord(
           NodeId("left-source"),

engine/flink/management/src/it/scala/pl/touk/nussknacker/engine/management/streaming/FlinkDeploymentManagerScenarioTestingSpec.scala

@@ -11,8 +11,11 @@ import pl.touk.nussknacker.engine.ConfigWithUnresolvedVersion
 import pl.touk.nussknacker.engine.api.ProcessVersion
 import pl.touk.nussknacker.engine.api.deployment.DMTestScenarioCommand
 import pl.touk.nussknacker.engine.api.process.ProcessName
-import pl.touk.nussknacker.engine.api.test.{ScenarioTestData, ScenarioTestJsonRecord}
+import pl.touk.nussknacker.engine.api.test.{ScenarioTestJsonRecord, SimpleScenarioTestData, TestCaseScenarioTestData}
 import pl.touk.nussknacker.engine.build.ScenarioBuilder
+import pl.touk.nussknacker.engine.graph.{Test, TestSourceInput}
+import pl.touk.nussknacker.engine.graph.expression.Expression
+import pl.touk.nussknacker.engine.graph.expression.Expression.Language.JsonTemplate
 import pl.touk.nussknacker.engine.testmode.TestProcess.ResultContext
 import pl.touk.nussknacker.test.{KafkaConfigProperties, VeryPatientScalaFutures, WithConfig}
 
@@ -43,7 +46,7 @@ class FlinkDeploymentManagerScenarioTestingSpec
       .withValue("category", fromAnyRef("Category1"))
   }
 
-  private val scenarioTestData = ScenarioTestData(
+  private val scenarioTestData = SimpleScenarioTestData(
     List(ScenarioTestJsonRecord("startProcess", Json.fromString("terefere")))
   )
 
@@ -73,6 +76,30 @@ class FlinkDeploymentManagerScenarioTestingSpec
     }
   }
 
+  it should "run scenario as a test case" in {
+    val testCaseTestData = TestCaseScenarioTestData(
+      Test(
+        "someTest",
+        Map("startProcess" -> List(TestSourceInput(Expression(JsonTemplate, "\"terefere\"")))),
+        Map.empty,
+        Map.empty
+      )
+    )
+
+    val processName    = ProcessName(UUID.randomUUID().toString)
+    val processVersion = ProcessVersion.empty.copy(processName = processName)
+
+    val process = SampleProcess.prepareProcess(processName)
+
+    whenReady(deploymentManager.processCommand(DMTestScenarioCommand(processVersion, process, testCaseTestData))) { r =>
+      r.nodeResults shouldBe Map(
+        "startProcess" -> List(ResultContext(s"$processName-startProcess-0-0", Map("input" -> variable("terefere")))),
+        "nightFilter"  -> List(ResultContext(s"$processName-startProcess-0-0", Map("input" -> variable("terefere")))),
+        "endSend"      -> List(ResultContext(s"$processName-startProcess-0-0", Map("input" -> variable("terefere"))))
+      )
+    }
+  }
+
   it should "return correct error messages" in {
     val processName    = ProcessName(UUID.randomUUID().toString)
     val processVersion = ProcessVersion.empty.copy(processName = processName)