单元测试因 Observable.FromAsync 和 Observable.Switch 而失败

Question

我在测试使用Observable.FromAsync<T>() 和Observable.Switch<T>() 的类时遇到了麻烦。它所做的是等待一个可观察的触发器产生一个值，然后它开始一个异步操作，最后在一个输出序列中重新收集所有操作的结果。它的要点是这样的：

var outputStream = triggerStream
  .Select(_ => Observable
    .FromAsync(token => taskProducer.DoSomethingAsync(token)))
  .Switch();

我用最少的部分进行了一些健全性检查测试，以了解发生了什么，这是 cmets 中的测试结果：

class test_with_rx : nspec
{
  void Given_async_task_and_switch()
  {
    Subject<Unit> triggerStream = null;
    TaskCompletionSource<long> taskDriver = null;
    ITestableObserver<long> testObserver = null;
    IDisposable subscription = null;

    before = () =>
    {
      TestScheduler scheduler = new TestScheduler();
      testObserver = scheduler.CreateObserver<long>();
      triggerStream = new Subject<Unit>();
      taskDriver = new TaskCompletionSource<long>();

      // build stream under test
      IObservable<long> streamUnderTest = triggerStream
        .Select(_ => Observable
          .FromAsync(token => taskDriver.Task))
        .Switch();

      /* Also tried with this Switch() overload
      IObservable<long> streamUnderTest = triggerStream
          .Select(_ => taskDriver.Task)
          .Switch(); */

      subscription = streamUnderTest.Subscribe(testObserver);
    };

    context["Before trigger"] = () =>
    {
      it["Should not notify"] = () => testObserver.Messages.Count.Should().Be(0);
      // PASSED
    };

    context["After trigger"] = () =>
    {
      before = () => triggerStream.OnNext(Unit.Default);

      context["When task completes"] = () =>
      {
        long result = -1;

        before = () =>
        {
          taskDriver.SetResult(result);
          //taskDriver.Task.Wait();  // tried with this too
        };

        it["Should notify once"] = () => testObserver.Messages.Count.Should().Be(1);
        // FAILED: expected 1, actual 0

        it["Should notify task result"] = () => testObserver.Messages[0].Value.Value.Should().Be(result);
        // FAILED: of course, index out of bound
      };
    };

    after = () =>
    {
      taskDriver.TrySetCanceled();
      taskDriver.Task.Dispose();
      subscription.Dispose();
    };
  }
}

在我也用模拟完成的其他测试中，我可以看到传递给 FromAsync 的 Func 实际上被调用了（例如taskProducer.DoSomethingAsync(token)），但是看起来没有更多的后续，并且输出流不会产生价值。

我还尝试在达到预期之前插入一些 Task.Delay(x).Wait() 或一些 taskDriver.Task.Wait()，但没有成功。

我阅读了this SO thread 并且知道调度程序，但乍一看我以为我不需要它们，没有使用ObserveOn()。我错了吗？我错过了什么？助教

为了完整性，测试框架是 NSpec，断言库是 FluentAssertions。

Answer 1

您遇到的是一起测试 Rx 和 TPL 的案例。 可以在 here 找到详尽的解释，但我会尝试为您的特定代码提供建议。

基本上您的代码运行良好，但您的测试却不行。 Observable.FromAsync 将在提供的任务上转换为 ContinueWith，这将在任务池上执行，因此是异步的。

修复您的测试的许多方法：（从丑陋到复杂）

1. 结果集后休眠（注意等待不起作用，因为等待不等待继续）

   taskDriver.SetResult(result);
   Thread.Sleep(50);
   

2. 在执行 FromAsync 之前设置结果（因为如果任务完成，FromAsync 将立即返回一个 IObservable，也就是将跳过 ContinueWith）

   taskDriver.SetResult(result);
   triggerStream.OnNext(Unit.Default);
   

3. 用可测试的替代方法替换 FromAsync，例如

   public static IObservable<T> ToObservable<T>(Task<T> task, TaskScheduler scheduler)
   {
       if (task.IsCompleted)
       {
           return task.ToObservable();
       }
       else
       {
           AsyncSubject<T> asyncSubject = new AsyncSubject<T>();
           task.ContinueWith(t => task.ToObservable().Subscribe(asyncSubject), scheduler);
           return asyncSubject.AsObservable<T>();
       }
   }
   

（使用synchronous TaskScheduler 或testable one）