#423 Add pipeline info to be available from Pramen object.

AbsaOSS · Jun 17, 2024 · b2b50ce · b2b50ce
1 parent e7446d8
commit b2b50ce
Show file tree

Hide file tree

Showing 4 changed files with 49 additions and 2 deletions.
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala
@@ -36,6 +36,9 @@ trait Pramen {
   /** This gives access to the current workflow configuration. */
   def workflowConfig: Config
 
+  /** General information about the running pipeline. */
+  def pipelineInfo: PipelineInfo
+
   /** Gets the notification builder that you can use to add custom information to email notifications. */
   def notificationBuilder: NotificationBuilder
 

diff --git a/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala b/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala
@@ -18,13 +18,15 @@ package za.co.absa.pramen.api.mocks
 
 import com.typesafe.config.Config
 import za.co.absa.pramen.api.common.BuildPropertiesRetriever
-import za.co.absa.pramen.api.{MetadataManager, NotificationBuilder, Pramen, TaskNotification}
+import za.co.absa.pramen.api.{MetadataManager, NotificationBuilder, PipelineInfo, Pramen, TaskNotification}
 
 class DummyPramen extends Pramen {
   override def buildProperties: BuildPropertiesRetriever = null
 
   override def workflowConfig: Config = null
 
+  override def pipelineInfo: PipelineInfo = null
+
   override def notificationBuilder: NotificationBuilder = null
 
   override def metadataManager: MetadataManager = null

diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala
@@ -19,7 +19,7 @@ package za.co.absa.pramen.core
 import com.typesafe.config.Config
 import za.co.absa.pramen.api.app.PramenFactory
 import za.co.absa.pramen.api.common.BuildPropertiesRetriever
-import za.co.absa.pramen.api.{MetadataManager, NotificationBuilder, Pramen, TaskNotification}
+import za.co.absa.pramen.api.{MetadataManager, NotificationBuilder, PipelineInfo, Pramen, TaskNotification}
 import za.co.absa.pramen.core.state.{NotificationBuilderImpl, PipelineState, PipelineStateImpl, PipelineStateSnapshot}
 import za.co.absa.pramen.core.utils.BuildPropertyUtils
 
@@ -38,6 +38,14 @@ class PramenImpl extends Pramen {
     throw new IllegalStateException("Workflow configuration is not available at the context.")
   )
 
+  override def pipelineInfo: PipelineInfo = {
+    val pipelineState = _pipelineState.getOrElse(
+      throw new IllegalStateException("Pipeline state is not available at the context.")
+    )
+
+    pipelineState.getState().pipelineInfo
+  }
+
   override def notificationBuilder: NotificationBuilder = notificationBuilderImpl
 
   override def metadataManager: MetadataManager = _metadataManager.getOrElse(

diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/PramenImplSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/PramenImplSuite.scala
@@ -108,6 +108,40 @@ class PramenImplSuite extends AnyWordSpec {
     }
   }
 
+  "pipelineInfo" should {
+    "return pipeline info when it is available" in {
+      val pramen = PramenImpl.instance.asInstanceOf[PramenImpl]
+
+      val taskResults = Seq(
+        TaskResultFactory.getDummyTaskResult(),
+        TaskResultFactory.getDummyTaskResult(runInfo = None),
+        TaskResultFactory.getDummyTaskResult(runStatus = RunStatus.NotRan)
+      )
+
+      val pipelineState = mock(classOf[PipelineState])
+      when(pipelineState.getState()).thenReturn(PipelineStateSnapshotFactory.getDummyPipelineStateSnapshot(taskResults = taskResults))
+
+      pramen.setPipelineState(pipelineState)
+
+      val pipelineInfo = PramenImpl.instance.pipelineInfo
+
+      assert(pipelineInfo.pipelineName == "Dummy Pipeline")
+      assert(pipelineInfo.environment == "DEV")
+
+      pramen.setPipelineState(null)
+    }
+
+    "throw an exception if pipeline state is not available" in {
+      val pramen = PramenImpl.instance.asInstanceOf[PramenImpl]
+
+      pramen.setPipelineState(null)
+
+      assertThrows[IllegalStateException] {
+        PramenImpl.instance.pipelineInfo
+      }
+    }
+  }
+
   "getCompletedTasks()" should {
     "return the metadata manager if it is available" in {
       val pramen = PramenImpl.instance.asInstanceOf[PramenImpl]