如何确保给定的未来在测试中首先完成？

Question

我正在为函数 bar 编写测试：

def bar(fut1: Future[Int], 
        fut2: Future[Int], 
        fut3: Future[Int]): Future[Result] = ???

bar 像这样返回Result：

 case class Result(
    x: Int,          // fut1 value
    oy: Option[Int], // if fut2 is complete then Some of fut2 value else None 
    oz: Option[Int]  // if fut3 is complete then Some of fut3 value else None 
 ) 

我想为所有测试用例编写测试：

• fut1 已完成，fut2 和 fut3 未完成
• fut1 完成，fut2 完成，fut3 未完成
• 等

所以我正在为这些测试编写函数foo1、foo2 和foo3 的假 实现。

def foo1(x: Int): Future[Int] = ??? 
def foo2(x: Int): Future[Int] = ??? 
def foo3(x: Int): Future[Int] = ??? 

Test #1 调用所有这些函数，检查 fut1 是否先完成，然后调用 bar

val fut1 = foo1(0)
val fut2 = foo2(0)
val fut3 = foo3(0)

// make sure `fut1` completes first

测试#2 调用所有这些函数，确保fut2 首先完成，然后调用bar。
测试#3 调用所有这些函数，确保fut3 首先完成，然后调用bar。

我的问题是如何实现函数foo1、foo2和foo3以及测试。

Answer 1

您可以尝试通过map 将完整性时间戳附加到每个未来，例如：

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent._
import scala.concurrent.duration._
import scala.language.postfixOps

def foo1(x: Int): Future[Int] = Future {Thread.sleep(200); 1} 
def foo2(x: Int): Future[Int] = Future {Thread.sleep(500); 2} 
def foo3(x: Int): Future[Int] = Future {Thread.sleep(500); 3}

def completeTs[T](future: Future[T]): Future[(T, Long)] = future.map(v => v -> System.currentTimeMillis())

val resutls = Await.result(Future.sequence(
  List(completeTs(foo1(1)), completeTs(foo2(1)), completeTs(foo3(1)))
), 2 seconds)

val firstCompleteTs = resutls.map(_._2).min
val firstCompleteIndex = resutls.indexWhere(_._2 == firstCompleteTs)
assert(firstCompleteIndex == 0)

斯卡蒂：https://scastie.scala-lang.org/L9g78DSNQIm2K1jGlQzXBg

Answer 2

您可以重新利用firstCompletedOf 来验证期货列表中给定指数的期货是否是第一个完成的指数：

import java.util.concurrent.atomic.AtomicReference
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.Try

def isFirstCompleted[T](idx: Int)(futures: List[Future[T]])(
    implicit ec: ExecutionContext): Future[Boolean] = {
  val promise = Promise[(T, Int)]()
  val pRef = new AtomicReference[Promise[(T, Int)]](promise)
  futures.zipWithIndex foreach { case (f, i) => f onComplete { case tt: Try[T] =>
      val p = pRef.getAndSet(null)
      if (p != null) p tryComplete tt.map((_, i))
    }
  }
  promise.future.map{ case (t, i) => i == idx }
}

试运行：

import scala.concurrent.ExecutionContext.Implicits.global

val futures = List(
  Future{Thread.sleep(100); 1},
  Future{Thread.sleep(150); throw new Exception("oops!")},
  Future{Thread.sleep(50); 3}
)

isFirstCompleted(0)(futures)  // Future(Success(false))
isFirstCompleted(2)(futures)  // Future(Success(true))

要编写测试用例，请考虑使用ScalaTest AsyncFlatSpec。

Answer 3

目前尚不清楚您要测试的究竟是什么。 如果你只是使用已经完成的期货，你会得到你描述的行为：

def f1 = Future.successful(1)
def f2 = Future.successful(2)
def f3 = Future.successful(3)

eventually {
   Future.firstCompletedOf(Seq(f1, f2, f3)).value shouldBe Some(1)
}

（请注意，您不能像您在问题中所做的那样直接与fut1 进行比较，这总是错误的，因为.firstCompletedOf 返回一个新的未来）。

你也可以只完成一个未来，而不要管其他的：

    val promise = Promise[Int].future
    def f1 = promise.future // or just Future.successful(1) ... or Future(1)
    def f2 = Future.never
    def f3 = Future.never
  
    result =  Future.firstCompletedOf(Seq(f1, f2, f3))
    promise.complete(Success(1)) 
    eventually {
       result.value shouldBe 1
    }

等等......也可以让其他期货也得到他们自己的承诺的支持，例如，如果你希望它们最终都完成（不确定它会给你带来什么，但话又说回来，我不确定你正在这里开始测试）。

另一种可能是让它们相互依赖：

    val promise = Promise[Int]
    def f1 = promise.future
    def f2 = promise.future.map(_ + 1)
    def f3 = promise.future.map(_ + 2)
    
    ... 
    promise.complete(Success(1))