Add stream matchers.

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## 0.12.19
 
+* Added the `StreamMatcher` class, as well as several built-in stream matchers:
+  `emits()`, `emitsError()`, `emitsDone, mayEmit()`, `mayEmitMultiple()`,
+  `emitsAnyOf()`, `emitsInOrder()`, `emitsInAnyOrder()`, and `neverEmits()`.
+
 * `expect()` now returns a Future for the asynchronous matchers `completes`,
   `completion()`, `throws*()`, and `prints()`.
 

README.md

@@ -6,6 +6,7 @@
   * [Platform Selectors](#platform-selectors)
   * [Running Tests on Dartium](#running-tests-on-dartium)
 * [Asynchronous Tests](#asynchronous-tests)
+  * [Stream Matchers](#stream-matchers)
 * [Running Tests With Custom HTML](#running-tests-with-custom-html)
 * [Configuring Tests](#configuring-tests)
   * [Skipping Tests](#skipping-tests)
@@ -14,7 +15,7 @@
   * [Whole-Package Configuration](#whole-package-configuration)
 * [Tagging Tests](#tagging-tests)
 * [Debugging](#debugging)
-* [Browser/VM Hybrid Tests](#browser-vm-hybrid-tests)
+* [Browser/VM Hybrid Tests](#browservm-hybrid-tests)
 * [Support for Other Packages](#support-for-other-packages)
   * [`term_glyph`](#term_glyph)
   * [`barback`](#barback)
@@ -355,6 +356,119 @@ void main() {
 
 [expectAsync]: http://www.dartdocs.org/documentation/test/latest/index.html#test/test@id_expectAsync
 
+### Stream Matchers
+
+The `test` package provides a suite of powerful matchers for dealing with
+[asynchronous streams][Stream]. They're expressive and composable, and make it
+easy to write complex expectations about the values emitted by a stream. For
+example:
+
+[Stream]: https://api.dartlang.org/stable/dart-async/Stream-class.html
+
+```dart
+import "dart:async";
+
+import "package:test/test.dart";
+
+void main() {
+  test("process emits status messages", () {
+    // Dummy data to mimic something that might be emitted by a process.
+    var stdoutLines = new Stream.fromIterable([
+      "Ready.",
+      "Loading took 150ms.",
+      "Succeeded!"
+    ]);
+
+    expect(stdoutLines, emitsInOrder([
+      // Values match individual events.
+      "Ready.",
+
+      // Matchers also run against individual events.
+      startsWith("Loading took"),
+
+      // Stream matchers can be nested. This asserts that one of two events are
+      // emitted after the "Loading took" line.
+      emitsAnyOf(["Succeeded!", "Failed!"]),
+
+      // By default, more events are allowed after the matcher finishes
+      // matching. This asserts instead that the stream emits a done event and
+      // nothing else.
+      emitsDone
+    ]));
+  });
+}
+```
+
+A stream matcher can also match the [`async`][async] package's
+[`StreamQueue`][StreamQueue] class, which allows events to be requested from a
+stream rather than pushed to the consumer. The matcher will consume the matched
+events, but leave the rest of the queue alone so that it can still be used by
+the test, unlike a normal `Stream` which can only have one subscriber. For
+example:
+
+[async]: https://pub.dartlang.org/packages/async
+[StreamQueue]: https://www.dartdocs.org/documentation/async/latest/async/StreamQueue-class.html
+
+```dart
+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 = new StreamQueue(new Stream.fromIterable([
+      "WebSocket URL:",
+      "ws://localhost:1234/",
+      "Waiting for connection..."
+    ]));
+
+    // Ignore lines from the process until it's about to emit the URL.
+    await expect(stdout, emitsThrough("WebSocket URL:"));
+
+    // Parse the next line as a URL.
+    var url = Uri.parse(await stdout.next);
+    expect(url.host, equals('localhost'));
+
+    // You can match against the same StreamQueue multiple times.
+    await expect(stdout, emits("Waiting for connection..."));
+  });
+}
+```
+
+The following built-in stream matchers are available:
+
+* [`emits()`][emits] matches a single data event.
+* [`emitsError()`][emitsError] matches a single error event.
+* [`emitsDone`][emitsDone] matches a single done event.
+* [`mayEmit()`][mayEmit] consumes events if they match an inner matcher, without
+  requiring them to match.
+* [`mayEmitMultiple()`][mayEmitMultiple] works like `mayEmit()`, but it matches
+  events against the matcher as many times as possible.
+* [`emitsAnyOf()`][emitsAnyOf] consumes events matching one (or more) of several
+  possible matchers.
+* [`emitsInOrder()`][emitsInOrder] consumes events matching multiple matchers in
+  a row.
+* [`emitsInAnyOrder()`][emitsInAnyOrder] works like `emitsInOrder()`, but it
+  allows the matchers to match in any order.
+* [`neverEmits()`][neverEmits] matches a stream that finishes *without* matching
+  an inner matcher.
+
+You can also define your own custom stream matchers by calling
+[`new StreamMatcher()`][new StreamMatcher].
+
+[emits]: https://www.dartdocs.org/documentation/test/latest/test/emits.html
+[emitsError]: https://www.dartdocs.org/documentation/test/latest/test/emitsError.html
+[emitsDone]: https://www.dartdocs.org/documentation/test/latest/test/emitsDone.html
+[mayEmit]: https://www.dartdocs.org/documentation/test/latest/test/mayEmit.html
+[mayEmitMultiple]: https://www.dartdocs.org/documentation/test/latest/test/mayEmitMultiple.html
+[emitsAnyOf]: https://www.dartdocs.org/documentation/test/latest/test/emitsAnyOf.html
+[emitsInOrder]: https://www.dartdocs.org/documentation/test/latest/test/emitsInOrder.html
+[emitsInAnyOrder]: https://www.dartdocs.org/documentation/test/latest/test/emitsInAnyOrder.html
+[neverEmits]: https://www.dartdocs.org/documentation/test/latest/test/neverEmits.html
+[new StreamMatcher]: https://www.dartdocs.org/documentation/test/latest/test/StreamMatcher/StreamMatcher.html
+
 ## Running Tests With Custom HTML
 
 By default, the test runner will generate its own empty HTML file for browser
@@ -626,7 +740,7 @@ example:
 
 [spawnHybridCode]: http://www.dartdocs.org/documentation/test/latest/index.html#test/test@id_spawnHybridCode
 [spawnHybridUri]: http://www.dartdocs.org/documentation/test/latest/index.html#test/test@id_spawnHybridUri
-[dart:isolate]: https://api.dartlang.org/stable/latest/dart-isolate/dart-isolate-library.html
+[dart:isolate]: https://api.dartlang.org/stable/dart-isolate/dart-isolate-library.html
 [StreamChannel]: https://pub.dartlang.org/packages/stream_channel
 
 ```dart

lib/src/frontend/stream_matcher.dart

@@ -0,0 +1,187 @@
+// Copyright (c) 2017, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:async/async.dart';
+import 'package:matcher/matcher.dart';
+
+import '../utils.dart';
+import 'async_matcher.dart';
+
+/// The type for [_StreamMatcher._matchQueue].
+typedef Future<String> _MatchQueue(StreamQueue queue);
+
+/// A matcher that matches events from [Stream]s or [StreamQueue]s.
+///
+/// Stream matchers are designed to make it straightforward to create complex
+/// expectations for streams, and to interleave expectations with the rest of a
+/// test. They can be used on a [Stream] to match all events it emits:
+///
+/// ```dart
+/// expect(stream, emitsInOrder([
+///   // Values match individual events.
+///   "Ready.",
+///
+///   // Matchers also run against individual events.
+///   startsWith("Loading took"),
+///
+///   // Stream matchers can be nested. This asserts that one of two events are
+///   // emitted after the "Loading took" line.
+///   emitsAnyOf(["Succeeded!", "Failed!"]),
+///
+///   // By default, more events are allowed after the matcher finishes
+///   // matching. This asserts instead that the stream emits a done event and
+///   // nothing else.
+///   emitsDone
+/// ]));
+/// ```
+///
+/// It can also match a [StreamQueue], in which case it consumes the matched
+/// events. The call to [expect] returns a [Future] that completes when the
+/// matcher is done matching. You can `await` this to consume different events
+/// at different times:
+///
+/// ```dart
+/// var stdout = new StreamQueue(stdoutLineStream);
+///
+/// // Ignore lines from the process until it's about to emit the URL.
+/// await expect(stdout, emitsThrough("WebSocket URL:"));
+///
+/// // Parse the next line as a URL.
+/// var url = Uri.parse(await stdout.next);
+/// expect(url.host, equals('localhost'));
+///
+/// // You can match against the same StreamQueue multiple times.
+/// await expect(stdout, emits("Waiting for connection..."));
+/// ```
+///
+/// Users can call [new StreamMatcher] to create custom matchers.
+abstract class StreamMatcher extends Matcher {
+  /// The description of this matcher.
+  ///
+  /// This is in the subjunctive mood, which means it can be used after the word
+  /// "should". For example, it might be "emit the right events".
+  String get description;
+
+  /// Creates a new [StreamMatcher] described by [description] that matches
+  /// events with [matchQueue].
+  ///
+  /// The [matchQueue] callback is used to implement [StreamMatcher.matchQueue],
+  /// and should follow all the guarantees of that method. In particular:
+  ///
+  /// * If it matches successfully, it should return `null` and possibly consume
+  ///   events.
+  /// * If it fails to match, consume no events and return a description of the
+  ///   failure.
+  /// * The description should be in past tense.
+  /// * The description should be gramatically valid when used after "the
+  ///   stream"—"emitted the wrong events", for example.
+  ///
+  /// The [matchQueue] callback may return the empty string to indicate a
+  /// failure if it has no information to add beyond the description of the
+  /// failure and the events actually emitted by the stream.
+  ///
+  /// The [description] should be in the subjunctive mood. This means that it
+  /// should be grammatically valid when used after the word "should". For
+  /// example, it might be "emit the right events".
+  factory StreamMatcher(
+          Future<String> matchQueue(StreamQueue queue),
+          String description)
+      = _StreamMatcher;
+
+  /// Tries to match events emitted by [queue].
+  ///
+  /// If this matches successfully, it consumes the matching events from [queue]
+  /// and returns `null`.
+  ///
+  /// If this fails to match, it doesn't consume any events and returns a
+  /// description of the failure. This description is in the past tense, and
+  /// could grammatically be used after "the stream". For example, it might
+  /// return "emitted the wrong events".
+  ///
+  /// The description string may also be empty, which indicates that the
+  /// matcher's description and the events actually emitted by the stream are
+  /// enough to understand the failure.
+  ///
+  /// If the queue emits an error, that error is re-thrown unless otherwise
+  /// indicated by the matcher.
+  Future<String> matchQueue(StreamQueue queue);
+}
+
+/// A concrete implementation of [StreamMatcher].
+///
+/// This is separate from the original type to hide the private [AsyncMatcher]
+/// interface.
+class _StreamMatcher extends AsyncMatcher implements StreamMatcher {
+  final String description;
+
+  /// The callback used to implement [matchQueue].
+  final _MatchQueue _matchQueue;
+
+  _StreamMatcher(this._matchQueue, this.description);
+
+  Future<String> matchQueue(StreamQueue queue) => _matchQueue(queue);
+
+  /*FutureOr<String>*/ matchAsync(item) {
+    StreamQueue queue;
+    if (item is StreamQueue) {
+      queue = item;
+    } else if (item is Stream) {
+      queue = new StreamQueue(item);
+    } else {
+      return "was not a Stream or a StreamQueue";
+    }
+
+    // Avoid async/await in the outer method so that we synchronously error out
+    // for an invalid argument type.
+    var transaction = queue.startTransaction();
+    var copy = transaction.newQueue();
+    return matchQueue(copy).then((result) async {
+      // Accept the transaction if the result is null, indicating that the match
+      // succeeded.
+      if (result == null) {
+        transaction.commit(copy);
+        return null;
+      }
+
+      // Get a list of events emitted by the stream so we can emit them as part
+      // of the error message.
+      var replay = transaction.newQueue();
+      var events = <Result>[];
+      var subscription = Result.captureStreamTransformer.bind(replay.rest)
+          .listen(events.add, onDone: () => events.add(null));
+
+      // Wait on a timer tick so all buffered events are emitted.
+      await new Future.delayed(Duration.ZERO);
+      subscription.cancel();
+
+      var eventsString = events.map((event) {
+        if (event == null) {
+          return "x Stream closed.";
+        } else if (event.isValue) {
+          return addBullet(event.asValue.value.toString());
+        } else {
+          var error = event.asError;
+          var text = "${error.error}\n${testChain(error.stackTrace)}";
+          return prefixLines(text, "  ", first: "! ");
+        }
+      }).join("\n");
+      if (eventsString.isEmpty) eventsString = "no events";
+
+      transaction.reject();
+
+      var buffer = new StringBuffer();
+      buffer.writeln(indent(eventsString,                  first: "emitted "));
+      if (result.isNotEmpty) buffer.writeln(indent(result, first: "  which "));
+      return buffer.toString().trimRight();
+    }, onError: (error) {
+      transaction.reject();
+      throw error;
+    });
+  }
+
+  Description describe(Description description) =>
+      description.add("should ").add(this.description);
+}