单元测试使用Wait的方法进行Reactive Extensions

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

摘要

问题是您在订阅之前创建了一个冷可观察对象并推进了调度程序.

详情

如果你Wait()在单线程测试中调用阻塞操作,那么你就会死在水中.这是因为TestScheduler只有当你调用Start()或使用其中一个AdvanceXXX()方法时,内部时钟才会前进,因为你有一个冷可观察对象,你指定的事件时间是相对于订阅点.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直到执行所有当前调度的事件.使用冷可观察性,相对于虚拟订阅时间安排事件.由于在点(A)没有订阅者,你会发现TestScheduler.Now.Ticks即使在调用之后也会报告0 Start().

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

因此很难实现确定性执行 - 没有很好的方法来宣布Wait()在推进时间之前已经发出了调用,因为Wait()调用本身将会阻塞.在调用之前插入足够长的延迟Start()将起作用,但有点失败使用的对象TestScheduler:

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

这个问题真正向我(IMHO)证明了调用Wait()和阻塞线程几乎总是一个坏主意.寻找使用类似的方法LastAsync(),和/或使用continuation来获取异步方法的结果.

由于复杂性,我不能推荐这种方法,但这是一个确定性的解决方案,它利用扩展方法在订阅时发出信号.

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>>更可读的方式实例.