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 'package:meta/meta.dart' show awaitNotRequired;

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.
  ///
  /// {@example /example/async/future/completes.dart}
  @awaitNotRequired
  Future<Subject<T>> completes([Condition<T>? completionCondition]) {
    return context.nestAsync<T>(
      () => ['completes to a value'],
      addPredicate: (predicateNoun) => 'completes to $predicateNoun',
      (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 [Condition.softCheck] or [Condition.softCheckSync]
  /// since there is no concrete end point where this condition has definitely
  /// succeeded.
  ///
  /// {@example /example/async/future/does_not_complete.dart}
  void get 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.
  ///
  /// {@example /example/async/future/throws.dart}
  @awaitNotRequired
  Future<Subject<E>> throws<E extends Object>([Condition<E>? errorCondition]) {
    return context.nestAsync<E>(
      () => ['completes to an error${E == Object ? '' : ' of type $E'}'],
      addPredicate: (predicateNoun) => 'throws $predicateNoun',
      (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.
  @awaitNotRequired
  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.
  ///
  /// {@example /example/async/stream_queue/emits.dart}
  @awaitNotRequired
  Future<Subject<T>> emits([Condition<T>? emittedCondition]) {
    return context.nestAsync<T>(
      () => ['emits a value'],
      addPredicate: (predicateNoun) => 'emits $predicateNoun',
      (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.
  ///
  /// {@example /example/async/stream_queue/emits_error.dart}
  @awaitNotRequired
  Future<Subject<E>> emitsError<E extends Object>([
    Condition<E>? errorCondition,
  ]) {
    return context.nestAsync<E>(
      () => ['emits an error${E == Object ? '' : ' of type $E'}'],
      addPredicate: (predicateNoun) => 'emits error $predicateNoun',
      (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.
  ///
  /// {@example /example/async/stream_queue/emits_through.dart}
  @awaitNotRequired
  Future<void> emitsThrough(Condition<T> condition) async {
    await _expectAsync(
      () {
        final description = condition.describe();
        if (description is Iterable<String>) {
          return ['emits any values then emits a value that:', ...description];
        }
        return [
          'emits any values then emits a value satisfying an asynchronous condition',
        ];
      },
      (actual) async {
        var count = 0;
        while (await actual.hasNext) {
          if (await condition.softCheck(await actual.next) == 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.
  ///
  /// 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.
  ///
  /// {@example /example/async/stream_queue/in_order.dart}
  @awaitNotRequired
  Future<void> inOrder(Iterable<Condition<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 condition.describe());
          final failure = await condition.softCheck(actual);
          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.collapsedDetails case (
                  final expected,
                  final actual,
                )) ...[
                  'because it:',
                  '  Expected: $expected',
                  '  Actual: $actual',
                  if (which != null) ...indent(prefixFirst('Which: ', which)),
                ] else 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.
  ///
  /// {@example /example/async/stream_queue/any_of.dart}
  @awaitNotRequired
  Future<void> anyOf(Iterable<Condition<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 = conditions
            .map((c) async => await c.describe())
            .wait;
        // soft checks are unexpected to throw, if they do the unhandled async
        // error should surface in the test runner.
        final failures = await conditions.map((condition) async {
          final copy = transaction.newQueue();
          final failure = await condition.softCheck(copy);
          if (failure == null &&
              (longestAccepted == null ||
                  copy.eventsDispatched > longestAccepted!.eventsDispatched)) {
            longestAccepted = copy;
          }
          return failure;
        }).wait;

        descriptions.addAll(await descriptionFuture);
        if (longestAccepted case final accepted?) {
          transaction.commit(accepted);
          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.collapsedDetails case (final expected, final actual)) {
            return [
              '$failed because it:',
              '  Expected: $expected',
              '  Actual: $actual',
              if (which != null) ...indent(prefixFirst('Which: ', which)),
            ];
          } else 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.
  ///
  /// {@example /example/async/stream_queue/never_emits.dart}
  @awaitNotRequired
  Future<void> neverEmits(Condition<T> condition) async {
    await _expectAsync(
      () {
        final description = condition.describe();
        if (description is Iterable<String>) {
          return ['never emits a value that:', ...description];
        }
        return ['never emits a value satisfying an asynchronous condition'];
      },
      (actual) async {
        var count = 0;
        await for (var emitted in actual.rest) {
          if (await condition.softCheck(emitted) == 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.
  ///
  /// {@example /example/async/stream_queue/may_emit.dart}
  @awaitNotRequired
  Future<void> mayEmit(Condition<T> condition) {
    return context.expectAsync(
      () {
        final description = condition.describe();
        if (description is Iterable<String>) {
          return ['may emit a value that:', ...description];
        }
        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 condition.softCheck(value) == 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.
  ///
  /// {@example /example/async/stream_queue/may_emit_multiple.dart}
  @awaitNotRequired
  Future<void> mayEmitMultiple(Condition<T> condition) {
    return context.expectAsync(
      () {
        final description = condition.describe();
        if (description is Iterable<String>) {
          return ['may emit a value that:', ...description];
        }
        return ['may emit a value satisfying an asynchronous condition'];
      },
      (actual) async {
        while (await actual.hasNext) {
          try {
            final value = await actual.peek;
            if (await condition.softCheck(value) == 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.
  ///
  /// {@example /example/async/stream_queue/is_done.dart}
  @awaitNotRequired
  Future<void> get 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.
  ///
  /// {@example /example/async/stream/with_queue.dart}
  Subject<StreamQueue<T>> get withQueue => context.nest(
    () => [],
    (actual) => Extracted.value(StreamQueue(actual)),
    atSameLevel: true,
  );
}

extension FutureSubjectExtension<T> on Future<Subject<T>> {
  /// Applies the expectations invoked in [condition] to this subject.
  ///
  /// Use this method when it would otherwise not be possible to check multiple
  /// expectations for this subject due to cascade notation already being used
  /// in a way that would conflict.
  ///
  /// ```dart
  /// await check(someFuture).completes().which(.it()
  ///   ..isLessThan(10)
  ///   ..isGreaterThan(0));
  /// ```
  ///
  /// {@example /example/async/future_subject/which.dart}
  @awaitNotRequired
  Future<void> which(Condition<T> condition) async {
    await condition.apply(await this);
  }
}