(texera) 01/01: test(amber): cover email notification dispatch (#7191)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7191-01071832cd2e64d45b54f46ffd193d1ffa359299
in repository https://gitbox.apache.org/repos/asf/texera.git

commit 91a036bf4f78d862b5f63b0360bcb5d769b6d2bb
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Fri Jul 31 22:33:02 2026 -0700

    test(amber): cover email notification dispatch (#7191)
    
    ### What changes were proposed in this PR?
    
    Adds focused unit coverage for email notification dispatch, suppression,
    and recovery from sender exceptions without requiring external
    infrastructure.
    
    ### Any related issues, documentation, discussions?
    
    Closes #7188
    
    ### How was this PR tested?
    
    - `WorkflowExecutionService/testOnly
    org.apache.texera.web.service.EmailNotificationServiceSpec` — 3 passed.
    - `WorkflowExecutionService/Test/scalafmtAll` and
    `WorkflowExecutionService/Test/scalafix`
    - `WorkflowExecutionService/Test/scalafmtCheck` and
    `WorkflowExecutionService/Test/scalafix --check`
    
    Mutation proof (each production change was reverted):
    
    | Production mutation | Focused test result |
    | --- | --- |
    | Inverted the send predicate | 3 failed |
    | Replaced status-email dispatch with a no-op | 2 failed |
    | Removed exception recovery | 1 failed |
    
    ### Was this PR authored or co-authored using generative AI tooling?
    
    Generated-by: Codex (GPT-5)
---
 .../web/service/EmailNotificationServiceSpec.scala | 99 ++++++++++++++++++++++
 1 file changed, 99 insertions(+)

diff --git a/amber/src/test/scala/org/apache/texera/web/service/EmailNotificationServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/EmailNotificationServiceSpec.scala
new file mode 100644
index 0000000000..5eff5eb325
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/service/EmailNotificationServiceSpec.scala
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.service
+
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import scala.concurrent.Await
+import scala.concurrent.duration.DurationInt
+import scala.jdk.CollectionConverters._
+
+class EmailNotificationServiceSpec extends AnyFlatSpec with Matchers {
+
+  private val completedState = WorkflowAggregatedState.COMPLETED
+  private val awaitTimeout = 2.seconds
+
+  private class RecordingNotifier(
+      shouldSend: Boolean,
+      failureOnSend: Option[RuntimeException] = None
+  ) extends EmailNotifier {
+    val evaluatedStates = new ConcurrentLinkedQueue[WorkflowAggregatedState]()
+    val sentStates = new ConcurrentLinkedQueue[WorkflowAggregatedState]()
+
+    override def shouldSendEmail(workflowState: WorkflowAggregatedState): Boolean = {
+      evaluatedStates.add(workflowState)
+      shouldSend
+    }
+
+    override def sendStatusEmail(state: WorkflowAggregatedState): Unit = {
+      sentStates.add(state)
+      failureOnSend.foreach(throw _)
+    }
+  }
+
+  private def withService(notifier: EmailNotifier)(test: EmailNotificationService => Unit): Unit = {
+    val service = new EmailNotificationService(notifier)
+    try {
+      test(service)
+    } finally {
+      service.shutdown()
+    }
+  }
+
+  "processEmailNotificationIfNeeded" should "send the exact state when the notifier requests it" in {
+    val notifier = new RecordingNotifier(shouldSend = true)
+
+    withService(notifier) { service =>
+      Await.result(service.processEmailNotificationIfNeeded(completedState), awaitTimeout)
+
+      notifier.evaluatedStates.asScala.toList shouldBe List(completedState)
+      notifier.sentStates.asScala.toList shouldBe List(completedState)
+    }
+  }
+
+  it should "not send an email when the notifier declines it" in {
+    val notifier = new RecordingNotifier(shouldSend = false)
+
+    withService(notifier) { service =>
+      Await.result(service.processEmailNotificationIfNeeded(completedState), awaitTimeout)
+
+      notifier.evaluatedStates.asScala.toList shouldBe List(completedState)
+      notifier.sentStates.asScala.toList shouldBe empty
+    }
+  }
+
+  it should "recover from an exception raised while sending an email" in {
+    val notifier = new RecordingNotifier(
+      shouldSend = true,
+      failureOnSend = Some(new RuntimeException("email provider unavailable"))
+    )
+
+    withService(notifier) { service =>
+      noException should be thrownBy {
+        Await.result(service.processEmailNotificationIfNeeded(completedState), awaitTimeout)
+      }
+
+      notifier.sentStates.asScala.toList shouldBe List(completedState)
+    }
+  }
+}