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 事件.

示例灵感来自 official doc：

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));
  });
}Ï

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

Dart async* stream breaks randomly in unit tests

stream

dart

flutter-test