feat: support testing of `emit.forEach`

[haf]

Description

I'm trying to test how the state changes based on a stream that's listened to via emit.ForEach. However, the stream returned from the process operation is never listened to in the test:

    blocTest(
      'Adding media triggers pipeline events',
      wait: Duration.zero,
      setUp: () async {
        testMedia = await testPhoto(
          'Image.heic',
          basePath: path.normalize(path.join(pwd(), '../../../core/media/test-data')),
        );

        // Simulate pipeline events
        final pipelineEvents = [
          PipelineSuccess(media: testMedia),
        ];

        when(pipeline.process(any)).thenAnswer((_) => Stream.fromIterable(pipelineEvents));
      },
      build: () => bloc,
      act: (EditListingsBloc bloc) => bloc.add(MediaAdded(listingId: testListingId, media: testMedia, isFirstMedia: true)),
      expect: () => [
        // this fails since the stream is never read
        predicate<EditListingsState>((state) {
          return state.listings[testListingId]?.medias?.isEmpty ?? false;
        }),
      ],
    );

Desired Solution

In a test situation, I'd expect testBloc to await all open emitters (the programmer should take care to ensure they close / complete).

Alternatives Considered

I've tried to understand the mechanism by which this does not work; and it would seem that bloc.close() in the testing frameworks, cancels the broadcast stream before it's even consumed.

[felangel]

Hi @haf 👋 Thanks for opening an issue!

You can take a look at the Flutter Todos Example for a reference. If that doesn't help, then feel free to share a link to a minimal reproduction sample and I'm happy to look and open a PR with suggestions.

[haf]

So I'm asking as if: https://github.com/felangel/bloc/blob/8bbd6c4f0c096a4796113de1c97eb43c7efa4f09/examples/flutter_todos/test/todos_overview/bloc/todos_overview_bloc_test.dart#L117-L120 returned a stream — I know it's being called and that's not what I want to test. Makes sense? https://github.com/felangel/bloc/blob/8bbd6c4f0c096a4796113de1c97eb43c7efa4f09/examples/flutter_todos/lib/todos_overview/bloc/todos_overview_bloc.dart#L31 does use forEach, but it doesn't do bloc.add inside a stream listening callback — which is what I'd like to test.

Let's start by understanding each other before either of us invests into building any examples or features :) I'd love to answer any questions and/or discuss as needed!

[haf]

Hope you've had a great weekend! Do you have any further questions that might be clarifying?

[felangel]

Hey @haf, thanks! Hope you had a great weekend as well!

Can you share a snippet of the event handler you want to test? I’m happy to help you write a test 👍

[haf]

Sure; so from one handler I have this code:

  // media
  Future<void> _onMediaAdded(MediaAdded event, Emitter<EditListingsState> emit) async {
    // things here... nextState = state.copyWith ...;
    emit(nextState);

    await emit.forEach(
      pipeline.process(event.media),
      onData: (pe) {
        add(ForwardedPipelineEvent(event: pe));
        return state;
      },
    );
  }

// then later

  Future<void> _onForwardedPipelineEvent(ForwardedPipelineEvent forwarded, Emitter<EditListingsState> emit) async {
     switch (forwarded.event) {
      case final BranchSuccess bsuc:
        // the pro tags should be displayed
        if (bsuc.branchName != proTag) {
          break;
        }
        // nextState = ...

        break;

      case final PipelineSuccess _:
        // Mark the corresponding media as uploaded in the state
        // nextState = ...
        break;

      default:
        break;
    }

    emit(nextState);

    // If the pipeline was successful, save the listing, as we probably have updated it with the new media
    if (forwarded.event is PipelineSuccess) {
      add(Save(listingId: lid, reason: 'onPipelineEvent(PipelineSuccess)'));
    }
  }