如何对 Rx.Net foreach 订阅进行单元测试答案

【问题标题】：How to unit test Rx.Net foreach subscription如何对 Rx.Net foreach 订阅进行单元测试
【发布时间】：2017-11-11 14:35:00
【问题描述】：

所有消息都应该发布到消息总线：

 upstream.Get().ForEachAsync(async e => await _bus.Publish(e, cancellationToken));

我想进行单元测试以验证发布方法是否被正确调用：

    [Theory, AutoMoqData]
    public void Publish_ShouldPublishAllDataSources(
        [Frozen]Mock<IBus> bus,
        [Frozen]Mock<IUpstream> upstream,
        SUT sut,
        TestScheduler scheduler,
        CancellationToken cancellationToken
        )
    {
        {
            Arrange();
            Act();
            Assert();
        }
        ITestableObservable<Event> inputObservable;

        void Arrange()
        {
            inputObservable =
                    scheduler.CreateColdObservable(
                        OnNext(1L,new Event("e1")),
                        OnNext(1L,new Event("e2")),
                        OnNext(1L,new Event("e3"))
                    );
            upstream
            .Setup(c => c.Get())
            .Returns(inputObservable);


        }

        void Act()
        {
            sut.Exercise(cancellationToken);
            //Do sth with scheduller to fire all events 
            //I've tested these:
            //scheduler.Start();
            //scheduler.AdvanceTo(30000L);
        }


        void Assert()
        {
            inputObservable.ForEach(e =>
                bus.Verify(b => 
                    b.Publish(e, cancellationToken)
                ));
        }
    }

但无论我尝试什么，测试都不会停止。

更新：

这是完整的生产代码：

class SUT 
{
     IUpstream _upstream;
     public SUT(IUpstream upstream)
     {
          _upstream = upstream;
     }
     void Exercise(CancellationToken cancellationToken)
     {
         _upstream .Get()
               .ForEachAsync(async 
                    e => await _bus.Publish(e, cancellationToken));
     }
}


interface IUpstream
{
    IObservable<string> Get();
}

interface IBus
{
    void Publish<T>(T,System.Threading.CancellationToken)
}

【问题讨论】：

您确实需要为此提供minimal reproducible example。我们可以运行一些代码来查看行为。最好只是作为一个简单的控制台应用程序。
感谢您的更新，但您是否尝试创建一个可以在控制台应用程序中运行的 minimal reproducible example？
我不知道您所说的“在控制台应用程序中运行代码的测试工具需要更多的仪式”是什么意思。这对我来说没有意义。
@Enigmativity：感谢您的建议。我更新了生产代码。我没有所有的实现。我的问题是关于 RX.net 测试工具的。在控制台应用程序中运行代码需要更多的仪式，实现接口只是其中的一小部分。
@Enigmativity：我为这个错字道歉。我在外面用平板电脑打字。

标签： c# unit-testing system.reactive rx.net

【解决方案1】：

ITestableObservable<T>.Messages 应该用于断言而不是ITestableObservable<T> 本身。

    void Assert()
    {
        inputObservable
                .Messages.ToList()
                .ForEach(e =>
                bus.Verify(b =>
                    b.Publish(e.Value.Value, cancellationToken)));
    }

所以动作应该是这样的：

    void Act()
    {
        sut.Exercise(cancellationToken);
        scheduler.Start();
        //OR:
        //scheduler.AdvanceBy(1L);
    }

【讨论】：