Feature/348 create checkpoint v2

agent/src/main/scala/za/co/absa/atum/agent/dispatcher/ConsoleDispatcher.scala

@@ -18,15 +18,7 @@ package za.co.absa.atum.agent.dispatcher
 
 import com.typesafe.config.Config
 import org.apache.spark.internal.Logging
-import za.co.absa.atum.model.dto.{
-  AdditionalDataDTO,
-  AdditionalDataItemDTO,
-  AdditionalDataPatchDTO,
-  AtumContextDTO,
-  CheckpointDTO,
-  PartitioningDTO,
-  PartitioningSubmitDTO
-}
+import za.co.absa.atum.model.dto._
 
 /**
  *  dispatcher useful for development, testing and debugging

agent/src/main/scala/za/co/absa/atum/agent/dispatcher/Dispatcher.scala

@@ -17,14 +17,7 @@
 package za.co.absa.atum.agent.dispatcher
 
 import com.typesafe.config.Config
-import za.co.absa.atum.model.dto.{
-  AdditionalDataDTO,
-  AdditionalDataPatchDTO,
-  AtumContextDTO,
-  CheckpointDTO,
-  PartitioningDTO,
-  PartitioningSubmitDTO
-}
+import za.co.absa.atum.model.dto._
 
 /**
  *  This class provides a contract for different dispatchers. It has a constructor foe eventual creation via reflection.

agent/src/main/scala/za/co/absa/atum/agent/dispatcher/HttpDispatcher.scala

@@ -22,6 +22,7 @@ import sttp.capabilities
 import sttp.client3._
 import sttp.model.Uri
 import sttp.client3.okhttp.OkHttpSyncBackend
+import sttp.model.StatusCode
 import za.co.absa.atum.agent.exception.AtumAgentException.HttpException
 import za.co.absa.atum.model.ApiPaths
 import za.co.absa.atum.model.dto._
@@ -33,7 +34,6 @@ class HttpDispatcher(config: Config) extends Dispatcher(config) with Logging {
 
   private val serverUrl: String = config.getString(UrlKey)
 
-  private val apiV1 = s"/${ApiPaths.Api}/${ApiPaths.V1}"
   private val apiV2 = s"/${ApiPaths.Api}/${ApiPaths.V2}"
 
   private val getPartitioningIdEndpoint = Uri.unsafeParse(s"$serverUrl$apiV2/${ApiPaths.V2Paths.Partitionings}")
@@ -64,8 +64,8 @@ class HttpDispatcher(config: Config) extends Dispatcher(config) with Logging {
 
     val response = backend.send(request)
 
-    response.code.code match {
-      case 404 => None
+    response.code match {
+      case StatusCode.NotFound => None
       case _ => Some(handleResponseBody(response).as[SingleSuccessResponse[PartitioningWithIdDTO]].data)
     }
   }
@@ -118,13 +118,27 @@ class HttpDispatcher(config: Config) extends Dispatcher(config) with Logging {
   }
 
   override protected[agent] def saveCheckpoint(checkpoint: CheckpointDTO): Unit = {
-    val endpoint = Uri.unsafeParse(s"$serverUrl$apiV1/${ApiPaths.V1Paths.CreateCheckpoint}")
+    val partitioningId = getPartitioningId(checkpoint.partitioning)
+
+    val checkpointV2DTO = CheckpointV2DTO(
+      id = checkpoint.id,
+      name = checkpoint.name,
+      author = checkpoint.author,
+      measuredByAtumAgent = checkpoint.measuredByAtumAgent,
+      processStartTime = checkpoint.processStartTime,
+      processEndTime = checkpoint.processEndTime,
+      measurements = checkpoint.measurements,
+      properties = checkpoint.properties
+    )
+
+    val endpoint = Uri.unsafeParse(
+      s"$serverUrl$apiV2/${ApiPaths.V2Paths.Partitionings}/$partitioningId/${ApiPaths.V2Paths.Checkpoints}"
+    )
     val request = commonAtumRequest
       .post(endpoint)
-      .body(checkpoint.asJsonString)
+      .body(checkpointV2DTO.asJsonString)
 
     val response = backend.send(request)
-
     handleResponseBody(response)
   }
 

agent/src/test/scala/za/co/absa/atum/agent/dispatcher/HttpDispatcherUnitTests.scala

@@ -9,10 +9,13 @@ import sttp.client3._
 import sttp.model.StatusCode
 import com.typesafe.config.Config
 import sttp.capabilities
+import za.co.absa.atum.agent.exception.AtumAgentException.HttpException
 import za.co.absa.atum.model.dto._
 import za.co.absa.atum.model.envelopes.SuccessResponse.{MultiSuccessResponse, SingleSuccessResponse}
 import za.co.absa.atum.model.utils.JsonSyntaxExtensions.JsonSerializationSyntax
 
+import java.time.ZonedDateTime
+
 class HttpDispatcherUnitTests extends AnyFlatSpec with Matchers with BeforeAndAfterEach {
 
   var mockBackend: SttpBackend[Identity, sttp.capabilities.WebSockets] = _
@@ -23,7 +26,7 @@ class HttpDispatcherUnitTests extends AnyFlatSpec with Matchers with BeforeAndAf
   val testPartitioningSubmitDTO = PartitioningSubmitDTO(testPartitioningDTO, None, "author")
   val createdPartitioningWithId = PartitioningWithIdDTO(123L, testPartitioningDTO, "author")
   val measures: Seq[MeasureDTO] = Seq(MeasureDTO("m1", Seq("c1")), MeasureDTO("m2", Seq("c2")))
-  val additionalData: Map[String, Option[AdditionalDataItemDTO]] = Map(
+  val additionalData = Map(
     "key1" -> Some(AdditionalDataItemDTO("val1", "author1")),
     "key2" -> None
   )
@@ -177,4 +180,83 @@ class HttpDispatcherUnitTests extends AnyFlatSpec with Matchers with BeforeAndAf
     result.measures shouldBe empty
     result.additionalData shouldBe empty
   }
+
+  "saveCheckpoint" should "successfully save a checkpoint when partitioning exists" in {
+    val checkpoint = CheckpointDTO(
+      id = java.util.UUID.randomUUID(),
+      name = "cp",
+      author = "author",
+      measuredByAtumAgent = true,
+      processStartTime = ZonedDateTime.now(),
+      processEndTime = Some(ZonedDateTime.now()),
+      measurements = Set.empty,
+      partitioning = testPartitioningDTO
+    )
+    val partitioningWithId = PartitioningWithIdDTO(123L, testPartitioningDTO, "author")
+    val getPartitioningResponse = Response(
+      Right(SingleSuccessResponse(partitioningWithId).asJsonString): Either[String, String],
+      StatusCode.Ok
+    )
+    val postCheckpointResponse = Response(Right("{}"): Either[String, String], StatusCode.Created)
+
+    stubGetPartitioning(testPartitioningDTO, getPartitioningResponse)
+    when(
+      mockBackend.send(
+        argThat[Request[Either[String, String], capabilities.WebSockets]](
+          req => req.method.method == "POST" && req.uri.path.mkString.contains("checkpoints")
+        )
+      )
+    ).thenReturn(postCheckpointResponse)
+
+    noException should be thrownBy dispatcher.saveCheckpoint(checkpoint)
+  }
+
+  it should "throw HttpException if partitioning is not found" in {
+    val checkpoint = CheckpointDTO(
+      id = java.util.UUID.randomUUID(),
+      name = "cp",
+      author = "author",
+      measuredByAtumAgent = true,
+      processStartTime = ZonedDateTime.now(),
+      processEndTime = Some(ZonedDateTime.now()),
+      measurements = Set.empty,
+      partitioning = testPartitioningDTO
+    )
+    val notFoundResponse = Response(Left("Not found"): Either[String, String], StatusCode.NotFound)
+
+    stubGetPartitioning(testPartitioningDTO, notFoundResponse)
+
+    an[HttpException] should be thrownBy dispatcher.saveCheckpoint(checkpoint)
+  }
+
+  it should "throw HttpException if server returns error on checkpoint save" in {
+    val checkpoint = CheckpointDTO(
+      id = java.util.UUID.randomUUID(),
+      name = "cp",
+      author = "author",
+      measuredByAtumAgent = true,
+      processStartTime = ZonedDateTime.now(),
+      processEndTime = Some(ZonedDateTime.now()),
+      measurements = Set.empty,
+      partitioning = testPartitioningDTO
+    )
+    val partitioningWithId = PartitioningWithIdDTO(123L, testPartitioningDTO, "author")
+    val getPartitioningResponse = Response(
+      Right(SingleSuccessResponse(partitioningWithId).asJsonString): Either[String, String],
+      StatusCode.Ok
+    )
+    val badRequestResponse = Response(Left("Bad request"): Either[String, String], StatusCode.BadRequest)
+
+    stubGetPartitioning(testPartitioningDTO, getPartitioningResponse)
+    when(
+      mockBackend.send(
+        argThat[Request[Either[String, String], capabilities.WebSockets]](
+          req => req.method.method == "POST" && req.uri.path.mkString.contains("checkpoints")
+        )
+      )
+    ).thenReturn(badRequestResponse)
+
+    an[HttpException] should be thrownBy dispatcher.saveCheckpoint(checkpoint)
+  }
+
 }