ReactiveUI 测试中的调度器

Question

所以，当我为我的系统开发新功能时，我也尝试做一个 TDD - 遗憾的是，现在为旧功能做这件事的代码太大了。

但是，我发现有时我在测试期间会碰壁 - 特别是在使用 Delay 和 Throttle 时。

我读了很多书，我想我知道的比一周前多了很多，但我想把所有这些都付诸实践。我写了一些实验：

    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Reactive;
    using System.Reactive.Concurrency;
    using System.Reactive.Linq;
    using System.Reactive.Threading.Tasks;
    using System.Text;
    using System.Threading.Tasks;
    using Microsoft.Reactive.Testing;
    using NUnit.Framework;
    using NUnit.Framework.Internal.Commands;
    using ReactiveUI;
    using ReactiveUI.Testing;

    namespace UtilsTests
    {
        [TestFixture]
        public class SchedulersTests
        {
            private int SecondsN = 1;

            [Test]
            public async Task NoScheduler()
            {
                var t = Observable.Return(Unit.Default).Delay(TimeSpan.FromSeconds(SecondsN), RxApp.MainThreadScheduler)
                    .ObserveOn(RxApp.MainThreadScheduler)
                    .ToTask();
                await t;
            }

            [Test]
            public Task ImmediateSchedulerExperiment()
            {
                return Scheduler.Immediate.With(async s =>
                {
                    var t = Observable.Return(Unit.Default).Delay(TimeSpan.FromSeconds(SecondsN), RxApp.MainThreadScheduler).ToTask();
                    await t;
                });
            }

            [Test]
            public Task ImmediateSchedulerExperiment2()
            {
                return Scheduler.Immediate.With(async s =>
                {
                    var t = Observable.Return(Unit.Default).Delay(TimeSpan.FromSeconds(SecondsN), s).FirstAsync().ToTask();
                    await t;
                });
            }

            [Test]
            public void ImmediateSchedulerExperiment3()
            {
                Scheduler.Immediate.With(s =>
                {
                    var t = false;
                    Observable.Return(Unit.Default).Delay(TimeSpan.FromSeconds(SecondsN), s)
                        .Subscribe(_ =>
                        {
                            t = true;
                        });

                    Assert.IsTrue(t);
                });
            }


            [Test]
            public void TestSchedulerExperiment_SchedulersNotSpecified()
            {
                new TestScheduler().With(s =>
                {
                    var t = false;
                    Observable.Return(Unit.Default).Delay(TimeSpan.FromSeconds(SecondsN), s)
                        .Subscribe(_ =>
                        {
                            t = true;
                        });

                    s.AdvanceByMs(SecondsN * 1000);

                    Assert.IsTrue(t);
                });
            }

            [Test]
            public void TestSchedulerExperiment_DeylaOn_RxMainThread()
            {
                new TestScheduler().With(s =>
                {
                    var t = false;
                    Observable.Return(Unit.Default).Delay(TimeSpan.FromSeconds(SecondsN), RxApp.MainThreadScheduler)
                        .Subscribe(_ =>
                        {
                            t = true;
                        });

                    s.AdvanceByMs(SecondsN * 1000);

                    Assert.IsTrue(t);
                });
            }

            [Test]
            public void TestSchedulerExperiment_DeylaOn_RxTaskPool()
            {
                new TestScheduler().With(s =>
                {
                    var t = false;
                    Observable.Return(Unit.Default).Delay(TimeSpan.FromSeconds(SecondsN), RxApp.TaskpoolScheduler)
                        .Subscribe(_ =>
                        {
                            t = true;
                        });

                    s.AdvanceByMs(SecondsN * 1000);

                    Assert.IsTrue(t);
                });
            }

            [Test]
            public void TestSchedulerExperiment_RunOnTaskPool_ObserveOnMainThread()
            {
                new TestScheduler().With(s =>
                {
                    var t = false;
                    Observable.Return(Unit.Default)
                        .Delay(TimeSpan.FromSeconds(SecondsN), RxApp.TaskpoolScheduler)
                        .ObserveOn(RxApp.MainThreadScheduler)
                        .Subscribe(_ =>
                        {
                            t = true;
                        });

                    s.AdvanceByMs(SecondsN * 1000);

                    Assert.IsTrue(t);
                });
            }

            [Test]
            public void TestSchedulerExperiment_RunOnTaskPool_ObserveOnTaskpool()
            {
                new TestScheduler().With(s =>
                {
                    var t = false;
                    Observable.Return(Unit.Default)
                        .Delay(TimeSpan.FromSeconds(SecondsN), RxApp.TaskpoolScheduler)
                        .ObserveOn(RxApp.TaskpoolScheduler)
                        .Subscribe(_ =>
                        {
                            t = true;
                        });

                    s.AdvanceByMs(SecondsN * 1000);
                    s.AdvanceByMs(1);

                    Assert.IsTrue(t);
                });
            }

            [Test]
            public void TestSchedulerExperiment_RunOnTaskPool_ObserveOnMainThread_MainThreadIsAnotherInstance()
            {
                new TestScheduler().With(s =>
                {
                    var mainThreadScheduler = new TestScheduler();
                    RxApp.MainThreadScheduler = mainThreadScheduler;
                    var t = false;
                    Observable.Return(Unit.Default)
                        .Delay(TimeSpan.FromSeconds(SecondsN), RxApp.TaskpoolScheduler)
                        .ObserveOn(RxApp.MainThreadScheduler)
                        .Subscribe(_ =>
                        {
                            t = true;
                        });

                    s.AdvanceByMs(SecondsN * 1000);
                    mainThreadScheduler.AdvanceBy(1);
                    Assert.IsTrue(t);
                });
            }

            [Test]
            public void TestSchedulerExperiment_RunOnTest_ObserveOnTest()
            {
                new TestScheduler().With(s =>
                {
                    var t = false;
                    var obs = Observable.Return(Unit.Default)
                        .Delay(TimeSpan.FromSeconds(SecondsN), s)
                        .ObserveOn(s);
                    obs
                        .Subscribe(_ =>
                        {
                            t = true;
                        });

    //                s.AdvanceByMs(SecondsN * 1000);
    //                s.AdvanceBy(1);

                    s.AdvanceUntil(obs);

                    Assert.IsTrue(t);
                });
            }
        }


    }

起初，我认为 Scheduler.Immediate 会成功，在延迟后当场执行事情，男孩，那是错误的。我找到了 this article, which explained things rather nicely. I found also this post，解释了哪个运算符使用哪个调度程序。

我现在知道了，在玩时间的时候，我应该使用 TestScheduler。否则，不要更改调度程序。

我现在知道，您不会在构造函数中执行任何异步操作，而是创建一个名为 Init 的命令，它会在激活时执行此操作，您可以在测试中等待它（例如延迟收集基于构造函数参数的创建以允许在视图完成时平滑 UI 动画）

但是，当我运行从上面进行那些测试时，我得到了：

不明白的地方不多

1) 为什么 Scheduler.Immediate 测试需要两倍的时间？我想我明白为什么 Take(1) 没有什么区别，但仍然...

2) 使用TestSchduler时，如何确定前进多少？

我注意到在测试 TestSchedulerExperiment_RunOnTest_ObserveOnTest 中我必须做额外的 AdvanceBy(1)，因为它也是观察者。所以，当链条越长，观察者越多的时候，就很难统计了。

scheduler.AdvanceBy(10000000000000); 是常见的做法吗？

我尝试创建 AdvanceUntil 扩展，但我知道它很糟糕，原因有很多（例如冷可观察量）。

 public static void AdvanceUntil<TIgnore>(this TestScheduler s, IObservable<TIgnore> obs, double? advanceByMs = null)
        {
            var done = false;
            obs.Subscribe(_ => done = true, (ex) => done = true, () => done = true);

            while(!done)
                s.AdvanceByMs(advanceByMs ?? 100);
        }

或者有什么我不知道的"flush"方法？

此外，我学会了等待 TestScheduler.With:

中的内容

    [Test]
    public Task TestSchedulerExperiment_await()
    {
        return new TestScheduler().With(async s =>
        {
            var v = false;

            var t = Observable.Return(true).Delay(TimeSpan.FromSeconds(SecondsN), s)
                .Take(1) // without hits the test never ends
                .ToTask();
            s.AdvanceByMs(SecondsN * 1000);
            v = await t;

            Assert.IsTrue(v);
        });

但我还需要知道时间。

为什么必须有 Take(1)？

Answer 1

scheduler.Start() 执行已安排的所有内容，因此您不需要该扩展方法。

我建议大多数时候不要将 async/await 与 Rx 混合使用，尤其是对于基于时间的功能，由于 Delay 运算符，这基本上是您的所有测试。否则，您可能需要等待几分钟才能完成单个测试。所以 async/await 在其中任何一个中都没有用。

例如，在像您的 TestSchedulerExperiment await 测试这样的场景中，您只需要测试调度程序和订阅即可。该测试将简单地变成：

// Passing test    
[Test]
public void TestSchedulerExperiment()
{
    new TestScheduler().With(s =>
    {
        var v = false;

        Observable
            .Return(true)
            .Delay(TimeSpan.FromSeconds(1), s)
            .Subscribe(_ => v = true);

        s.Start();
        Console.WriteLine("Scheduler clock value: {0}", s.Clock);

        Assert.True(v);
    });
}

Why with Scheduler.Immediate the tests take twice the time?

如果您真的想深入研究并了解幕后发生的事情，我强烈推荐此 Spy extension by James 并添加时间戳。

var t = Observable
    .Return(Unit.Default).Spy("Return")
    .Delay(TimeSpan.FromSeconds(2), RxApp.MainThreadScheduler).Spy("Delay")
    .ToTask();
await t;

// Partial output
Return: OnNext(()) on Thread: 1, 23:22:41.2631845
Delay: OnNext(()) on Thread: 1, 23:22:43.2891836
Return: OnCompleted() on Thread: 1, 23:22:43.2921808
Delay: OnCompleted() on Thread: 1, 23:22:45.2958130

Return 使用 ImmediateScheduler，您可能知道，RxApp.MainThreadScheduler = ImmediateScheduler 在单元测试运行器中。因为这个 scheduler is synchronous Return 和 Delay 通知都必须互相等待。 Return 在延迟触发 OnNext 之前无法触发其 OnCompleted，然后延迟的 OnCompleted 通知再延迟 2 秒。

ReactiveUI 测试中的调度器

Schedulers in ReactiveUI testing

c#

unit-testing

system.reactive

reactiveui