Approach for testing streams

[Mike278]

We have a bunch of tests that take the form of: listen to 1 or more streams, dump all events into a log, do some writes, then assert the log looks as we expect. Something like:

final log = [];
getSomeStream().listen(log.add);
await pumpEventQueue();

await someWrite();
await pumpEventQueue();

await someOtherWrite();
await pumpEventQueue();

expect(log, [
  null,
  'foo',
  'bar',
]);

Now that we're using isolates, pumpEventQueue doesn't seem to wait long enough, which I think this makes sense since it's the event loop of the background isolate which still has pending events, not the main isolate where pumpEventQueue is called.

Semantically what we're tying to say is "wait until the stream has had an opportunity to emit (or not emit) before continuing".

What approaches are people taking for writing this kind of test? Using a StreamQueue like you've done here? And how would you test that a stream does not emit, for example because it's using .distinct()?

[simolus3]

wait until the stream has had an opportunity to emit (or not emit) before continuing

I don't know if this is possible when you're on different isolates. You might be able to do this by writing a version of pumpEventQueue() that does Isolate.ping(priority: Isolate.beforeNextEvent) a bunch of times. But I didn't actually try this, it might not be enough.

Using a StreamQueue like you've done here?

If the intention is "wait for a new event, then continue with the rest of the test", I think a StreamQueue is a decent tool.

And how would you test that a stream does not emit, for example because it's using .distinct()?

Again I didn't try this, but it seems like stream matchers like neverEmits also support StreamQueues. Perhaps something like this could work?

final queue = StreamQueue(getSomeStream());
await expectLater(queue, emits(expectedInitialData));

await someWrite();
await expectLater(queue, emits(expectedUpdatedData));

await someWriteCausingADuplicate();
expect(queue, neverEmits(expectedUpdatedData));

// ...

Alternatively, you could probably use something likeStreamQueue(originalStream.doOnData(log.add)). When you want to wait for a new event, you can do it via the stream queue. In cases where you didn't expect a new event, you might be able to do it via the log.

Again, there's a somewhat fundamental issue here: You can't know that a stream is up-to-date and definitely decided to not emit a new event. If none of these options work, I'm open to adding a diagnosis API to drift for this which could check whether a given query stream currently has a pending operation and potentially await it.