Rx第六部分测试

测试软件的根源在于调试和推演代码。由于过去的手工测试在很大程度上已经成熟，这些测试试图“破坏应用程序”，现代的质量保证标准需要实现自动化来帮助评估和防止错误。虽然测试专家团队很常见，但是越来越多的程序员期望通过自动化测试套件提供质量保证。

到目前为止，我们已经涵盖了Rx的全部知识点，并且我们已经有足够的知识开始使用Rx !尽管如此，许多开发人员还是希望在编写代码前编写测试用例。测试可以用来证明代码实际上满足了需求，提供了抵御回归的安全网，甚至可以帮助代码文档化。本章假设您熟悉依赖注入和单元测试的概念，如mock或stub。

Testing software has its roots in debugging and demonstrating code. Having largely matured past manual tests that try to "break the application", modern quality assurance standards demand a level of automation that can help evaluate and prevent bugs. While teams of testing specialists are common, more and more coders are expected to provide quality guarantees via automated test suites.

Up to this point, we have covered a broad scope of Rx, and we have almost enough knowledge to start using Rx in anger! Still, many developers would not dream of coding without first being able to write tests. Tests can be used to prove that code is in fact satisfying requirements, provide a safety net against regression and can even help document the code. This chapter makes the assumption that you are familiar with the concepts of dependency injection and unit testing with test-doubles, such as mocks or stubs.

Rx给测试驱动社区引入了如下有趣的问题：

在测试场景中，调度(以及线程)通常被避免，因为它会引入可能导致非确定性测试的竞争条件
测试应该尽可能快地运行
对于许多人来说，Rx是一种新的技术/库。很自然地，当我们逐步掌握Rx的过程中，我们可能想重构一些以前的Rx代码。我们希望使用测试来确保重构没有改变代码库的内部行为
同样，当我们升级Rx版本时，测试将保证功能没有被破坏。

Rx poses some interesting problems to our Test-Driven community:

Scheduling, and therefore threading, is generally avoided in test scenarios as it can introduce race conditions which may lead to non-deterministic tests
Tests should run as fast as possible
For many, Rx is a new technology/library. Naturally, as we progress on our journey to mastering Rx, we may want to refactor some of our previous Rx code. We want to use tests to ensure that our refactoring has not altered the internal behavior of our code base
Likewise, tests will ensure nothing breaks when we upgrade versions of Rx.

虽然我们确实想要测试我们的代码，但我们不想引入缓慢或不确定的测试；事实上，后者会引入假阴性或假阳性。如果我们看一下Rx库，就会发现有很多方法涉及到调度(隐式或显式的)，要高效的使用Rx就很难避免使用调度时序。这个LINQ查询告诉我们，至少有26个扩展方法接受IScheduler作为参数。

While we do want to test our code, we don't want to introduce slow or non-deterministic tests; indeed, the later would introduce false-negatives or false-positives. If we look at the Rx library, there are plenty of methods that involve scheduling (implicitly or explicitly), so using Rx effectively makes it hard to avoid scheduling. This LINQ query shows us that there are at least 26 extension methods that accept an IScheduler as a parameter.

var query = from method in typeof(Observable).GetMethods()
from parameter in method.GetParameters()
where typeof (IScheduler).IsAssignableFrom(parameter.ParameterType)
group method by method.Name
into morderby m.Keyselect m.Key;
foreach (var methodName in query)
{Console.WriteLine(methodName);
}

输出:

Buffer

Delay

Empty

Generate

Interval

Merge

ObserveOn

Range

Repeat

Replay

Return

Sample

Start

StartWith

SubscribeOn

Take

Throttle

Throw

TimeInterval

Timeout

Timer

Timestamp

ToAsync

ToObservable

Window

这些方法中的许多都有一个重载，它不使用IScheduler类型参数，而是使用他的一个默认实例。TDD/Test 程序员希望选择接受IScheduler的重载，这样他们就可以在测试中控制调度时序。我很快会解释为什么。

考虑一下这个例子，我们创建了一个序列，该序列在5秒钟内每秒发布一个数值。

Many of these methods also have an overload that does not take an IScheduler and instead uses a default instance. TDD/Test First coders will want to opt for the overload that accepts the IScheduler, so that they can have some control over scheduling in our tests. I will explain why soon.

Consider this example, where we create a sequence that publishes values every second for five seconds.

var interval = Observable

.Interval(TimeSpan.FromSeconds(1))

.Take(5);

如果我们要编写一个测试来确保我们接收到5个值，并且它们每一个值相隔1秒，那么运行需要5秒。这样不是很好；我想要在五秒内运行成百上千个测试。另一个非常常见的需求是测试超时。在这里，我们尝试测试一分钟的超时。

If we were to write a test that ensured that we received five values and they were each one second apart, it would take five seconds to run. That would be no good; I want hundreds if not thousands of tests to run in five seconds. Another very common requirement is to test a timeout. Here, we try to test a timeout of one minute.

var never = Observable.Never<int>();

var exceptionThrown = false;

never.Timeout(TimeSpan.FromMinutes(1))

.Subscribe(

i => Console.WriteLine("This will never run."),

ex => exceptionThrown = true);

Assert.IsTrue(exceptionThrown);

这里有两个问题：

要么断言运行得太快，测试毫无意义，因为它总是失败，或者
为了进行准确的测试，我们必须增加一分钟的延迟

为了使这个测试可用，运行它需要一分钟。运行一分钟的单元测试是不可接受的。

We have two problems here:

either the Assert runs too soon, and the test is pointless as it always fails, or
we have to add a delay of one minute to perform an accurate test

For this test to be useful, it would therefore take one minute to run. Unit tests that take one minute to run are not acceptable.

TestScheduler

为解决上述问题引入了TestScheduler；它引入了虚拟调度器的概念，允许我们模拟和控制时序。

可以将虚拟调度器理解为要执行的动作队列。每一个都被分配了一个时间点，当它们应该被执行的时候。我们使用TestScheduler作为生产IScheduler类型的替代或测试替身。使用这个虚拟调度器，我们可以执行所有队列的操作，或者只执行指定时间点的操作。

在本例中，我们使用简单的重载(schedule (Action))将任务调度到队列上，以便立即运行。然后，我们将虚拟时钟向前推进了一分钟。通过这样做，我们可以及时执行到那个时间点的所有计划。请注意，尽管我们计划立即执行一个操作，但是在手动推进时钟之前，它实际上不会执行。

To our rescue comes the TestScheduler; it introduces the concept of a virtual scheduler to allow us to emulate and control time.

A virtual scheduler can be conceptualized as a queue of actions to be executed. Each are assigned a point in time when they should be executed. We use the TestScheduler as a substitute, or test double, for the production IScheduler types. Using this virtual scheduler, we can either execute all queued actions, or only those up to a specified point in time.

In this example, we schedule a task onto the queue to be run immediately by using the simple overload (Schedule(Action)). We then advance the virtual clock forward by one tick. By doing so, we execute everything scheduled up to that point in time. Note that even though we schedule an action to be executed immediately, it will not actually be executed until the clock is manually advanced.

var scheduler = new TestScheduler();
var wasExecuted = false;
scheduler.Schedule(() => wasExecuted = true);
Assert.IsFalse(wasExecuted);
scheduler.AdvanceBy(1); //execute 1 tick of queued actions
Assert.IsTrue(wasExecuted);

运行和调试这个示例可以帮助您更好地理解TestScheduler的基础知识。

TestScheduler实现了IScheduler接口，并且扩展了它，允许我们控制和监视虚拟时序。我们已经熟悉IScheduler.Schedule方法，然而TestScheduler所特有的AdvanceBy(long)、AdvanceTo(long)和Start()方法是最有趣的。同样，Clock属性也很有趣，因为它可以帮助我们理解内部发生了什么。

Running and debugging this example may help you to better understand the basics of the TestScheduler.

The TestScheduler implements the IScheduler interface (naturally) and also extends it to allow us to control and monitor virtual time. We are already familiar with the IScheduler.Schedule methods, however the AdvanceBy(long), AdvanceTo(long) and Start() methods unique to the TestScheduler are of most interest. Likewise, the Clock property will also be of interest, as it can help us understand what is happening internally.

public class TestScheduler : ...
{//Implementation of ISchedulerpublic DateTimeOffset Now { get; }public IDisposable Schedule<TState>(TState state,Func<IScheduler, TState, IDisposable> action)public IDisposable Schedule<TState>(TState state,TimeSpan dueTime,Func<IScheduler, TState, IDisposable> action)public IDisposable Schedule<TState>(TState state,DateTimeOffset dueTime,Func<IScheduler, TState, IDisposable> action)//Useful extensions for testingpublic bool IsEnabled { get; private set; }public TAbsolute Clock { get; protected set; }public void Start()public void Stop()public void AdvanceTo(long time)public void AdvanceBy(long time)//Other methods...
}

AdvanceTo方法

AdvanceTo(long)方法将执行已调度到指定的绝对时间的所有操作。TestScheduler使用CPU时钟周期（ticks）作为它的时间度量。在本例中，我们计划在第10个CPU时钟周期和第20个CPU时钟周期时调用操作。

The AdvanceTo(long) method will execute all the actions that have been scheduled up to the absolute time specified. The TestScheduler uses ticks as its measurement of time. In this example, we schedule actions to be invoked now, in 10 ticks, and in 20 ticks.

var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A"));
//Schedule immediatelys
cheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));
Console.WriteLine("scheduler.AdvanceTo(1);");
scheduler.AdvanceTo(1);
Console.WriteLine("scheduler.AdvanceTo(10);");
scheduler.AdvanceTo(10);
Console.WriteLine("scheduler.AdvanceTo(15);");
scheduler.AdvanceTo(15);
Console.WriteLine("scheduler.AdvanceTo(20);");
scheduler.AdvanceTo(20);

输出:

scheduler.AdvanceTo(1);

scheduler.AdvanceTo(10);

scheduler.AdvanceTo(15);

scheduler.AdvanceTo(20);

注意前移15个时钟周期时没有任何输出。前两个动作在第15个时钟周期之前已经执行，最后一个动作还需要继续前移。

Note that nothing happened when we advanced to 15 ticks. All work scheduled before 15 ticks had been performed and we had not advanced far enough yet to get to the next scheduled action.

AdvanceBy

AdvanceBy(long)方法可以让我们向前移动相对的时钟周期。同样单位时时钟周期。可将上例改为AdvanceBy(long)。

The AdvanceBy(long) method allows us to move the clock forward a relative amount of time. Again, the measurements are in ticks. We can take the last example and modify it to use AdvanceBy(long).

var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A"));
//Schedule immediately
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));
Console.WriteLine("scheduler.AdvanceBy(1);");
scheduler.AdvanceBy(1);
Console.WriteLine("scheduler.AdvanceBy(9);");
scheduler.AdvanceBy(9);
Console.WriteLine("scheduler.AdvanceBy(5);");
scheduler.AdvanceBy(5);
Console.WriteLine("scheduler.AdvanceBy(5);");
scheduler.AdvanceBy(5);

输出:

scheduler.AdvanceBy(1);

scheduler.AdvanceBy(9);

scheduler.AdvanceBy(5);

Start

TestScheduler.Start()方法执行所有已经调度的动作。我们在次使用相同的范例，这里将AdvanceBy(long)替换为Start()。

The TestScheduler's Start() method is an effective way to execute everything that has been scheduled. We take the same example again and swap out the AdvanceBy(long) calls for a single Start() call.

var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A"));
//Schedule immediately
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));
Console.WriteLine("scheduler.Start();");
scheduler.Start();
Console.WriteLine("scheduler.Clock:{0}", scheduler.Clock);

输出:

scheduler.Start();

scheduler.Clock:20

注意，一旦执行了所有以调度的操作，虚拟时钟就会匹配上一个最近的项(20时针周期)。

进一步扩展我们的示例，在调用了Start()方法后继续调度一个动作。

Note that once all of the scheduled actions have been executed, the virtual clock matches our last scheduled item (20 ticks).

We further extend our example by scheduling a new action to happen after Start() has already been called.

var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A"));
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));
Console.WriteLine("scheduler.Start();");
scheduler.Start();
Console.WriteLine("scheduler.Clock:{0}", scheduler.Clock);
scheduler.Schedule(() => Console.WriteLine("D"));

输出:

scheduler.Start();

scheduler.Clock:20

注意输出是完全一致的；如果向执行第四个动作，需要再次调用Start()方法。

Note that the output is exactly the same; If we want our fourth action to be executed, we will have to call Start() again.

Stop

在Rx以前的版本中，Start()方法被称为Run()。现在有一个Stop()方法，它的名字暗示了与Start()的对称性。但是，它所做的只是将IsEnabled属性设置为false。此属性用作内部标志，用于检查是否应该继续执行操作的内部队列。队列的处理可以由Start()发起，也可以使用AdvanceTo或AdvanceBy。

在本例中，我们将展示如何使用Stop()暂停以调度动作的处理。

In previous releases of Rx, the Start() method was called Run(). Now there is a Stop() method whose name seems to imply some symmetry with Start(). All it does however, is set the IsEnabled property to false. This property is used as an internal flag to check whether the internal queue of actions should continue being executed. The processing of the queue may indeed be instigated by Start(), however AdvanceTo or AdvanceBy can be used too.

In this example, we show how you could use Stop() to pause processing of scheduled actions.

var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A"));
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(15), scheduler.Stop);
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));
Console.WriteLine("scheduler.Start();");
scheduler.Start();
Console.WriteLine("scheduler.Clock:{0}", scheduler.Clock);

输出:

scheduler.Start();

scheduler.Clock:15

注意我们在第15时钟周期时停止了动作执行，C没有从控制台输出。到目前为止，我已经成功地测试了Rx近两年了，但是我还没有发现使用Stop()方法的必要性。我想有些情况下是值得使用的;然而，我只是想说明一点，您不必担心在您的测试中没有使用它。

Note that "C" never gets printed as we stop the clock at 15 ticks. I have been testing Rx successfully for nearly two years now, yet I have not found the need to use the Stop() method. I imagine that there are cases that warrant its use; however I just wanted to make the point that you do not have to be concerned about the lack of use of it in your tests.

调度冲突（Schedule collisions）

在调度操作时，有可能甚至有可能将许多操作安排在同一时间点。这通常发生在同时调度多个操作时。还可能在多个动作延时到同一时间点执行。TestScheduler有一种简单的方法来处理这个问题。当动作被调度时，它们标记出执行的时刻。如果多个项目被安排在同一时间点，那么它们将按照被安排的顺序排队;当时钟前移时，该时间点的所有动作都按照调度的顺序执行。

When scheduling actions, it is possible and even likely that many actions will be scheduled for the same point in time. This most commonly would occur when scheduling multiple actions for now. It could also happen that there are multiple actions scheduled for the same point in the future. The TestScheduler has a simple way to deal with this. When actions are scheduled, they are marked with the clock time they are scheduled for. If multiple items are scheduled for the same point in time, they are queued in order that they were scheduled; when the clock advances, all items for that point in time are executed in the order that they were scheduled.

var scheduler = new TestScheduler();
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("A"));
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("C"));
Console.WriteLine("scheduler.Start();");
scheduler.Start();
Console.WriteLine("scheduler.Clock:{0}", scheduler.Clock);

输出:

scheduler.AdvanceTo(10);

scheduler.Clock:10

注意虚拟时针移到第10个时针周期。

Note that the virtual clock is at 10 ticks, the time we advanced to.

测试Rx代码（Testing Rx code）

现在我们已经了解了一些TestScheduler的知识，让我们看看如何使用它来测试使用Interval和Timeout的两个初始代码片段。我们希望尽可能快地执行测试，但同时也要保持时序的语义。在本例中，我们每隔一秒钟生成5个值，但是将TestScheduler传递给要使用的Interval方法，而不是默认的调度器。

Now that we have learnt a little bit about the TestScheduler, let's look at how we could use it to test our two initial code snippets that use Interval and Timeout. We want to execute tests as fast as possible but still maintain the semantics of time. In this example we generate our five values one second apart but pass in our TestScheduler to the Interval method to use instead of the default scheduler.

[TestMethod]
public void Testing_with_test_scheduler()
{var expectedValues = new long[] {0, 1, 2, 3, 4};var actualValues = new List<long>();var scheduler = new TestScheduler();var interval = Observable.Interval(TimeSpan.FromSeconds(1), scheduler).Take(5);interval.Subscribe(actualValues.Add);scheduler.Start();CollectionAssert.AreEqual(expectedValues, actualValues);//Executes in less than 0.01s "on my machine"
}

有点意思，但我认为更重要的是我们如何测试一段真正的代码。想象一下，在ViewModel上订阅价格序列。价格抛出后，将其添加到一个集合中。假设这是一个WPF或Silverlight实现，我们可以轻松的强制在ThreadPool上执行订阅，并在Dispatcher上执行观察。

While this is mildly interesting, what I think is more important is how we would test a real piece of code. Imagine, if you will, a ViewModel that subscribes to a stream of prices. As prices are published, it adds them to a collection. Assuming this is a WPF or Silverlight implementation, we take the liberty of enforcing that the subscription be done on the ThreadPool and the observing is executed on the Dispatcher.

public class MyViewModel : IMyViewModel
{private readonly IMyModel _myModel;private readonly ObservableCollection<decimal> _prices;public MyViewModel(IMyModel myModel){_myModel = myModel;_prices = new ObservableCollection<decimal>();}public void Show(string symbol){//TODO: resource mgt, exception handling etc..._myModel.PriceStream(symbol).SubscribeOn(Scheduler.ThreadPool).ObserveOn(Scheduler.Dispatcher).Timeout(TimeSpan.FromSeconds(10), Scheduler.ThreadPool).Subscribe(Prices.Add,ex=>{if(ex is TimeoutException)IsConnected = false;});IsConnected = true;}public ObservableCollection<decimal> Prices{get { return _prices; }}public bool IsConnected { get; private set; }
}

注入调度器依赖（Injecting scheduler dependencies）

虽然上面的代码片段可能完成我们希望的工作，但是由于它通过静态属性访问调度器，因此很难进行测试。为了帮助我的测试，我创建了自己的接口，与Scheduler类实现的IScheduler接口完全一致，我建议您也采用这个接口。

While the snippet of code above may do what we want it to, it will be hard to test as it is accessing the schedulers via static properties. To help my testing, I have created my own interface that exposes the same IScheduler implementations that the Scheduler type does, i suggest you adopt this interface too.

public interface ISchedulerProvider
{IScheduler CurrentThread { get; }IScheduler Dispatcher { get; }IScheduler Immediate { get; }IScheduler NewThread { get; }IScheduler ThreadPool { get; }//IScheduler TaskPool { get; }
}

TaskPool属性是否应该包含取决于您的目标平台。如果您采用了这中方式，请根据您的命名约定来命名这个类型，例如，SchedulerService, Schedulers。用于生产环境下的默认实现如下:

Whether the TaskPool property should be included or not depends on your target platform. If you adopt this concept, feel free to name this type in accordance with your naming conventions e.g. SchedulerService, Schedulers. The default implementation that we would run in production is implemented as follows:

public sealed class SchedulerProvider : ISchedulerProvider
{public IScheduler CurrentThread{get { return Scheduler.CurrentThread; }}public IScheduler Dispatcher{get { return DispatcherScheduler.Instance; }}public IScheduler Immediate{get { return Scheduler.Immediate; }}public IScheduler NewThread{get { return Scheduler.NewThread; }}public IScheduler ThreadPool{get { return Scheduler.ThreadPool; }}//public IScheduler TaskPool { get { return Scheduler.TaskPool; } }
}

现在，我们来编写一个用于测试的ISchedulerProvider的实现类。ISchedulerProvider可能很简陋，但很容易实现一个用于测试目的的实现。我的测试实现如下。

This now allows me to substitute implementations of ISchedulerProvider to help with testing. I could mock the ISchedulerProvider, but I find it easier to provide a test implementation. My implementation for testing is as follows.

public sealed class TestSchedulers : ISchedulerProvider

{

private readonly TestScheduler _currentThread = new TestScheduler();

private readonly TestScheduler _dispatcher = new TestScheduler();

private readonly TestScheduler _immediate = new TestScheduler();

private readonly TestScheduler _newThread = new TestScheduler();

private readonly TestScheduler _threadPool = new TestScheduler();

#region Explicit implementation of ISchedulerService

IScheduler ISchedulerProvider.CurrentThread { get { return _currentThread; } }

IScheduler ISchedulerProvider.Dispatcher { get { return _dispatcher; } }

IScheduler ISchedulerProvider.Immediate { get { return _immediate; } }

IScheduler ISchedulerProvider.NewThread { get { return _newThread; } }

IScheduler ISchedulerProvider.ThreadPool { get { return _threadPool; } }

#endregion

public TestScheduler CurrentThread { get { return _currentThread; } }

public TestScheduler Dispatcher { get { return _dispatcher; } }

public TestScheduler Immediate { get { return _immediate; } }

public TestScheduler NewThread { get { return _newThread; } }

public TestScheduler ThreadPool { get { return _threadPool; } }

}

注意ISchedulerProvider是显式实现的。这意味着，在我们的测试中，我们可以直接访问TestScheduler实例，但是我们的系统测试(SUT)仍然只能看到接口实例。现在我可以为我的ViewModel编写一些测试。下面，我们测试MyViewModel类的一个修改版本，它使用ISchedulerProvider，而不是调度器类中的静态调度器。我们使用流行的Moq框架来模拟我们的模型。

Note that ISchedulerProvider is implemented explicitly. This means that, in our tests, we can access the TestScheduler instances directly, but our system under test (SUT) still just sees the interface implementation. I can now write some tests for my ViewModel. Below, we test a modified version of the MyViewModel class that takes an ISchedulerProvider and uses that instead of the static schedulers from the Scheduler class. We also use the popular Moq framework in order to mock out our model.

[TestInitialize]

public void SetUp()

{

_myModelMock = new Mock<IMyModel>();

_schedulerProvider = new TestSchedulers();

_viewModel = new MyViewModel(_myModelMock.Object, _schedulerProvider);

}

[TestMethod]

public void Should_add_to_Prices_when_Model_publishes_price()

{

decimal expected = 1.23m;

var priceStream = new Subject<decimal>();

_myModelMock.Setup(svc => svc.PriceStream(It.IsAny<string>())).Returns(priceStream);

_viewModel.Show("SomeSymbol");

//Schedule the OnNext

_schedulerProvider.ThreadPool.Schedule(() => priceStream.OnNext(expected));

Assert.AreEqual(0, _viewModel.Prices.Count);

//Execute the OnNext action

_schedulerProvider.ThreadPool.AdvanceBy(1);

Assert.AreEqual(0, _viewModel.Prices.Count);

//Execute the OnNext handler

_schedulerProvider.Dispatcher.AdvanceBy(1);

Assert.AreEqual(1, _viewModel.Prices.Count);

Assert.AreEqual(expected, _viewModel.Prices.First());

}

[TestMethod]

public void Should_disconnect_if_no_prices_for_10_seconds()

{

var timeoutPeriod = TimeSpan.FromSeconds(10);

var priceStream = Observable.Never<decimal>();

_myModelMock.Setup(svc => svc.PriceStream(It.IsAny<string>())).Returns(priceStream);

_viewModel.Show("SomeSymbol");

_schedulerProvider.ThreadPool.AdvanceBy(timeoutPeriod.Ticks - 1);

Assert.IsTrue(_viewModel.IsConnected);

_schedulerProvider.ThreadPool.AdvanceBy(timeoutPeriod.Ticks);

Assert.IsFalse(_viewModel.IsConnected);

}

输出:

2 passed, 0 failed, 0 skipped, took 0.41 seconds (MSTest 10.0).

这两个测试验证了五个事实：

随着模型不断发送数据项，Price属性中不断添加新的价格
序列订阅到线程池上
Price属性在Dispatcher上进行更新，及序列在Dispatcher上观察
十秒内没有给ViewModel设置价格数据，则超时断开
测试运行得很快。虽然运行测试的时间并不是很长，但大部分时间似乎都花在了测试工具的本身。此外，将测试计数增加到10只会增加0.03秒。通常，在现代的CPU上，我希望看到单元测试以每秒+1000个测试的速度运行

These two tests ensure five things:

That the Price property has prices added to it as the model produces them
That the sequence is subscribed to on the ThreadPool
That the Price property is updated on the Dispatcher i.e. the sequence is observed on the Dispatcher
That a timeout of 10 seconds between prices will set the ViewModel to disconnected.
The tests run fast. While the time to run the tests is not that impressive, most of that time seems to be spent warming up my test harness. Moreover, increasing the test count to 10 only adds 0.03seconds. In general, on a modern CPU, I expect to see unit tests run at a rate of +1000 tests per second

通常，每次测试用例不会有一个以上的断言/验证，但是在这里它确实有助于说明问题。在第一个测试中，我们可以看到，只有运行了ThreadPool和Dispatcher调度器之后，我们才能得到结果。在第二个测试中，它有助于验证不少于10秒的超时。

在某些场景中，您对调度器不感兴趣，希望将测试集中在其他功能上。如果是这样，那么您可能想要创建ISchedulerProvider的另一个测试实现，它的所有成员都返回ImmediateScheduler 。这可以帮助你减少测试中的噪音。

Usually, I would not have more than one assert/verify per test, but here it does help illustrate a point. In the first test, we can see that only once both the ThreadPool and the Dispatcher schedulers have been run will we get a result. In the second test, it helps to verify that the timeout is not less than 10 seconds.

In some scenarios, you are not interested in the scheduler and you want to be focusing your tests on other functionality. If this is the case, then you may want to create another test implementation of the ISchedulerProvider that returns the ImmediateScheduler for all of its members. That can help reduce the noise in your tests.

public sealed class ImmediateSchedulers : ISchedulerService

{

public IScheduler CurrentThread { get { return Scheduler.Immediate; } }

public IScheduler Dispatcher { get { return Scheduler.Immediate; } }

public IScheduler Immediate { get { return Scheduler.Immediate; } }

public IScheduler NewThread { get { return Scheduler.Immediate; } }

public IScheduler ThreadPool { get { return Scheduler.Immediate; } }

}

高级特性 - ITestableObserver

TestScheduler提供了更高级特性。我发现没有这些方法也可以，但其他人可能会觉得它们有用。也许这是因为我发现自己已经习惯于使用早期版本的Rx进行测试。

The TestScheduler provides further advanced features. I find that I am able to get by quite well without these methods, but others may find them useful. Perhaps this is because I have found myself accustomed to testing without them from using earlier versions of Rx.

Start(Func<IObservable<T>>)

Start函数有三个重载，它们用于在给定时间启动一个可观察序列，记录它发出的通知并在给定时间进行订阅和释放订阅。这在一开始可能会让人困惑，因为这与没有参数的Start重载是完全不相关的。这三个重载返回一个ITestableObserver，它允许你记录来自一个可观察序列的通知，很像我们在Transformation chapter中看到的Materialize方法。

There are three overloads to Start, which are used to start an observable sequence at a given time, record the notifications it makes and dispose of the subscription at a given time. This can be confusing at first, as the parameterless overload of Start is quite unrelated. These three overloads return an ITestableObserver<T> which allows you to record the notifications from an observable sequence, much like the Materialize method we saw in the Transformation chapter.

public interface ITestableObserver<T> : IObserver<T>

{

// Gets recorded notifications received by the observer.

IList<Recorded<Notification<T>>> Messages { get; }

}

虽然有三个重载，但我们将首先看最具体的一个。这个重载需要4个参数:

一个可观测序列工厂委托
指定执行工厂的时间
指定订阅工厂返回的可观测序列的时间
指定释放订阅的时间

While there are three overloads, we will look at the most specific one first. This overload takes four parameters:

an observable sequence factory delegate
the point in time to invoke the factory
the point in time to subscribe to the observable sequence returned from the factory
the point in time to dispose of the subscription

最后三个时间参数以时针周期为单位，可参考TestScheduler的其他成员。

The time for the last three parameters is measured in ticks, as per the rest of the TestScheduler members.

public ITestableObserver<T> Start<T>(

Func<IObservable<T>> create,

long created,

long subscribed,

long disposed)

{...}

我们可以用这种方法来测试。工厂方法。在这里，我们创建一个可观察序列，每秒钟生成一个值，持续4秒。我们使用TestScheduler.Start方法立即创建并订阅它(通过传递0作为第二个和第三个参数)。我们在5秒后释放订阅。一旦开始方法运行，我们就输出所记录的内容。

We could use this method to test the Observable.Interval factory method. Here, we create an observable sequence that spawns a value every second for 4 seconds. We use the TestScheduler.Start method to create and subscribe to it immediately (by passing 0 for the second and third parameters). We dispose of our subscription after 5 seconds. Once the Start method has run, we output what we have recorded.

var scheduler = new TestScheduler();

var source = Observable.Interval(TimeSpan.FromSeconds(1), scheduler)

.Take(4);

var testObserver = scheduler.Start(

() => source,

TimeSpan.FromSeconds(5).Ticks);

Console.WriteLine("Time is {0} ticks", scheduler.Clock);

Console.WriteLine("Received {0} notifications", testObserver.Messages.Count);

foreach (Recorded<Notification<long>> message in testObserver.Messages)

{

Console.WriteLine("{0} @ {1}", message.Value, message.Time);

}

Output:

Time is 50000000 ticks

Received 5 notifications

OnNext(0) @ 10000001

OnNext(1) @ 20000001

OnNext(2) @ 30000001

OnNext(3) @ 40000001

OnCompleted() @ 40000001

注意ITestObserver<T>记录了OnNext和OnCompleted通知。如果序列错误中止，ITestObserver<T>将记录一个OnError通知。

我们可以使用输入变量来观察它的影响。我们知道Observable.Interval方法是冷可观测序列，因此创建的虚拟时间无关紧要。更改订阅的虚拟时间可以更改我们的结果。如果我们将其更改为2秒，我们会注意到如果我们将释放时间设置为5秒，我们将会错过一些消息。

Note that the ITestObserver<T> records OnNext and OnCompleted notifications. If the sequence was to terminate in error, the ITestObserver<T> would record the OnError notification instead.

We can play with the input variables to see the impact it makes. We know that the Observable.Interval method is a Cold Observable, so the virtual time of the creation is not relevant. Changing the virtual time of the subscription can change our results. If we change it to 2 seconds, we will notice that if we leave the disposal time at 5 seconds, we will miss some messages.

var testObserver = scheduler.Start(

() => Observable.Interval(TimeSpan.FromSeconds(1), scheduler).Take(4),

TimeSpan.FromSeconds(2).Ticks,

TimeSpan.FromSeconds(5).Ticks);

Output:

Time is 50000000 ticks

Received 2 notifications

OnNext(0) @ 30000000

OnNext(1) @ 40000000

我们从2秒开始订阅;Interval

在之后的每秒钟都会发送值(即秒3和4)，我们在第5秒上释放订阅。因此，我们错过了另外两个OnNext消息以及OnCompleted消息。

方法TestScheduler.Start还有两个重载。

We start the subscription at 2 seconds; the Interval produces values after each second (i.e. second 3 and 4), and we dispose on second 5. So we miss the other two OnNext messages as well as the OnCompleted message.

There are two other overloads to this TestScheduler.Start method.

public ITestableObserver<T> Start<T>(Func<IObservable<T>> create, long disposed)

{

if (create == null)

throw new ArgumentNullException("create");

else

return this.Start<T>(create, 100L, 200L, disposed);

}

public ITestableObserver<T> Start<T>(Func<IObservable<T>> create)

{

if (create == null)

throw new ArgumentNullException("create");

else

return this.Start<T>(create, 100L, 200L, 1000L);

}

如您所见，这些重载只是调用了我们一直在研究的实现，只是传递了一些默认值。我不知道为什么这些默认值是特殊的;我无法想象您为什么要使用这两个方法，除非您的特定用例与特定配置完全匹配。

As you can see, these overloads just call through to the variant we have been looking at, but passing some default values. I am not sure why these default values are special; I can not imagine why you would want to use these two methods, unless your specific use case matched that specific configuration exactly.

CreateColdObservable

正如我们可以记录可观测序列，通用可以使用CreateColdObservable来回播Recorded<Notification<int>>集合。CreateColdObservable方法的签名有一个记录通知的变体数组参数。

Just as we can record an observable sequence, we can also use CreateColdObservable to playback a set of Recorded<Notification<int>>. The signature for CreateColdObservable simply takes a params array of recorded notifications.

// Creates a cold observable from an array of notifications.

// Returns a cold observable exhibiting the specified message behavior.

public ITestableObservable<T> CreateColdObservable<T>(

params Recorded<Notification<T>>[] messages)

{...}

CreateColdObservable 返回 ITestableObservable<T>. 这个接口继承了IObservable<T> ，添加了订阅List和产生的消息List。

The CreateColdObservable returns an ITestableObservable<T>. This interface extends IObservable<T> by exposing the list of "subscriptions" and the list of messages it will produce.

public interface ITestableObservable<T> : IObservable<T>

{

// Gets the subscriptions to the observable.

IList<Subscription> Subscriptions { get; }

// Gets the recorded notifications sent by the observable.

IList<Recorded<Notification<T>>> Messages { get; }

}

使用CreateColdObservable, 可以模拟Observable.Interval.

Using CreateColdObservable, we can emulate the Observable.Interval test we had earlier.

var scheduler = new TestScheduler();

var source = scheduler.CreateColdObservable(

new Recorded<Notification<long>>(10000000, Notification.CreateOnNext(0L)),

new Recorded<Notification<long>>(20000000, Notification.CreateOnNext(1L)),

new Recorded<Notification<long>>(30000000, Notification.CreateOnNext(2L)),

new Recorded<Notification<long>>(40000000, Notification.CreateOnNext(3L)),

new Recorded<Notification<long>>(40000000, Notification.CreateOnCompleted<long>())

);

var testObserver = scheduler.Start(

() => source,

TimeSpan.FromSeconds(5).Ticks);

Console.WriteLine("Time is {0} ticks", scheduler.Clock);

Console.WriteLine("Received {0} notifications", testObserver.Messages.Count);

foreach (Recorded<Notification<long>> message in testObserver.Messages)

{

Console.WriteLine(" {0} @ {1}", message.Value, message.Time);

}

Output:

Time is 50000000 ticks

Received 5 notifications

OnNext(0) @ 10000001

OnNext(1) @ 20000001

OnNext(2) @ 30000001

OnNext(3) @ 40000001

OnCompleted() @ 40000001

注意输出与前面的Observable.Interval范例完全一致。

Note that our output is exactly the same as the previous example with Observable.Interval.

CreateHotObservable

也可使用CreateHotObservable函数创建热测试可观测序列。与CreateColdObservable函数参数一致；不同之处在于消息的虚拟时间与可观测序列的创建相关，而不在像CreateColdObservable方法一样与订阅相关。

下例与冷序列范例相同，只是创建了一个热可观测序列。

We can also create hot test observable sequences using the CreateHotObservable method. It has the same parameters and return value as CreateColdObservable; the difference is that the virtual time specified for each message is now relative to when the observable was created, not when it is subscribed to as per the CreateColdObservable method.

This example is just that last "cold" sample, but creating a Hot observable instead.

var scheduler = new TestScheduler();

var source = scheduler.CreateHotObservable(

new Recorded<Notification<long>>(10000000, Notification.CreateOnNext(0L)),

...

Output:

Time is 50000000 ticks

Received 5 notifications

OnNext(0) @ 10000000

OnNext(1) @ 20000000

OnNext(2) @ 30000000

OnNext(3) @ 40000000

OnCompleted() @ 40000000

注意输出也是完全一样。创建和订阅的时序对热可观测序列没有影响，因此通知比冷可观测序列早1个时钟周期。

最主要的区别是对热可观测对象传递的虚拟创建时间和虚拟订阅时间不同。对于冷可观察序列，虚拟创建时间没有实际影响，因为订阅的时刻才是动作的起始时刻。即在冷可观测序列上不会丢失任何早期的消息。而对于，热可观测序列，如果订阅太晚会丢失消息。这里，我们理解创建热可观测序列，但一秒后才订阅（因此丢失了第一个消息）。

Note that the output is almost the same. Scheduling of the creation and subscription do not affect the Hot Observable, therefore the notifications happen 1 tick earlier than their Cold counterparts.

We can see the major difference a Hot Observable bears by changing the virtual create time and virtual subscribe time to be different values. With a Cold Observable, the virtual create time has no real impact, as subscription is what initiates any action. This means we can not miss any early message on a Cold Observable. For Hot Observables, we can miss messages if we subscribe too late. Here, we create the Hot Observable immediately, but only subscribe to it after 1 second (thus missing the first message).

var scheduler = new TestScheduler();

var source = scheduler.CreateHotObservable(

new Recorded>Notification>long<<(10000000, Notification.CreateOnNext(0L)),

new Recorded>Notification>long<<(20000000, Notification.CreateOnNext(1L)),

new Recorded>Notification>long<<(30000000, Notification.CreateOnNext(2L)),

new Recorded>Notification>long<<(40000000, Notification.CreateOnNext(3L)),

new Recorded>Notification>long<<(40000000, Notification.CreateOnCompleted>long<())

);

var testObserver = scheduler.Start(

() =< source,

TimeSpan.FromSeconds(1).Ticks,

TimeSpan.FromSeconds(5).Ticks);

Console.WriteLine("Time is {0} ticks", scheduler.Clock);

Console.WriteLine("Received {0} notifications", testObserver.Messages.Count);

foreach (Recorded>Notification>long<< message in testObserver.Messages)

{

Console.WriteLine(" {0} @ {1}", message.Value, message.Time);

}

Output:

Time is 50000000 ticks

Received 4 notifications

OnNext(1) @ 20000000

OnNext(2) @ 30000000

OnNext(3) @ 40000000

OnCompleted() @ 40000000

CreateObserver

最后如果不想使用TestScheduler.Start方法，则需要对观察者进行细粒度的控制，可以使用TestScheduler.CreateObserver()方法。这个方法返回一个ITestObservable对象，可用于管理对可观测序列的订阅。进而，还是没有对记录的消息和订阅者进行隔离。

目前的行业标准要求自动化单元测试的广泛覆盖，以满足质量保证标准。然而，并发编程常常是一个很难测试的领域。Rx提供了设计良好的测试特性实现，允许确定性和高吞吐量测试。TestScheduler提供了一些方法来控制虚拟时间并生成用于测试的可观察序列。这种轻松可靠地测试并发系统的能力使Rx优于许多其他库。

Finally, if you do not want to use the TestScheduler.Start methods, and you need more fine-grained control over your observer, you can use TestScheduler.CreateObserver(). This will return an ITestObserver that you can use to manage the subscriptions to your observable sequences with. Furthermore, you will still be exposed to the recorded messages and any subscribers.

Current industry standards demand broad coverage of automated unit tests to meet quality assurance standards. Concurrent programming, however, is often a difficult area to test well. Rx delivers a well-designed implementation of testing features, allowing deterministic and high-throughput testing. The TestScheduler provides methods to control virtual time and produce observable sequences for testing. This ability to easily and reliably test concurrent systems sets Rx apart from many other libraries.