pkgs/checks/lib/src/extensions/async.dart

// Copyright (c) 2022, 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 'dart:convert';

import 'package:async/async.dart';

import '../../context.dart';

extension FutureChecks<T> on Subject<Future<T>> {
  /// Expects that the `Future` completes to a value without throwing.
  ///
  /// Fails if the future completes as an error.
  ///
  /// Pass [completionCondition] to check expectations on the completion result.
  ///
  /// The returned future will complete when the subject future has completed,
  /// and [completionCondition] has optionally been checked.
  Future<void> completes([AsyncCondition<T>? completionCondition]) {
    return context.nestAsync<T>(() => ['completes to a value'], (actual) async {
      try {
        return Extracted.value(await actual);
      } catch (e, st) {
        return Extracted.rejection(
          actual: ['a future that completes as an error'],
          which: [
            ...prefixFirst('threw ', postfixLast(' at:', literal(e))),
            ...indent(LineSplitter.split(st.toString())),
          ],
        );
      }
    }, completionCondition);
  }

  /// Expects that the `Future` never completes as a value or an error.
  ///
  /// Immediately returns and does not cause the test to remain running if it
  /// ends.
  /// If the future completes at any time, raises a test failure. This may
  /// happen after the test has already appeared to succeed.
  ///
  /// Not compatible with [softCheck] or [softCheckAsync] since there is no
  /// concrete end point where this condition has definitely succeeded.
  void doesNotComplete() {
    context.expectUnawaited(() => ['does not complete'], (actual, reject) {
      unawaited(
        actual.then(
          (r) {
            reject(
              Rejection(
                actual: prefixFirst('a future that completed to ', literal(r)),
              ),
            );
          },
          onError: (Object e, StackTrace st) {
            reject(
              Rejection(
                actual: ['a future that completed as an error'],
                which: [
                  ...prefixFirst('threw ', postfixLast(' at:', literal(e))),
                  ...indent(LineSplitter.split(st.toString())),
                ],
              ),
            );
          },
        ),
      );
    });
  }

  /// Expects that the `Future` completes as an error.
  ///
  /// Fails if the future completes to a value.
  ///
  /// Pass [errorCondition] to check expectations on the error thrown by the
  /// future.
  ///
  /// The returned future will complete when the subject future has completed,
  /// and [errorCondition] has optionally been checked.
  Future<void> throws<E extends Object>([AsyncCondition<E>? errorCondition]) {
    return context.nestAsync<E>(
      () => ['completes to an error${E == Object ? '' : ' of type $E'}'],
      (actual) async {
        try {
          return Extracted.rejection(
            actual: prefixFirst('completed to ', literal(await actual)),
            which: ['did not throw'],
          );
        } on E catch (e) {
          return Extracted.value(e);
        } catch (e, st) {
          return Extracted.rejection(
            actual: prefixFirst('completed to error ', literal(e)),
            which: [
              'threw an exception that is not a $E at:',
              ...indent(LineSplitter.split(st.toString())),
            ],
          );
        }
      },
      errorCondition,
    );
  }
}

/// Expectations on a [StreamQueue].
///
/// Streams should be wrapped in user test code so that any reuse of the same
/// Stream, and the full stream lifecycle, is explicit.
extension StreamChecks<T> on Subject<StreamQueue<T>> {
  /// Calls [Context.expectAsync] and wraps [predicate] with a transaction.
  ///
  /// The transaction is committed if the check passes, or rejected if it fails.
  Future<void> _expectAsync(
    Iterable<String> Function() clause,
    FutureOr<Rejection?> Function(StreamQueue<T>) predicate,
  ) => context.expectAsync(clause, (actual) async {
    final transaction = actual.startTransaction();
    final copy = transaction.newQueue();
    final result = await predicate(copy);
    if (result == null) {
      transaction.commit(copy);
    } else {
      transaction.reject();
    }
    return result;
  });

  /// Expect that the `Stream` emits a value without first emitting an error.
  ///
  /// Fails if the stream emits an error instead of a value, or closes without
  /// emitting a value.
  ///
  /// If an error is emitted the queue will be left in its original state, the
  /// error will not be consumed.
  /// If an event is emitted, it will be consumed from the queue.
  ///
  /// Pass [emittedCondition] to check expectations on the value emitted by the
  /// stream.
  ///
  /// The returned future will complete when the stream has emitted, errored, or
  /// ended, and the [emittedCondition] has optionally been checked.
  Future<void> emits([AsyncCondition<T>? emittedCondition]) {
    return context.nestAsync<T>(() => ['emits a value'], (actual) async {
      if (!await actual.hasNext) {
        return Extracted.rejection(
          actual: ['a stream'],
          which: ['closed without emitting enough values'],
        );
      }
      try {
        await actual.peek;
        return Extracted.value(await actual.next);
      } catch (e, st) {
        return Extracted.rejection(
          actual: prefixFirst('a stream with error ', literal(e)),
          which: [
            'emitted an error instead of a value at:',
            ...indent(LineSplitter.split(st.toString())),
          ],
        );
      }
    }, emittedCondition);
  }

  /// Expects that the stream emits an error of type [E].
  ///
  /// Fails if the stream emits any value.
  /// Fails if the stream emits an error with an incorrect type.
  /// Fails if the stream closes without emitting an error.
  ///
  /// If an event is emitted the queue will be left in its original state, the
  /// event will not be consumed.
  /// If an error is emitted, it will be consumed from the queue.
  ///
  /// Pass [errorCondition] to check expectations on the error emitted by the
  /// stream.
  ///
  /// The returned future will complete when the stream has emitted, errored, or
  /// ended, and the [errorCondition] has optionally been checked.
  Future<void> emitsError<E extends Object>([
    AsyncCondition<E>? errorCondition,
  ]) {
    return context.nestAsync<E>(
      () => ['emits an error${E == Object ? '' : ' of type $E'}'],
      (actual) async {
        if (!await actual.hasNext) {
          return Extracted.rejection(
            actual: ['a stream'],
            which: ['closed without emitting an expected error'],
          );
        }
        try {
          final value = await actual.peek;
          return Extracted.rejection(
            actual: prefixFirst('a stream emitting value ', literal(value)),
            which: ['closed without emitting an error'],
          );
        } on E catch (e) {
          await actual.next.then<void>((_) {}, onError: (_) {});
          return Extracted.value(e);
        } catch (e, st) {
          return Extracted.rejection(
            actual: prefixFirst('a stream with error ', literal(e)),
            which: [
              'emitted an error which is not $E at:',
              ...indent(LineSplitter.split(st.toString())),
            ],
          );
        }
      },
      errorCondition,
    );
  }

  /// Expects that the `Stream` emits any number of events before emitting an
  /// event that satisfies [condition].
  ///
  /// Returns a `Future` that completes after the stream has emitted an event
  /// that satisfies [condition].
  ///
  /// Fails if the stream emits an error or closes before emitting a matching
  /// event.
  ///
  /// If this expectation fails, the source queue will be left in its original
  /// state.
  /// If this expectation succeeds, consumes the matching event and all prior
  /// events.
  Future<void> emitsThrough(AsyncCondition<T> condition) async {
    await _expectAsync(
      () => [
        'emits any values then emits a value that:',
        ...describe(condition),
      ],
      (actual) async {
        var count = 0;
        while (await actual.hasNext) {
          if (softCheck(await actual.next, condition) == null) {
            return null;
          }
          count++;
        }
        return Rejection(
          actual: ['a stream'],
          which: ['ended after emitting $count elements with none matching'],
        );
      },
    );
  }

  /// Expects that the stream satisfies each condition in [conditions] serially.
  ///
  /// Waits for each condition to be satisfied or rejected before checking the
  /// next. Subsequent conditions will not see any events consumed by earlier
  /// conditions.
  ///
  /// ```dart
  /// await check(someStream).withQueue.inOrder([
  ///   (s) => s.emits((e) => e.equals(0)),
  ///   (s) => s.emits((e) => e.equals(1)),
  /// ]);
  /// ```
  ///
  /// If this expectation fails, the source queue will be left in its original
  /// state.
  /// If this expectation succeeds, consumes as many events from the source
  /// stream as are consumed by all the conditions.
  Future<void> inOrder(
    Iterable<AsyncCondition<StreamQueue<T>>> conditions,
  ) async {
    conditions = conditions.toList();
    final descriptions = <String>[];
    await _expectAsync(
      () => descriptions.isEmpty
          ? ['satisfies ${conditions.length} conditions in order']
          : descriptions,
      (actual) async {
        var satisfiedCount = 0;
        for (var condition in conditions) {
          descriptions.addAll(await describeAsync(condition));
          final failure = await softCheckAsync(actual, condition);
          if (failure != null) {
            final which = failure.rejection.which;
            return Rejection(
              actual: ['a stream'],
              which: [
                if (satisfiedCount > 0)
                  'satisfied $satisfiedCount conditions then',
                'failed to satisfy the condition at index $satisfiedCount',
                if (failure.detail.depth > 0) ...[
                  'because it:',
                  ...indent(
                    failure.detail.actual.skip(1),
                    failure.detail.depth - 1,
                  ),
                  ...indent(
                    prefixFirst('Actual: ', failure.rejection.actual),
                    failure.detail.depth,
                  ),
                  if (which != null)
                    ...indent(
                      prefixFirst('Which: ', which),
                      failure.detail.depth,
                    ),
                ] else ...[
                  if (which != null) ...prefixFirst('because it ', which),
                ],
              ],
            );
          }
          satisfiedCount++;
        }
        return null;
      },
    );
  }

  /// Expects that the stream statisfies at least one condition from
  /// [conditions].
  ///
  /// If this expectation fails, the source queue will be left in its original
  /// state.
  /// If this expectation succeeds, consumes the same events from the source
  /// queue as the satisfied condition. If multiple conditions are satisfied,
  /// chooses the condition which consumed the most events.
  Future<void> anyOf(Iterable<AsyncCondition<StreamQueue<T>>> conditions) {
    conditions = conditions.toList();
    if (conditions.isEmpty) {
      throw ArgumentError('conditions may not be empty');
    }
    final descriptions = <Iterable<String>>[];
    return context.expectAsync(
      () => descriptions.isEmpty
          ? ['satisfies any of ${conditions.length} conditions']
          : [
              'satisfies one of:',
              for (var i = 0; i < descriptions.length; i++) ...[
                ...descriptions[i],
                if (i < descriptions.length - 1) 'or,',
              ],
            ],
      (actual) async {
        final transaction = actual.startTransaction();
        StreamQueue<T>? longestAccepted;
        final descriptionFuture = Future.wait(conditions.map(describeAsync));
        final failures = await Future.wait(
          conditions.map((condition) async {
            final copy = transaction.newQueue();
            final failure = await softCheckAsync(copy, condition);
            if (failure == null &&
                (longestAccepted == null ||
                    copy.eventsDispatched >
                        longestAccepted!.eventsDispatched)) {
              longestAccepted = copy;
            }
            return failure;
          }),
        );
        descriptions.addAll(await descriptionFuture);
        if (longestAccepted != null) {
          transaction.commit(longestAccepted!);
          return null;
        }
        transaction.reject();
        Iterable<String> failureDetails(int index, CheckFailure? failure) {
          final actual = failure!.rejection.actual;
          final which = failure.rejection.which;
          final detail = failure.detail;
          final failed = 'failed the condition at index $index';
          if (detail.depth > 0) {
            return [
              '$failed because it:',
              ...indent(detail.actual.skip(1), detail.depth - 1),
              ...indent(prefixFirst('Actual: ', actual), detail.depth),
              if (which != null)
                ...indent(prefixFirst('Which: ', which), detail.depth),
            ];
          } else {
            return [
              if (which == null)
                failed
              else ...[
                '$failed because it:',
                ...indent(which),
              ],
            ];
          }
        }

        return Rejection(
          actual: ['a stream'],
          which: [
            'failed to satisfy any condition',
            for (var i = 0; i < failures.length; i++)
              ...failureDetails(i, failures[i]),
          ],
        );
      },
    );
  }

  /// Expects that the stream closes without emitting any event that satisfies
  /// [condition].
  ///
  /// Returns a `Future` that completes after the stream has closed.
  ///
  /// Fails if the stream emits any even that satisfies [condition].
  ///
  /// If this expectation fails, the source queue will be left in its original
  /// state.
  /// If this expectation succeeds, consumes all the events that did not satisfy
  /// [condition] until the end of the stream.
  Future<void> neverEmits(AsyncCondition<T> condition) async {
    await _expectAsync(
      () => ['never emits a value that:', ...describe(condition)],
      (actual) async {
        var count = 0;
        await for (var emitted in actual.rest) {
          if (softCheck(emitted, condition) == null) {
            return Rejection(
              actual: ['a stream'],
              which: [
                ...prefixFirst('emitted ', literal(emitted)),
                if (count > 0) 'following $count other items',
              ],
            );
          }
          count++;
        }
        return null;
      },
    );
  }

  /// Optionally consumes an event that matches [condition] from the stream.
  ///
  /// This expectation never fails.
  ///
  /// If a non-matching event is emitted, no events are consumed.
  /// If a matching event is emitted, that event is consumed.
  Future<void> mayEmit(AsyncCondition<T> condition) {
    return context.expectAsync(
      () {
        try {
          return ['may emit a value that:', ...describe(condition)];
        } on AsyncConditionDisallowed {
          return ['may emit a value satisfying an asynchronous condition'];
        }
      },
      (actual) async {
        if (!await actual.hasNext) return null;
        try {
          final value = await actual.peek;
          if (await softCheckAsync(value, condition) == null) {
            await actual.next;
          }
        } catch (_) {
          // Ignore an emitted error - it does not match he event.
        }
        return null;
      },
    );
  }

  /// Optionally consumes events that match [condition] from the stream.
  ///
  /// This expectation never fails.
  ///
  /// Consumes matching events until one of the following happens:
  /// - A non-matching event is emitted.
  /// - An error is emitted.
  /// - The stream closes.
  Future<void> mayEmitMultiple(AsyncCondition<T> condition) {
    return context.expectAsync(
      () {
        try {
          return ['may emit a value that:', ...describe(condition)];
        } on AsyncConditionDisallowed {
          return ['may emit a value satisfying an asynchronous condition'];
        }
      },
      (actual) async {
        while (await actual.hasNext) {
          try {
            final value = await actual.peek;
            if (await softCheckAsync(value, condition) == null) {
              await actual.next;
            } else {
              return null;
            }
          } catch (_) {
            return null;
          }
        }
        return null;
      },
    );
  }

  /// Expects that the stream closes without emitting any events or errors.
  ///
  /// If this expectation fails, the source queue will be left in its original
  /// state, the event or error that caused it to fail will not be consumed.
  Future<void> isDone() async {
    await _expectAsync(() => ['is done'], (actual) async {
      if (!await actual.hasNext) return null;
      try {
        return Rejection(
          actual: ['a stream'],
          which: prefixFirst(
            'emitted an unexpected value: ',
            literal(await actual.next),
          ),
        );
      } catch (e, st) {
        return Rejection(
          actual: ['a stream'],
          which: [
            ...prefixFirst(
              'emitted an unexpected error: ',
              postfixLast(' at:', literal(e)),
            ),
            ...indent(LineSplitter.split(st.toString())),
          ],
        );
      }
    });
  }
}

extension WithQueueExtension<T> on Subject<Stream<T>> {
  /// Wrap the stream in a [StreamQueue] to allow using checks from
  /// [StreamChecks].
  ///
  /// Stream expectations operate on a queue, instead of directly on the stream,
  /// so that they can support conditional expectations and check multiple
  /// possibilities from the same point in the stream.
  Subject<StreamQueue<T>> get withQueue => context.nest(
    () => [],
    (actual) => Extracted.value(StreamQueue(actual)),
    atSameLevel: true,
  );
}