c# - Observable.FromAsync 和 Observable.Switch 的单元测试失败

Question

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

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

我用最少的部分进行了一些健全性检查测试，以了解发生了什么，下面是评论中的测试结果:

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) )，但随后看起来没有更多内容，并且输出流不产生值(value)。

我也试过插入一些 Task.Delay(x).Wait() , 或者一些 taskDriver.Task.Wait()在达到预期之前，但没有运气。

我读了this SO thread我知道调度程序，但乍一看我认为我不需要它们，不需要 ObserveOn()正在被使用。我错了吗？我错过了什么？助教

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

Answer 1

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

基本上您的代码可以正常工作，但您的测试却不行。Observable.FromAsync 将在提供的任务上转换为 ContinueWith，该任务将在任务池上执行，因此是异步的。

修复测试的多种方法:(从难看到复杂)

1. 在结果集之后休眠(注意等待不起作用，因为 Wait 不等待继续)

   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 之一)