#531 Notify users when some of the emails are not correct or not allo…

…wed.

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala

@@ -47,4 +47,6 @@ trait PipelineNotificationBuilder {
   def addCustomEntries(entries: Seq[NotificationEntry]): Unit
 
   def addSignature(signature: TextElement*): Unit
+
+  def addValidatedEmails(validatedEmails: ValidatedEmails): Unit
 }

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala

@@ -69,6 +69,7 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
   var isDryRun = false
   var isUndercover = false
   var customSignature = Seq.empty[TextElement]
+  var validatedEmailsOpt: Option[ValidatedEmails] = None
 
   val completedTasks = new ListBuffer[TaskResult]
   val pipelineNotificationFailures = new ListBuffer[PipelineNotificationFailure]
@@ -124,6 +125,8 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
 
   override def addSignature(signature: TextElement*): Unit = customSignature = signature
 
+  override def addValidatedEmails(validatedEmailsIn: ValidatedEmails): Unit = validatedEmailsOpt = Option(validatedEmailsIn)
+
   def renderSubject(): String = {
     val timeCreatedStr = ZonedDateTime.now(zoneId).format(timestampFmt)
 
@@ -142,6 +145,8 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
 
     renderHeader(builder)
 
+    renderValidatedEmails(builder)
+
     renderCompletedTasks(builder)
 
     val allSchemaChanges = completedTasks
@@ -776,6 +781,26 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
     }
   }
 
+  private def renderValidatedEmails(builder: MessageBuilderHtml): Unit = {
+    validatedEmailsOpt.foreach { validatedEmails =>
+      if (validatedEmails.invalidFormatEmails.nonEmpty) {
+        val emailText = TextElement(validatedEmails.invalidFormatEmails.mkString(", "), Style.Error)
+        builder.withParagraph(Seq(
+          TextElement("Warning! ", Style.Error),
+          TextElement(s"Some email recipients are not proper emails: ", Style.Exception)
+        ) :+ emailText)
+      }
+
+      if (validatedEmails.invalidDomainEmails.nonEmpty) {
+        val emailText = TextElement(validatedEmails.invalidDomainEmails.mkString(", "), Style.Error)
+        builder.withParagraph(Seq(
+          TextElement("Warning! ", Style.Error),
+          TextElement(s"Some email recipients have domain names that are not allowed: ", Style.Exception)
+        ) :+ emailText)
+      }
+    }
+  }
+
   private[core] def getTransientTextStyle(task: TaskResult): Style = {
     if (task.isTransient)
       Style.Italic

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationDirector.scala

@@ -57,6 +57,7 @@ object PipelineNotificationDirector {
 
     notificationBuilder.addCustomEntries(notification.customEntries)
     notificationBuilder.addSignature(notification.customSignature: _*)
+    notificationBuilder.addValidatedEmails(validatedEmails)
 
     notificationBuilder
   }

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationEmail.scala

@@ -34,10 +34,32 @@ class PipelineNotificationEmail(notification: PipelineNotification)
 
   private val allowedDomains = getDomainList(ConfigUtils.getOptListStrings(conf, Keys.MAIL_ALLOWED_DOMAINS))
 
+  private lazy val validatedEmails = {
+    val validatedEmails = validateRecipientEmails(getEmailRecipients, allowedDomains)
+
+    validatedEmails.invalidFormatEmails.foreach(email =>
+      log.error(s"${Emoji.FAILURE} Invalid email format: $email")
+    )
+
+    validatedEmails.invalidDomainEmails.foreach(email =>
+      log.error(s"${Emoji.FAILURE} Invalid email domain: $email")
+    )
+
+    if (validatedEmails.invalidFormatEmails.nonEmpty || validatedEmails.invalidDomainEmails.nonEmpty) {
+      if (validatedEmails.validEmails.nonEmpty) {
+        log.warn(s"${Emoji.WARNING} Sending the notification to valid emails only: ${validatedEmails.validEmails.mkString(", ")}")
+      } else {
+        log.error(s"${Emoji.FAILURE} No valid emails found. The notification will not be sent.")
+      }
+    }
+
+    validatedEmails
+  }
+
   private lazy val notificationBuilder = {
     val builder = new PipelineNotificationBuilderHtml
 
-    PipelineNotificationDirector.build(builder, notification, validateRecipientEmails(getEmailRecipients, allowedDomains))
+    PipelineNotificationDirector.build(builder, notification, validatedEmails)
     builder
   }
 
@@ -46,16 +68,6 @@ class PipelineNotificationEmail(notification: PipelineNotification)
   override def getFrom: String = conf.getString(Keys.MAIL_FROM)
 
   override def getTo: String = {
-    val validatedEmails = validateRecipientEmails(getEmailRecipients, allowedDomains)
-
-    validatedEmails.invalidFormatEmails.foreach(email =>
-      log.error(s"${Emoji.FAILURE} Invalid email format: $email")
-    )
-
-    validatedEmails.invalidDomainEmails.foreach(email =>
-      log.error(s"${Emoji.FAILURE} Invalid email domain: $email")
-    )
-
     validatedEmails.validEmails.mkString(", ")
   }
 
@@ -113,7 +125,7 @@ object PipelineNotificationEmail {
       }
     }
 
-    ValidatedEmails(validEmails, invalidFormatEmails, invalidDomainEmails)
+    ValidatedEmails(validEmails.toList, invalidFormatEmails.toList, invalidDomainEmails.toList)
   }
 
   private[core] def getDomainList(domains: Seq[String]): Seq[String] = {

pramen/core/src/test/resources/test/notify/test_pipeline_email_body_complex.txt

@@ -106,6 +106,10 @@
 
 <p>Job started at <b>2022-03-21 16:29 +0200</b>, finished at <b>2022-03-21 19:16 +0200</b>. Elapsed time: <b>2 hours and 46 minutes</b>. The job ran in <i>undercover</i> mode - no updates to bookkeeping tables are saved.</p>
 
+<p><span class="tderr">Warning! </span><span class="tdred">Some email recipients are not proper emails: </span><span class="tderr">invalid_email</span></p>
+
+<p><span class="tderr">Warning! </span><span class="tdred">Some email recipients have domain names that are not allowed: </span><span class="tderr">[email protected]</span></p>
+
 <div class="datagrid" style="width:fit-content"><table style="width:100%">
 <thead><tr><th>Job</th>
 <th>Table</th>

pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala

@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.mocks.notify
 
 import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
 import za.co.absa.pramen.api.status.{PipelineNotificationFailure, TaskResult}
-import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilder
+import za.co.absa.pramen.core.notify.pipeline.{PipelineNotificationBuilder, ValidatedEmails}
 
 import java.time.Instant
 
@@ -35,6 +35,7 @@ class PipelineNotificationBuilderSpy extends PipelineNotificationBuilder {
   var minRps = 0
   var goodRps = 0
   var customSignature = Seq.empty[TextElement]
+  var validatedEmailsOpt: Option[ValidatedEmails] = None
 
   var addCompletedTaskCalled = 0
   var addPipelineNotificationFailure = 0
@@ -71,4 +72,6 @@ class PipelineNotificationBuilderSpy extends PipelineNotificationBuilder {
   override def addCustomEntries(entries: Seq[NotificationEntry]): Unit = addCustomEntriesCalled += 1
 
   override def addSignature(signature: TextElement*): Unit = customSignature = signature
+
+  override def addValidatedEmails(validatedEmailsIn: ValidatedEmails): Unit = validatedEmailsOpt = Option(validatedEmailsIn)
 }

pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationBuilderHtmlSuite.scala

@@ -20,13 +20,13 @@ import com.typesafe.config.{Config, ConfigFactory}
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.notification.NotificationEntry.Paragraph
 import za.co.absa.pramen.api.notification._
-import za.co.absa.pramen.api.status.{DependencyWarning, NotificationFailure, PipelineNotificationFailure, RunStatus, TaskRunReason}
+import za.co.absa.pramen.api.status._
 import za.co.absa.pramen.core.exceptions.{CmdFailedException, ProcessFailedException}
 import za.co.absa.pramen.core.fixtures.TextComparisonFixture
 import za.co.absa.pramen.core.mocks.{RunStatusFactory, SchemaDifferenceFactory, TaskResultFactory, TestPrototypes}
 import za.co.absa.pramen.core.notify.message.{MessageBuilderHtml, ParagraphBuilder}
-import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilderHtml
 import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilderHtml.{NOTIFICATION_EXCEPTION_MAX_LENGTH_KEY, NOTIFICATION_REASON_MAX_LENGTH_KEY}
+import za.co.absa.pramen.core.notify.pipeline.{PipelineNotificationBuilderHtml, ValidatedEmails}
 import za.co.absa.pramen.core.utils.ResourceUtils
 
 import java.time.{Instant, LocalDate}
@@ -146,6 +146,8 @@ class PipelineNotificationBuilderHtmlSuite extends AnyWordSpec with TextComparis
 
       builder.addSignature(TextElement("Test signature"))
 
+      builder.addValidatedEmails(ValidatedEmails(Seq("[email protected]"), Seq("invalid_email"), Seq("[email protected]")))
+
       val actual = builder.renderBody()
 
       compareText(actual, expected)

pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala

@@ -71,6 +71,7 @@ class PipelineNotificationDirectorSuite extends AnyWordSpec {
       assert(builderSpy.addCompletedTaskCalled == 3)
       assert(builderSpy.addCustomEntriesCalled == 1)
       assert(builderSpy.addPipelineNotificationFailure == 1)
+      assert(builderSpy.validatedEmailsOpt.isDefined)
     }
   }