Unit testing for an event using Reactive Extensions

I'm using Reactive Extensions for .NET (Rx) to expose events as IObservable<T>. I want to create an unit test where I assert that a particular event is fired. Here is a simplified version of the class I want to test:

public sealed class ClassUnderTest : IDisposable {

  Subject<Unit> subject = new Subject<Unit>();

  public IObservable<Unit> SomethingHappened {
    get { return this.subject.AsObservable(); }
  }

  public void DoSomething() {
    this.subject.OnNext(new Unit());
  }

  public void Dispose() {
    this.subject.OnCompleted();
  }

}

Obviously my real classes are more complex. My goal is to verify that performing some actions with the class under test leads to a sequence of events signaled on the IObservable. Luckily the classes I want to test implement IDisposable and calling OnCompleted on the subject when the object is disposed makes it much easier to test.

Here is how I test:

// Arrange
var classUnderTest = new ClassUnderTest();
var eventFired = false;
classUnderTest.SomethingHappened.Subscribe(_ => e开发者_如何学GoventFired = true);

// Act
classUnderTest.DoSomething();

// Assert
Assert.IsTrue(eventFired);

Using a variable to determine if an event is fired isn't too bad, but in more complex scenarios I may want to verify that a particular sequence of events are fired. Is that possible without simply recording the events in variables and then doing assertions on the variables? Being able to use a fluent LINQ-like syntax to make assertions on an IObservable would hopefully make the test more readable.

---

This answer has been updated to the now released version 1.0 of Rx.

Official documentation is still scant but Testing and Debugging Observable Sequences on MSDN is a good starting place.

The test class should derive from ReactiveTest in the Microsoft.Reactive.Testing namespace. The test is based around a TestScheduler that provides virtual time for the test.

The TestScheduler.Schedule method can be used to queue up activities at certain points (ticks) in virtual time. The test is executed by TestScheduler.Start. This will return an ITestableObserver<T> that can be used for asserting for instance by using the ReactiveAssert class.

public class Fixture : ReactiveTest {

  public void SomethingHappenedTest() {
    // Arrange 
    var scheduler = new TestScheduler();
    var classUnderTest = new ClassUnderTest();

    // Act 
    scheduler.Schedule(TimeSpan.FromTicks(20), () => classUnderTest.DoSomething());
    var actual = scheduler.Start(
      () => classUnderTest.SomethingHappened,
      created: 0,
      subscribed: 10,
      disposed: 100
    );

    // Assert
    var expected = new[] { OnNext(20, new Unit()) };
    ReactiveAssert.AreElementsEqual(expected, actual.Messages);
  }

}

TestScheduler.Schedule is used to schedule a call to DoSomething at time 20 (measured in ticks).

Then TestScheduler.Start is used to perform the actual test on the observable SomethingHappened. The lifetime of the subscription is controlled by the arguments to the call (again measured in ticks).

Finally ReactiveAssert.AreElementsEqual is used to verify that OnNext was called at time 20 as expected.

The test verifies that calling DoSomething immediately fires the observable SomethingHappened.

---

This kind of testing for observables would be incomplete. Just recently, RX team published the test scheduler and some extensions (which BTW they use internally for testing the library). Using these, you can not only check if something happened or not, but also make sure the timing and order is correct. As a bonus the test scheduler allows you to run your tests in "virtual time", so the tests run instantly, no matter how big delays you're using inside.

Jeffrey van Gogh from RX team published an article on how to do such kind of testing.

The above test, using the mentioned approach, will look like this:

[TestMethod]
public void SimpleTest()
{
    var sched = new TestScheduler();
    var subject = new Subject<Unit>();
    var observable = subject.AsObservable();

    var o = sched.CreateHotObservable(OnNext(210, new Unit()), OnCompleted<Unit>(250));
    var results = sched.Run(() =>
    {
        o.Subscribe(subject);
        return observable;
    });

    results.AssertEqual(OnNext(210, new Unit()), OnCompleted<Unit>(250));
}

EDIT: You can also call .OnNext (or some other method) implicitly:

        var o = sched.CreateHotObservable(OnNext(210, new Unit()));
        var results = sched.Run(() =>
        {
            o.Subscribe(_ => subject.OnNext(new Unit()));
            return observable;
        });
        results.AssertEqual(OnNext(210, new Unit()));

My point is - in simplest situations, you need just to make sure the event is fired (f.e. you are checking your Where is working correctly). But then as you progress in complexity, you start testing for timing, or completion, or something else that requires the virtual scheduler. But the nature of the test using virtual scheduler, opposite to "normal" tests is to test the entire observable at once, not "atomic" operations.

So probably you would have to switch to the virtual scheduler somewhere down the road - why not start with it in the beginning?

P.S. Also, you would have to resort to a different logic for each test case - f.e. you would have very different observable for testing that something did not happen opposite to that something happened.

---

Not sure about more fluent but this will do the trick without introducing a variable.

var subject = new Subject<Unit>();
subject
    .AsObservable()
    .Materialize()
    .Take(1)
    .Where(n => n.Kind == NotificationKind.OnCompleted)
    .Subscribe(_ => Assert.Fail());

subject.OnNext(new Unit());
subject.OnCompleted();