单元测试广播流Repo类时测试无限运行的问题排查

问题 #

我想要对一个返回广播流的Repo类进行单元测试。该Repo初始化时会先输出StateOk状态，调用onEvent()方法时会输出StateError状态，代码如下：

class Repo {
  final _controller = StreamController<State>.broadcast();

  Stream<State> get changes async* {
    yield StateOk();
    yield* _controller.stream;
  }

  Future<void> onEvent() async {
    _controller.add(StateError());
  }
}

abstract class State extends Equatable {
  @override
  List<Object?> get props => [];
}
class StateOk extends State {}
class StateError extends State {}

当我运行如下单元测试时：

void main() {
  test('testError', () async {
    final Repo repo = Repo();
    expectLater(
        repo.changes,
        emitsInOrder([
          StateOk(),
          StateError(),
        ]));

    //await Future.delayed(const Duration(milliseconds: 1));
    repo.onEvent();
  });
}

测试会无限运行。如果在调用onEvent()前添加Future.delayed()延迟，测试就能成功完成。我最初以为是时序问题，导致丢失了初始的StateOk元素，所以expectLater无法完成。但如果在emitsInOrder()中移除StateError()的期望，测试无需额外等待就能正常运行，这说明我们并没有丢失初始的StateOk。请问这是什么原因？

原因分析 #

核心问题出在广播流的订阅时机和async*生成器的执行时序上：

• changes是async*生成的流，只有当expectLater开始监听它时，生成器才会启动执行：先yield StateOk()，再去订阅_controller这个广播流。
• 广播流的特性是：只有订阅完成后，发送的事件才会被接收。你直接调用repo.onEvent()的时机，早于_controller.stream被订阅的时刻——因为async*生成器里的yield* _controller.stream是在yield StateOk()之后才执行的，这中间存在微任务调度的间隙。
• 此时StateError()事件被发送给广播流，但expectLater还没完成对_controller.stream的订阅，事件直接丢失，expectLater一直等待未收到的StateError()，导致测试无限挂起。
• 当移除StateError()的期望时，expectLater只需要等待StateOk()，生成器执行到yield StateOk()后就满足了期望，测试直接结束，不需要走到订阅_controller.stream的环节。
• 添加Future.delayed本质是让当前同步代码执行完毕，给async*生成器留出完成_controller.stream订阅的时间，之后发送的事件就能被监听到。

解决方案 #

不需要用Future.delayed，改用微任务调度确保订阅完成后再发送事件，更高效：

void main() {
  test('testError', () async {
    final Repo repo = Repo();
    expectLater(
        repo.changes,
        emitsInOrder([
          StateOk(),
          StateError(),
        ]));

    // 使用微任务确保流订阅完成后再发送事件
    await Future.microtask(() => repo.onEvent());
  });
}

内容的提问来源于stack exchange，提问作者X0Rz1st