unit-testing - 使用 Wait 方法对响应式(Reactive)扩展进行单元测试

Question

我正在尝试对在 IObservable 上使用 Wait() 方法的方法进行单元测试，但是我的测试从未完成 - Wait 从未完成。我的测试包含以下内容:

var scheduler = new TestScheduler();
var input1 = scheduler.CreateColdObservable<List<string>>(
   new Recorded<Notification<List<string>>>(100, Notification.CreateOnNext(new List<string> { "John", "Harry" })),
   new Recorded<Notification<List<string>>>(200, Notification.CreateOnCompleted<List<string>>())
          );

我正在使用 Moq 通过返回 input1 来设置我的方法的响应。例如

myObj.Setup(f => f.GetStrings()).Returns(input1);

myObj 的细节实际上并不重要。我启动调度程序并调用包含 Wait 的方法(例如，在我调用的方法中的某处)

var results = myObj.GetStrings().Wait();

但这永远不会回来。我怀疑我使用了错误的调度程序，但我不确定。

问候
艾伦

Answer 1

概括

问题是您正在创建一个冷的 observable 并在您订阅它之前推进调度程序。

细节

如果您调用阻止 Wait()在单线程测试上运行，你在这一点上是死在水中。这是因为 TestScheduler的内部时钟只有在您调用 Start() 时才会提前或 AdvanceXXX() 之一方法，并且由于您有一个冷的 observable，您指定的事件时间是相对于订阅点的。调用 Start() 也有一些细微差别我将在下面解释。

所以，如 Wait会阻塞，您可能会尝试在另一个线程上调用它，但这仍然很棘手。考虑以下代码，它与您的类似:

void Main()
{
    var scheduler = new TestScheduler();
    var source = scheduler.CreateColdObservable(
        new Recorded<Notification<int>>(100, Notification.CreateOnNext(1)),
        new Recorded<Notification<int>>(200, Notification.CreateOnCompleted<int>()));

    // (A)

    int result = 0;

    var resultTask = Task.Run(() => { result = source.Wait(); });

    // (B)

    resultTask.Wait();

    Console.WriteLine(result);
}

此代码尝试在后台线程上等待。如果我们插入对 scheduler.Start() 的调用在点 (A)，然后 source.Wait()将永远阻塞。

这是因为 Start()只会提前 TestScheduler 的内部时钟直到所有当前调度的事件都被执行。使用冷 observable，事件是相对于订阅的虚拟时间安排的。由于点 (A) 处没有订阅者，您会发现 TestScheduler.Now.Ticks即使在调用 Start() 后也会报告 0 .

嗯。如果我们将调用转移到 scheduler.Start()，事情会变得更糟。到 B 点。现在我们有一个竞争条件!这是一种竞争条件，几乎总是会导致测试在调用 resultTask.Wait() 时挂起。 .这是因为很有可能是 resultTask将没有时间执行它的操作并订阅 source之前 scheduler.Start()调用执行 - 所以时间不会再向前推进。

因此，确定性执行很难实现 - 没有很好的方式来宣布 Wait()调用已在提前时间之前发出，因为 Wait() 调用本身会阻塞。在调用 Start() 之前插入足够长的延迟会起作用，但有点违背使用 TestScheduler 的目的:

// (B)
Task.Delay(2000).Wait();        
scheduler.AdvanceBy(200);

这个问题真正向我展示的(恕我直言)是调用 Wait()并且阻塞线程几乎总是一个坏主意。寻找使用类似 LastAsync() 的方法相反，和/或使用延续来获取异步方法的结果。

由于复杂性，我不能推荐这种方法，但这里有一个确定性的解决方案，它使用扩展方法在进行订阅时发出信号。

void Main()
{
    var scheduler = new TestScheduler();
    var source = scheduler.CreateColdObservable(
        new Recorded<Notification<int>>(100, Notification.CreateOnNext(1)),
        new Recorded<Notification<int>>(200, Notification.CreateOnCompleted<int>()));

    // (A)      
    var waitHandle = new AutoResetEvent(false);

    int result = 0;                         
    var resultTask = Task.Run(() =>
    {
        result = source.AnnounceSubscription(waitHandle).Wait();
    });

    // (B)
    waitHandle.WaitOne();   
    scheduler.Start();

    resultTask.Wait();

    Console.WriteLine(result);
}

public static class ObservableExtensions
{
    public static IObservable<T> AnnounceSubscription<T>(
        this IObservable<T> source, AutoResetEvent are)
    {
        return Observable.Create<T>(o =>
        {
            var sub = source.Subscribe(o);
            are.Set();
            return sub;
        });
    }
}

测试 Rx 的推荐方法
TestScheduler 的更惯用用法是创建一个观察者来收集结果，然后断言它们符合预期。就像是:

void Main()
{
    var scheduler = new TestScheduler();
    var source = scheduler.CreateColdObservable(
        new Recorded<Notification<int>>(100, Notification.CreateOnNext(1)),
        new Recorded<Notification<int>>(200, Notification.CreateOnCompleted<int>()));

    var results = scheduler.CreateObserver<int>();

    // here you would append to source the Rx calls that do something interesting
    source.Subscribe(results);

    scheduler.Start();

    results.Messages.AssertEqual(
        new Recorded<Notification<int>>(100, Notification.CreateOnNext(1)),
        new Recorded<Notification<int>>(200, Notification.CreateOnCompleted<int>()));   
}

最后，如果你从 ReactiveTest 派生一个单元测试类您可以利用 OnNext , OnCompleted和 OnError创建的辅助方法 Recorded<Notification<T>>以更具可读性的方式实例化。