Revert "Remove T from TestResults"

This reverts commit 9aca47b.
TouK · Oct 27, 2023 · d5fa855 · d5fa855
1 parent b309a27
commit d5fa855
Show file tree

Hide file tree

Showing 36 changed files with 271 additions and 254 deletions.
diff --git a/components-api/src/main/scala/pl/touk/nussknacker/engine/api/Context.scala b/components-api/src/main/scala/pl/touk/nussknacker/engine/api/Context.scala
@@ -24,8 +24,6 @@ object Context {
 
   def apply(id: String): Context = Context(id, Map.empty, None)
 
-  def apply(id: String, variables: Map[String, Any]): Context = Context(id, variables, None)
-
 }
 
 case class ContextId(value: String)

diff --git a/...ager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentManager.scala b/...ager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentManager.scala
@@ -59,11 +59,12 @@ trait DeploymentManager extends AutoCloseable {
 
   def cancel(name: ProcessName, deploymentId: DeploymentId, user: User): Future[Unit]
 
-  def test(
+  def test[T](
       name: ProcessName,
       canonicalProcess: CanonicalProcess,
-      scenarioTestData: ScenarioTestData
-  ): Future[TestResults]
+      scenarioTestData: ScenarioTestData,
+      variableEncoder: Any => T
+  ): Future[TestResults[T]]
 
   def getProcessStates(name: ProcessName)(
       implicit freshnessPolicy: DataFreshnessPolicy

diff --git a/...l/touk/nussknacker/engine/api/deployment/cache/CachingProcessStateDeploymentManager.scala b/...l/touk/nussknacker/engine/api/deployment/cache/CachingProcessStateDeploymentManager.scala
@@ -70,12 +70,13 @@ class CachingProcessStateDeploymentManager(delegate: DeploymentManager, cacheTTL
   override def cancel(name: ProcessName, deploymentId: DeploymentId, user: User): Future[Unit] =
     delegate.cancel(name, deploymentId, user)
 
-  override def test(
+  override def test[T](
       name: ProcessName,
       canonicalProcess: CanonicalProcess,
-      scenarioTestData: ScenarioTestData
-  ): Future[TestProcess.TestResults] =
-    delegate.test(name, canonicalProcess, scenarioTestData)
+      scenarioTestData: ScenarioTestData,
+      variableEncoder: Any => T
+  ): Future[TestProcess.TestResults[T]] =
+    delegate.test(name, canonicalProcess, scenarioTestData, variableEncoder)
 
   override def processStateDefinitionManager: ProcessStateDefinitionManager = delegate.processStateDefinitionManager
 

diff --git a/...manager-api/src/main/scala/pl/touk/nussknacker/engine/testing/DeploymentManagerStub.scala b/...manager-api/src/main/scala/pl/touk/nussknacker/engine/testing/DeploymentManagerStub.scala
@@ -55,11 +55,12 @@ class DeploymentManagerStub extends DeploymentManager with AlwaysFreshProcessSta
 
   override def cancel(name: ProcessName, deploymentId: DeploymentId, user: User): Future[Unit] = Future.successful(())
 
-  override def test(
+  override def test[T](
       name: ProcessName,
       canonicalProcess: CanonicalProcess,
-      scenarioTestData: ScenarioTestData
-  ): Future[TestProcess.TestResults] = ???
+      scenarioTestData: ScenarioTestData,
+      variableEncoder: Any => T
+  ): Future[TestProcess.TestResults[T]] = ???
 
   // We map lastStateAction to state to avoid some corner/blocking cases with the deleting/canceling scenario on tests..
   override def getProcessState(idWithName: ProcessIdWithName, lastStateAction: Option[ProcessAction])(

diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/api/ManagementResources.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/api/ManagementResources.scala
@@ -16,7 +16,7 @@ import pl.touk.nussknacker.engine.api.{Context, DisplayJson}
 import pl.touk.nussknacker.engine.api.component.{ComponentInfo, NodeComponentInfo}
 import pl.touk.nussknacker.engine.api.deployment._
 import pl.touk.nussknacker.engine.api.exception.NuExceptionInfo
-import pl.touk.nussknacker.engine.testmode.TestProcess.{ExpressionInvocationResult, _}
+import pl.touk.nussknacker.engine.testmode.TestProcess._
 import pl.touk.nussknacker.engine.util.json.BestEffortJsonEncoder
 import pl.touk.nussknacker.restmodel.displayedgraph.DisplayableProcess
 import pl.touk.nussknacker.restmodel.{CustomActionRequest, CustomActionResponse}
@@ -42,25 +42,37 @@ object ManagementResources {
 
   import pl.touk.nussknacker.engine.api.CirceUtil._
 
-  implicit val resultsWithCountsEncoder: Encoder[ResultsWithCounts] = deriveConfiguredEncoder
+  val testResultsVariableEncoder: Any => io.circe.Json = {
+    case displayable: DisplayJson =>
+      def safeString(a: String) = Option(a).map(Json.fromString).getOrElse(Json.Null)
 
-  implicit val testResultsEncoder: Encoder[TestResults] = new Encoder[TestResults]() {
+      val displayableJson = displayable.asJson
+      displayable.originalDisplay match {
+        case None           => Json.obj("pretty" -> displayableJson)
+        case Some(original) => Json.obj("original" -> safeString(original), "pretty" -> displayableJson)
+      }
+    case null => Json.Null
+    case a =>
+      Json.obj(
+        "pretty" -> BestEffortJsonEncoder(failOnUnknown = false, a.getClass.getClassLoader).circeEncoder.apply(a)
+      )
+  }
 
-    implicit val anyEncoder: Encoder[Any] = {
-      case displayable: DisplayJson =>
-        def safeString(a: String) = Option(a).map(Json.fromString).getOrElse(Json.Null)
+  implicit val resultsWithCountsEncoder: Encoder[ResultsWithCounts[Json]] = deriveConfiguredEncoder
 
-        val displayableJson = displayable.asJson
-        displayable.originalDisplay match {
-          case None           => Json.obj("pretty" -> displayableJson)
-          case Some(original) => Json.obj("original" -> safeString(original), "pretty" -> displayableJson)
-        }
-      case null => Json.Null
-      case a =>
-        Json.obj(
-          "pretty" -> BestEffortJsonEncoder(failOnUnknown = false, a.getClass.getClassLoader).circeEncoder.apply(a)
-        )
-    }
+  implicit val testResultsEncoder: Encoder[TestResults[Json]] = new Encoder[TestResults[Json]]() {
+
+    implicit val nodeResult: Encoder[NodeResult[Json]]                                 = deriveConfiguredEncoder
+    implicit val expressionInvocationResult: Encoder[ExpressionInvocationResult[Json]] = deriveConfiguredEncoder
+    implicit val externalInvocationResult: Encoder[ExternalInvocationResult[Json]]     = deriveConfiguredEncoder
+    implicit val componentInfo: Encoder[ComponentInfo]                                 = deriveConfiguredEncoder
+    implicit val nodeComponentInfo: Encoder[NodeComponentInfo]                         = deriveConfiguredEncoder
+    implicit val resultContext: Encoder[ResultContext[Json]]                           = deriveConfiguredEncoder
+
+    implicit val mapAnyEncoder: Encoder[Map[String, Any]] = (value: Map[String, Any]) =>
+      value.map { case (key, value) => key -> testResultsVariableEncoder(value) }.asJson
+
+    implicit val throwableEncoder: Encoder[Throwable] = Encoder[Option[String]].contramap(th => Option(th.getMessage))
 
     // TODO: do we want more information here?
     implicit val contextEncoder: Encoder[Context] = (a: Context) =>
@@ -69,30 +81,6 @@ object ManagementResources {
         "variables" -> a.variables.asJson
       )
 
-    implicit val nodeResultEncoder: Encoder[NodeResult]               = deriveConfiguredEncoder
-    implicit val componentInfoEncoder: Encoder[ComponentInfo]         = deriveConfiguredEncoder
-    implicit val nodeComponentInfoEncoder: Encoder[NodeComponentInfo] = deriveConfiguredEncoder
-
-    val throwableEncoder: Encoder[Throwable] = Encoder[Option[String]].contramap(th => Option(th.getMessage))
-
-    // FIXME: It has to be done manually, some reason deriveConfiguredEncoder doesn't work properly with value: Any
-    implicit val externalInvocationResultEncoder: Encoder[ExternalInvocationResult] =
-      (value: ExternalInvocationResult) =>
-        Json.obj(
-          "name"      -> Json.fromString(value.name),
-          "contextId" -> Json.fromString(value.contextId),
-          "value"     -> value.value.asJson,
-        )
-
-    // FIXME: It has to be done manually, some reason deriveConfiguredEncoder doesn't work properly with value: Any
-    implicit val expressionInvocationResultEncoder: Encoder[ExpressionInvocationResult] =
-      (value: ExpressionInvocationResult) =>
-        Json.obj(
-          "name"      -> Json.fromString(value.name),
-          "contextId" -> Json.fromString(value.contextId),
-          "value"     -> value.value.asJson,
-        )
-
     implicit val exceptionsEncoder: Encoder[NuExceptionInfo[_ <: Throwable]] =
       (value: NuExceptionInfo[_ <: Throwable]) =>
         Json.obj(
@@ -101,8 +89,8 @@ object ManagementResources {
           "context"           -> value.context.asJson
         )
 
-    override def apply(a: TestResults): Json = a match {
-      case TestResults(nodeResults, invocationResults, externalInvocationResults, exceptions) =>
+    override def apply(a: TestResults[Json]): Json = a match {
+      case TestResults(nodeResults, invocationResults, externalInvocationResults, exceptions, _) =>
         Json.obj(
           "nodeResults"       -> nodeResults.map { case (node, list) => node -> list.sortBy(_.context.id) }.asJson,
           "invocationResults" -> invocationResults.map { case (node, list) => node -> list.sortBy(_.contextId) }.asJson,
@@ -236,7 +224,8 @@ class ManagementResources(
                             .performTest(
                               idWithName,
                               displayableProcess,
-                              RawScenarioTestData(testDataContent)
+                              RawScenarioTestData(testDataContent),
+                              testResultsVariableEncoder
                             )
                             .flatMap { results =>
                               Marshal(results).to[MessageEntity].map(en => HttpResponse(entity = en))
@@ -265,7 +254,8 @@ class ManagementResources(
                                 .performTest(
                                   idWithName,
                                   displayableProcess,
-                                  rawScenarioTestData
+                                  rawScenarioTestData,
+                                  testResultsVariableEncoder
                                 )
                                 .flatMap { results =>
                                   Marshal(results).to[MessageEntity].map(en => HttpResponse(entity = en))
@@ -294,7 +284,8 @@ class ManagementResources(
                             .performTest(
                               idWithName,
                               testParametersRequest.displayableProcess,
-                              testParametersRequest.sourceParameters
+                              testParametersRequest.sourceParameters,
+                              testResultsVariableEncoder
                             )
                             .flatMap { results =>
                               Marshal(results).to[MessageEntity].map(en => HttpResponse(entity = en))

diff --git a/...rc/main/scala/pl/touk/nussknacker/ui/process/deployment/ScenarioTestExecutorService.scala b/...rc/main/scala/pl/touk/nussknacker/ui/process/deployment/ScenarioTestExecutorService.scala
@@ -11,30 +11,32 @@ import scala.concurrent.{ExecutionContext, Future}
 
 trait ScenarioTestExecutorService {
 
-  def testProcess(
+  def testProcess[T](
       id: ProcessIdWithName,
       canonicalProcess: CanonicalProcess,
       category: String,
       processingType: ProcessingType,
       scenarioTestData: ScenarioTestData,
-  )(implicit loggedUser: LoggedUser, ec: ExecutionContext): Future[TestResults]
+      variableEncoder: Any => T
+  )(implicit loggedUser: LoggedUser, ec: ExecutionContext): Future[TestResults[T]]
 
 }
 
 class ScenarioTestExecutorServiceImpl(scenarioResolver: ScenarioResolver, dispatcher: DeploymentManagerDispatcher)
     extends ScenarioTestExecutorService {
 
-  override def testProcess(
+  override def testProcess[T](
       id: ProcessIdWithName,
       canonicalProcess: CanonicalProcess,
       category: String,
       processingType: ProcessingType,
-      scenarioTestData: ScenarioTestData
-  )(implicit loggedUser: LoggedUser, ec: ExecutionContext): Future[TestResults] = {
+      scenarioTestData: ScenarioTestData,
+      variableEncoder: Any => T
+  )(implicit loggedUser: LoggedUser, ec: ExecutionContext): Future[TestResults[T]] = {
     for {
       resolvedProcess <- Future.fromTry(scenarioResolver.resolveScenario(canonicalProcess, category))
       manager = dispatcher.deploymentManagerUnsafe(processingType)
-      testResult <- manager.test(id.name, resolvedProcess, scenarioTestData)
+      testResult <- manager.test[T](id.name, resolvedProcess, scenarioTestData, variableEncoder)
     } yield testResult
   }
 

diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/test/ResultsWithCounts.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/test/ResultsWithCounts.scala
@@ -3,4 +3,4 @@ package pl.touk.nussknacker.ui.process.test
 import pl.touk.nussknacker.engine.testmode.TestProcess.TestResults
 import pl.touk.nussknacker.ui.processreport.NodeCount
 
-final case class ResultsWithCounts(results: TestResults, counts: Map[String, NodeCount])
+final case class ResultsWithCounts[T](results: TestResults[T], counts: Map[String, NodeCount])
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/test/ScenarioTestService.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/test/ScenarioTestService.scala
@@ -85,8 +85,9 @@ class ScenarioTestService(
   def performTest[T](
       idWithName: ProcessIdWithName,
       displayableProcess: DisplayableProcess,
-      rawTestData: RawScenarioTestData
-  )(implicit ec: ExecutionContext, user: LoggedUser): Future[ResultsWithCounts] = {
+      rawTestData: RawScenarioTestData,
+      testResultsVariableEncoder: Any => T
+  )(implicit ec: ExecutionContext, user: LoggedUser): Future[ResultsWithCounts[T]] = {
     val testInfoProvider = testInfoProviders.forTypeUnsafe(displayableProcess.processingType)
     for {
       preliminaryScenarioTestData <- preliminaryScenarioTestDataSerDe
@@ -101,27 +102,28 @@ class ScenarioTestService(
         canonical,
         displayableProcess.category,
         displayableProcess.processingType,
-        scenarioTestData
+        scenarioTestData,
+        testResultsVariableEncoder
       )
-      _ <- {
-        assertTestResultsAreNotTooBig(testResults)
-      }
+      _ <- assertTestResultsAreNotTooBig(testResults)
     } yield ResultsWithCounts(testResults, computeCounts(canonical, testResults))
   }
 
   def performTest[T](
       idWithName: ProcessIdWithName,
       displayableProcess: DisplayableProcess,
-      parameterTestData: TestSourceParameters
-  )(implicit ec: ExecutionContext, user: LoggedUser): Future[ResultsWithCounts] = {
+      parameterTestData: TestSourceParameters,
+      testResultsVariableEncoder: Any => T
+  )(implicit ec: ExecutionContext, user: LoggedUser): Future[ResultsWithCounts[T]] = {
     val canonical = toCanonicalProcess(displayableProcess)
     for {
       testResults <- testExecutorService.testProcess(
         idWithName,
         canonical,
         displayableProcess.category,
         displayableProcess.processingType,
-        ScenarioTestData(parameterTestData.sourceId, parameterTestData.parameterExpressions)
+        ScenarioTestData(parameterTestData.sourceId, parameterTestData.parameterExpressions),
+        testResultsVariableEncoder
       )
       _ <- assertTestResultsAreNotTooBig(testResults)
     } yield ResultsWithCounts(testResults, computeCounts(canonical, testResults))
@@ -132,7 +134,7 @@ class ScenarioTestService(
     processResolving.resolveExpressions(displayableProcess, validationResult.typingInfo)
   }
 
-  private def assertTestResultsAreNotTooBig(testResults: TestResults): Future[Unit] = {
+  private def assertTestResultsAreNotTooBig(testResults: TestResults[_]): Future[Unit] = {
     val testDataResultApproxByteSize = RamUsageEstimator.sizeOf(testResults)
     if (testDataResultApproxByteSize > testDataSettings.resultsMaxBytes) {
       logger.info(
@@ -144,7 +146,7 @@ class ScenarioTestService(
     }
   }
 
-  private def computeCounts(canonical: CanonicalProcess, results: TestResults): Map[String, NodeCount] = {
+  private def computeCounts(canonical: CanonicalProcess, results: TestResults[_]): Map[String, NodeCount] = {
     val counts = results.nodeResults.map { case (key, nresults) =>
       key -> RawCount(
         nresults.size.toLong,

diff --git a/.../scala/pl/touk/nussknacker/development/manager/DevelopmentDeploymentManagerProvider.scala b/.../scala/pl/touk/nussknacker/development/manager/DevelopmentDeploymentManagerProvider.scala
@@ -152,11 +152,12 @@ class DevelopmentDeploymentManager(actorSystem: ActorSystem)
     Future.unit
   }
 
-  override def test(
+  override def test[T](
       name: ProcessName,
       canonicalProcess: CanonicalProcess,
-      scenarioTestData: ScenarioTestData
-  ): Future[TestProcess.TestResults] = ???
+      scenarioTestData: ScenarioTestData,
+      variableEncoder: Any => T
+  ): Future[TestProcess.TestResults[T]] = ???
 
   override protected def getFreshProcessStates(name: ProcessName): Future[List[StatusDetails]] =
     Future.successful(memory.get(name).toList)