How to unit test async function call inside stream listener

Consider the following code that needs to be unit tested

void run() {
    _activityRepo.activityUpdateStream.listen((token) async {
      await _userRepo.updateToken(token: token);
    });
  }

where _activityRepo.activityUpdateStream is a Stream<String> that emits String events.

The goal here is to test that updateToken function is called every time _activityRepo.activityUpdateStream emits a new event. The unit test is currently like:

test('should update notification token when refreshed', () async {
    const fakePushToken = 'token';
    final fakeStream = StreamController<String>();

    when(() => liveActivityRepo.activityUpdateStream).thenAnswer((_) => fakeStream.stream);

    useCase.run();

    fakeStream.add(fakeUpdate);

    verify(() => userRepo.updateLiveActivityToken(token: fakePushToken)).called(1);
  });

This test fails because verify is being called before updateToken function inside the stream listener. How should I rewrite this test so I can make sure the stream listener is fully executed before running the verification step in the test?

This unit test fails because, as you explained, verify is executed before the asynchronous operations of the stream listener.

Those asynchronous operations may take some time, and therefore depend on some kind of timers (including those that handle timeouts). Think remote API calls. Those operations are put in something called the Event Queue because their completion depend on some external events happening.

But they may just be some delayed operations that complete quickly, called microtasks. They can be confusing because they look like synchronous operations and we can expect them to complete synchronously. They are put in something called the Microtask Queue.

I don't know what kind of async task your updateToken performs, therefore I'm giving you both solutions.

I made this example:

import 'package:fake_async/fake_async.dart';
import 'package:flutter/foundation.dart';
import 'package:flutter_test/flutter_test.dart';

main() {
  test("Sync with timers", () {
    int n = 0;
    var stream = Stream.fromIterable([1, 2, 3, 4, 5]);
    stream.listen((value) async {
      await Future.delayed(Duration(seconds: value));
      n = value;
      debugPrint(value.toString());
    });
    expect(n, 0);
  });

  test("Sync with microtasks", () {
    int n = 0;
    var stream = Stream.fromIterable([1, 2, 3, 4, 5]);
    stream.listen((value) async {
      n = value;
      debugPrint(value.toString());
    });
    expect(n, 0);
  });

  test("Fake async with timers", () {
    fakeAsync((fa) {
      int n = 0;
      var stream = Stream.fromIterable([1, 2, 3, 4, 5]);
      stream.listen((value) async {
        await Future.delayed(Duration(seconds: value));
        n = value;
        debugPrint(value.toString());
      });
      fa.flushTimers();
      expect(n, 5);
    });
  });

  test("Fake async with microtasks", () {
    fakeAsync((fa) {
      int n = 0;
      var stream = Stream.fromIterable([1, 2, 3, 4, 5]);
      stream.listen((value) async {
        n = value;
        debugPrint(value.toString());
      });
      fa.flushMicrotasks();
      expect(n, 5);
    });
  });
}

The first two tests show what we expect to happen in a synchronous unit test, like the one you did. n is not going to increase before expect is called. The second test is the most surprising, because even if our "async" task is actually synchronous (it just assigns a value and prints it), it only runs after the sync statements, including expect.

By using the fake_async package, you can manipulate the Event Queue and the Microtask Queue. Your unit test is nested into a fakeAsync call, and by calling flushTimers or flushMicrotasks you can complete all the async operations of the stream listener.

Notice that the fakeAsync tests would fail if you swapped the calls to flushTimers and flushMicrotasks, which shows that which of them you need to call depends on the concrete nature of the async tasks.