使用 Rx.NET 测试 IObservable 是否在特定时间段内发出或未发出值

Question

我目前正在重构一个程序以使用 Membus 消息总线进行事件聚合，并且此消息总线允许我通过返回可以订阅的 IObservable 实例来“观察”总线上的事件到。

在我的单元测试中，我想确保我的组件只在适当的时候在总线上发布特定的消息。我尝试这样做的方法是在我的测试规范类中包含以下类型的设置逻辑：

private readonly IBus messageBus;
private readonly IObservable<Model> myObservable;

public ComponentModelGatewaySpec()
{
    messageBus = TestHelper.DefaultMessageBus;
    myObservable = messageBus.Observe<ModelPublishedEventMessage>().Select(it => it.Model);
}

然后在一个测试用例中，我想做如下的事情：

public async Task Only_Publish_Incomplete_Models_After_Receiving_Request()
{
    var defaultTimeout = TimeSpan.FromMilliseconds(1000);

    // GIVEN a component model gateway and an incomplete model update.
    var modelUpdate = new Model { IntProperty = 1, BoolProperty = null };
    Assert.False(modelUpdate.IsComplete);
    var gateway = MockComponentModelGateway;
    gateway.SetMessageBus(messageBus);

    // EXPECT no current model is published after publishing the incomplete model update.
    Task<bool> noModelPublished = myObservable.WithinWindow(defaultTimeout).NoEmissionsOccurred().ToTask();
    messageBus.Publish(new ModelUpdateEventMessage(modelUpdate));
    Assert.True(await noModelPublished);

    // WHEN we publish a current model query.
    Task<Model> publishedModel = myObservable.WithinWindow(defaultTimeout).FirstAsync().ToTask();
    messageBus.Publish(new ModelQueryRequestedEventMessage());

    // THEN the model should be published.
    Assert.Equal(modelUpdate, await publishedModel);
}

我主要追求的是一种测试方式：

• “在我表演后没有发布此特定类型的事件 这（一系列）行动，至少在 X 时间内。”
• “在我执行此（一系列）操作后发布了此特定类型的事件，这就是我期望该事件的属性。”

我希望能够异步处理所有这些逻辑，否则我会有一堆测试用例最终阻塞 1 秒或更长时间。

也许可以为此使用Timeout，但Timeout 会导致在超时时引发异常，这似乎是我期望它们被抛出时处理事情的一种笨拙的方式。我确实在可观察的组合中使用Timeout，但仅在发生超时意味着测试应该失败的情况下。

目前，我正在尝试使用 Window、Buffer、FirstAsync 等的各种组合来完成此操作，但在所有测试用例中都没有得到我期望的行为。

编辑

我已经添加了自己的解决方案，但我将其视为临时措施，直到我能够采纳 Lee Campbell 的建议（请参阅下面的答案）。

Answer 1

如果可以，您希望在单元测试中避免并发（以及多线程）。并发单元测试可能是不确定的，并且运行速度也慢得多，即必须实时运行。例如，如果您试图证明 10 秒的超时会引发错误，那么您必须让测试运行 10 秒。这不是可扩展的做法。

请考虑使用TestScheduler。 这意味着您将需要有 seam，您可以在其中为您的操作员提供调度程序。 希望公开这些 Observable 序列的 API 对测试友好。

public async Task Only_Publish_Incomplete_Models_After_Receiving_Request()
{
    var gateway = MockComponentModelGateway;
    gateway.SetMessageBus(messageBus);


    var defaultTimeout = TimeSpan.FromMilliseconds(1000);
    var scheduler = new TestScheduler();


    // GIVEN a component model gateway and an incomplete model update.
    var modelUpdate = new Model { IntProperty = 1, BoolProperty = null };
    Assert.False(modelUpdate.IsComplete);

    scheduler.Schedule(TimeSpan.FromMilliseconds(100),() => {
        messageBus.Publish(new ModelUpdateEventMessage(modelUpdate));
    });

    scheduler.Schedule(TimeSpan.FromMilliseconds(200),() =>
    {
        messageBus.Publish(new ModelQueryRequestedEventMessage());
    });

    var observer = scheduler.CreateObserver<Model>();

    myObservable.Subscribe(observer);

    scheduler.Start();

    CollectionAssert.AreEqual(
        new[]{
            ReactiveTest.OnNext(TimeSpan.FromMilliseconds(200).Ticks, modelUpdate)
        },
        observer.Messages);
}

在这里您不必测试是否缺席 (Assert.True(await noModelPublished);)，因为您可以在输出中看到，直到执行 messageBus.Publish(new ModelQueryRequestedEventMessage()); 的虚拟时间点（200 毫秒）才推送值。

现在您的测试应该同步运行，但能够验证其他异步流程。

Answer 2

让我先说我在根据 Lee Campbell 的建议采取行动之前开发了这个解决方案，并且基于他的建议的解决方案可能会好得多，因为他（字面上）写了关于这个主题的书。也就是说，我提出的解决方案对于我的特定用例来说效果很好。

使用我在原始帖子中使用的相同示例测试用例，我现在有了这个：

[Theory]
[PairwiseData]
public async Task Adapted_Component_Model_Gateway_Should_Publish_Current_Model_When_Requested(
    [CombinatorialValues(null, 1, 2)] int? intValue,
    [CombinatorialValues(null, true, false)] bool? boolValue)
{
    var model = new AllowStaleDetailsMockModel { IntProperty = intValue, BoolProperty = boolValue };
    if (model.IsComplete) return;

    // GIVEN an initialized adapted component model gateway and a given _INCOMPLETE_ current model.
    var adaptedGateway = AdaptGateway(MockComponentModelGateway);
    adaptedGateway.SetMessageBus(messageBus);
    adaptedGateway.Initialize();

    // EXPECT no current model is published after publishing the incomplete model update.
    var messagePublished = allowStaleCurrentModelObservable.BufferEmissions().ContainsEvents();
    messageBus.Publish(new CurrentModelUpdateReadyEventArgs<AllowStaleDetailsMockModel>(model));
    Assert.False(await messagePublished);

    // WHEN we publish a current model query.
    var actualModel = allowStaleCurrentModelObservable.WaitForEmission();
    messageBus.Publish(new CurrentModelQueryRequestedEventArgs());

    // THEN the current model should be published.
    Assert.Equal(model, await actualModel);
}

在“测试实用程序”类中，我创建了以下内容：

public static class TestHelper
{
    public static readonly TimeSpan DefaultTimeout = TimeSpan.FromMilliseconds(1000);

    public static IBus DefaultMessageBus
    {
        get
        {
            return BusSetup.StartWith<Conservative>().Construct();
        }
    }

    public static async Task<bool> ContainsEvents<T>(this Task<IList<T>> eventBufferTask)
    {
        return (await eventBufferTask).Any();
    }
}

public static class ObservableExtensions
{
    public static Task<T> WaitForEmission<T>(this IObservable<T> observable)
    {
        return observable.WaitForEmission(TestHelper.DefaultTimeout);
    }

    public static Task<T> WaitForEmission<T>(this IObservable<T> observable, TimeSpan timeout)
    {
        return observable.FirstAsync().Timeout(timeout).ToTask();
    }

    public static Task<IList<T>> BufferEmissions<T>(this IObservable<T> observable)
    {
        return observable.BufferEmissions(TestHelper.DefaultTimeout);
    }

    public static Task<IList<T>> BufferEmissions<T>(this IObservable<T> observable, TimeSpan bufferWindow)
    {
        return observable.Buffer(bufferWindow).FirstAsync().ToTask();
    }
}

我在测试用例中使用Observable.WaitForEmissions，然后再做一些我期望应该发布特定类型消息的事情。这将返回一个任务：

• 返回 observable 发出的单个值
• 如果在给定的时间长度（默认为 1 秒）内未检测到发射，则会引发超时错误。

我在测试用例中使用Observable.BufferEmissions，我希望发布多个值并且我想将它们全部收集起来，或者我想检查值是否在给定的时间长度内发布而不捕获@ 987654325@ 错误（Task<IList<T>>.ContainsEvents 非常适合此问题）。

我项目中的所有测试用例都按预期执行，大约 600 多个测试用例在大约 30 秒内被发现并执行，这让我很满意。