我在测试使用 Observable.FromAsync<T>()
的类时遇到问题和 Observable.Switch<T>()
.它所做的是等待触发器 observable 产生一个值,然后它开始一个异步操作,最后在一个输出序列中重新收集所有操作的结果。它的要点是这样的:
var outputStream = triggerStream
.Select(_ => Observable
.FromAsync(token => taskProducer.DoSomethingAsync(token)))
.Switch();
我用最少的部分进行了一些健全性检查测试,以了解发生了什么,下面是评论中的测试结果:
class test_with_rx : nspec
{
void Given_async_task_and_switch()
{
Subject<Unit> triggerStream = null;
TaskCompletionSource<long> taskDriver = null;
ITestableObserver<long> testObserver = null;
IDisposable subscription = null;
before = () =>
{
TestScheduler scheduler = new TestScheduler();
testObserver = scheduler.CreateObserver<long>();
triggerStream = new Subject<Unit>();
taskDriver = new TaskCompletionSource<long>();
// build stream under test
IObservable<long> streamUnderTest = triggerStream
.Select(_ => Observable
.FromAsync(token => taskDriver.Task))
.Switch();
/* Also tried with this Switch() overload
IObservable<long> streamUnderTest = triggerStream
.Select(_ => taskDriver.Task)
.Switch(); */
subscription = streamUnderTest.Subscribe(testObserver);
};
context["Before trigger"] = () =>
{
it["Should not notify"] = () => testObserver.Messages.Count.Should().Be(0);
// PASSED
};
context["After trigger"] = () =>
{
before = () => triggerStream.OnNext(Unit.Default);
context["When task completes"] = () =>
{
long result = -1;
before = () =>
{
taskDriver.SetResult(result);
//taskDriver.Task.Wait(); // tried with this too
};
it["Should notify once"] = () => testObserver.Messages.Count.Should().Be(1);
// FAILED: expected 1, actual 0
it["Should notify task result"] = () => testObserver.Messages[0].Value.Value.Should().Be(result);
// FAILED: of course, index out of bound
};
};
after = () =>
{
taskDriver.TrySetCanceled();
taskDriver.Task.Dispose();
subscription.Dispose();
};
}
}
在我也用模拟完成的其他测试中,我可以看到传递给 FromAsync 的 Func 实际上被调用了(例如 taskProducer.DoSomethingAsync(token)
),但随后看起来没有更多内容,并且输出流不产生值(value)。
我也试过插入一些 Task.Delay(x).Wait()
, 或者一些 taskDriver.Task.Wait()
在达到预期之前,但没有运气。
我读了this SO thread我知道调度程序,但乍一看我认为我不需要它们,不需要 ObserveOn()
正在被使用。我错了吗?我错过了什么?助教
为了完整起见,测试框架是NSpec,断言库是FluentAssertions。
最佳答案
您遇到的是一起测试 Rx 和 TPL 的情况。 可以找到详尽的解释 here但我会尝试为您的特定代码提供建议。
基本上您的代码可以正常工作,但您的测试却不行。
Observable.FromAsync
将在提供的任务上转换为 ContinueWith
,该任务将在任务池上执行,因此是异步的。
修复测试的多种方法:(从难看到复杂)
在结果集之后休眠(注意等待不起作用,因为 Wait 不等待继续)
taskDriver.SetResult(result); Thread.Sleep(50);
在执行
FromAsync
之前设置结果(因为如果任务完成,FromAsync 将立即返回一个 IObservable,也就是跳过ContinueWith
)taskDriver.SetResult(result); triggerStream.OnNext(Unit.Default);
用可测试的替代方案替换 FromAsync,例如
public static IObservable<T> ToObservable<T>(Task<T> task, TaskScheduler scheduler) { if (task.IsCompleted) { return task.ToObservable(); } else { AsyncSubject<T> asyncSubject = new AsyncSubject<T>(); task.ContinueWith(t => task.ToObservable().Subscribe(asyncSubject), scheduler); return asyncSubject.AsObservable<T>(); } }
(使用 synchronous TaskScheduler 或 testable 之一)
关于c# - Observable.FromAsync 和 Observable.Switch 的单元测试失败,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31983809/