Dart-async*流在单元测试中随机中断

Question

一个简单的用例，我从服务器中获取数据列表。我想把它转换成一个流。在此之前，我想在单元测试中测试事情是否正常。

环境

• Flutter(通道稳定，1.22.4，在Linux上(
• Dart SDK版本：2.10.4(稳定(

下面是合并在一个文件中的简化代码。

import 'package:flutter_test/flutter_test.dart';
class BeanList<T> {
final List<T> list;
final int count;
BeanList({this.list, this.count});
}
abstract class PageProvider<T> {
Future<BeanList<T>> provide();
}
class IntPageProvider implements PageProvider<int> {
final size = 5;
final count = 9;
int offset = 0;
List<int> _generateList(int startFrom, int size) {
List<int> list = [];
for (int i = startFrom; i < startFrom + size; i++) {
list.add(i);
}
return list;
}
@override
Future<BeanList<int>> provide() async {
final list = _generateList(offset, size);
print('provide: $list');
offset += size;
return BeanList<int>(list: list, count: count);
}
}
Stream<T> itemStream<T>(PageProvider<T> provider) async* {
int total;
int sent = 0;
bool done = false;
while (true) {
final BeanList<T> beanList = await provider.provide();
final list = beanList.list;
final count = beanList.count;
total = beanList.count;
print('Got count:$count, list:$list');
for (final item in list) {
print('yield $item');
yield item;
sent++;
done = sent == total;
if (done) break;
}
if (done) break;
}
}
void main() {
test('Randomly breaking stream', () {
itemStream(IntPageProvider()).listen(print);
});
}

预期输出我应该得到数字0到8，我有时会这样做，但大多数时候测试会在任何随机位置停止。

Answer 1

感谢Christopher Moore抽出时间，我发现了问题。

问题是我如何在测试中使用流。

目前是这样做的。

test('Randomly breaking stream', () {
itemStream(IntPageProvider()).listen(print);
});

Stream.listen()在测试中异步运行，没有任何东西可以让测试等待流被完全消耗。因此，测试结束，测试中的所有过程都会突然停止，从而成为随机停止点。

但是，如果我们对测试进行一些更改。

test('Test waiting for the stream consumption', () async {
await itemStream(IntPageProvider()).listen(print).asFuture();
});

这样可以确保测试在流被完全消耗之前不会结束。

我希望这将帮助任何陷入同样问题的人。

Answer 2

确保流被完全消耗的另一种方法是将流包装在StreamQueue(来自async包(中，这样流就可以有多个订阅者，然后等待emitsDone事件。

受官方文档启发的示例：

import 'dart:async';
import 'package:async/async.dart';
import 'package:test/test.dart';
void main() {
test('process emits a WebSocket URL', () async {
// Wrap the Stream in a StreamQueue so that we can request events.
var stdout = StreamQueue(Stream.fromIterable([
'WebSocket URL:',
'ws://localhost:1234/',
'Waiting for connection...'
]));
// Ignore lines from the process until it's about to emit the URL.
await expectLater(stdout, emitsThrough('WebSocket URL:'));

await expectLater(stdout, emitsThrough(emitsDone));
});
}Ï