c# - RX + TPL + 测试

Question

我目前正在努力解决在 `RX 语句中测试 TPL 函数的问题。问题还在于我无权访问可观察对象。

这是我的测试:

[Test]
public void TestRaceConditions()
{
    // Arrange
    TestScheduler testScheduler = new TestScheduler();
    List<string> source = new List<string> {"Hello"};
    List<string> results = new List<string>();

    source.ToObservable().SelectMany(v => Task.Run(() => 
    {
        Thread.Sleep(10000); // Simulate long running operation
        return v;
    }).ToObservable(testScheduler))
    .ObserveOn(testScheduler)
    .Subscribe(s => results.Add(s));

    // Act
    testScheduler.Start();

    Assert.That(results.Count, Is.EqualTo(1));
    Assert.That(results[0], Is.EqualTo("Hello"));
}

我四处搜索并找到了许多文章和建议，例如:

• 不要混合使用 TPL 和 RX
• 为 TPL 代码创建一个包装器，您可以在其中模拟并返回已完成的任务。 ==> 但是集成测试呢？

我还尝试创建一个自定义的 TaskScheduler，但是我必须用 Task.Factory.StartNew 替换每个 Task.Run >.

那么问题是:

• 如何仅使用我知道何时处理和完成所有元素的 Scheduler 来控制测试中的任务或可观察对象？

更新:真实示例

public class ResultController : IDisposable
{
    private readonly IServiceClient _serviceClient;
    private readonly IDisposable _subscription;

    public ResultController(IEventProvider eventProvider, 
                            IServiceClient serviceClient, 
                            ISchedulerProvider schedulerProvider)
    {
        _serviceClient = serviceClient;
        _subscription = eventProvider.Events.SelectMany(e => LoadDataAsync(e.Id))
                                            .SubscribeOn(schedulerProvider.TaskPool)
                                            .ObserveOn(schedulerProvider.Dispatcher)
                                            .Subscribe(OnNewDataLoaded);
    }

    private void OnNewDataLoaded(Data data)
    {

    }

    public async Task<Data> LoadDataAsync(int id)
    {
        ////return Task.Run(() =>
        ////{
        ////    Thread.Sleep(10000);
        ////    return new Data();
        ////});

        return await _serviceClient.LoadDataByIdAsync(id);
    }

    public void Dispose()
    {
        _subscription.Dispose();
    }
}

此解决方案在生产中没有问题，但在测试中我无法控制任务。

• 模拟 ServiceClient 并返回完成的 Task 工作正常。
• 但是在使用 SpecFlow 的 IntegrationTests 中，如果不引入用于同步的 hacking 属性，我不知道工作何时完成。

Answer 1

我认为这里发生的事情是您在完成工作之前断言工作结果:

// starts a work in a different thread
// and continues the test right after firing work up
testScheduler.Start();

// results here are empty for a 10 seconds
Assert.That(results.Count, Is.EqualTo(1));
Assert.That(results[0], Is.EqualTo("Hello"));

您在这里可以做的是订阅您正在由 Task.Run 安排的任务的延续。 ，您还可以引入一系列任务并将该任务添加到其中并等待它们，然后才进行检查。

引入自定义调度程序可能对您的模拟工作没有帮助 - 您仍然需要等待 10 秒才能唤醒线程。也许另一种选择是使用 Task.Delay方法，它可以被您的测试调度程序忽略。