flutter/packages/flutter_tools/test/integration.shard/debug_adapter/test_client.dart

// Copyright 2014 The Flutter Authors. 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:dds/src/dap/logging.dart';
import 'package:dds/src/dap/protocol_generated.dart';
import 'package:dds/src/dap/protocol_stream.dart';
import 'package:flutter_tools/src/debug_adapters/flutter_adapter_args.dart';

import 'test_server.dart';

/// A helper class to simplify acting as a client for interacting with the
/// [DapTestServer] in tests.
///
/// Methods on this class should map directly to protocol methods. Additional
/// helpers are available in [DapTestClientExtension].
class DapTestClient {
  DapTestClient._(
    this._channel,
    this._logger, {
    this.captureVmServiceTraffic = false,
  }) {
    // Set up a future that will complete when the 'dart.debuggerUris' event is
    // emitted by the debug adapter so tests have easy access to it.
    vmServiceUri = event('dart.debuggerUris').then<Uri?>((Event event) {
      final Map<String, Object?> body = event.body! as Map<String, Object?>;
      return Uri.parse(body['vmServiceUri']! as String);
    }).catchError((Object? e) => null);

    _subscription = _channel.listen(
      _handleMessage,
      onDone: () {
        if (_pendingRequests.isNotEmpty) {
          _logger?.call(
              'Application terminated without a response to ${_pendingRequests.length} requests');
        }
        _pendingRequests.forEach((int id, _OutgoingRequest request) => request.completer.completeError(
            'Application terminated without a response to request $id (${request.name})'));
        _pendingRequests.clear();
      },
    );
  }

  final ByteStreamServerChannel _channel;
  late final StreamSubscription<String> _subscription;
  final Logger? _logger;
  final bool captureVmServiceTraffic;
  final Map<int, _OutgoingRequest> _pendingRequests = <int, _OutgoingRequest>{};
  final StreamController<Event> _eventController = StreamController<Event>.broadcast();
  int _seq = 1;
  late final Future<Uri?> vmServiceUri;

  /// Returns a stream of [OutputEventBody] events.
  Stream<OutputEventBody> get outputEvents => events('output')
      .map((Event e) => OutputEventBody.fromJson(e.body! as Map<String, Object?>));

  /// Returns a stream of [StoppedEventBody] events.
  Stream<StoppedEventBody> get stoppedEvents => events('stopped')
      .map((Event e) => StoppedEventBody.fromJson(e.body! as Map<String, Object?>));

  /// Returns a stream of the string output from [OutputEventBody] events.
  Stream<String> get output => outputEvents.map((OutputEventBody output) => output.output);

  /// Returns a stream of the string output from [OutputEventBody] events with the category 'stdout'.
  Stream<String> get stdoutOutput => outputEvents
      .where((OutputEventBody output) => output.category == 'stdout')
      .map((OutputEventBody output) => output.output);

  /// Sends a custom request to the server and waits for a response.
  Future<Response> custom(String name, [Object? args]) async {
    return sendRequest(args, overrideCommand: name);
  }

  /// Returns a Future that completes with the next [event] event.
  Future<Event> event(String event) => _eventController.stream.firstWhere(
      (Event e) => e.event == event,
      orElse: () => throw Exception('Did not receive $event event before stream closed'));

  /// Returns a stream for [event] events.
  Stream<Event> events(String event) {
    return _eventController.stream.where((Event e) => e.event == event);
  }

  /// Returns a stream of custom 'dart.serviceExtensionAdded' events.
  Stream<Map<String, Object?>> get serviceExtensionAddedEvents =>
      events('dart.serviceExtensionAdded')
          .map((Event e) => e.body! as Map<String, Object?>);

  /// Returns a stream of custom 'flutter.serviceExtensionStateChanged' events.
  Stream<Map<String, Object?>> get serviceExtensionStateChangedEvents =>
      events('flutter.serviceExtensionStateChanged')
          .map((Event e) => e.body! as Map<String, Object?>);

  /// Returns a stream of 'dart.testNotification' custom events from the
  /// package:test JSON reporter.
  Stream<Map<String, Object?>> get testNotificationEvents =>
      events('dart.testNotification')
          .map((Event e) => e.body! as Map<String, Object?>);

  /// Sends a custom request to the debug adapter to trigger a Hot Reload.
  Future<Response> hotReload() {
    return custom('hotReload');
  }

  /// Sends a custom request to the debug adapter to trigger a Hot Restart.
  Future<Response> hotRestart() {
    return custom('hotRestart');
  }

  /// Send an initialize request to the server.
  ///
  /// This occurs before the request to start running/debugging a script and is
  /// used to exchange capabilities and send breakpoints and other settings.
  Future<Response> initialize({
    String exceptionPauseMode = 'None',
    bool? supportsRunInTerminalRequest,
  }) async {
    final List<ProtocolMessage> responses = await Future.wait(<Future<ProtocolMessage>>[
      event('initialized'),
      sendRequest(InitializeRequestArguments(
        adapterID: 'test',
        supportsRunInTerminalRequest: supportsRunInTerminalRequest,
      )),
      sendRequest(
        SetExceptionBreakpointsArguments(
          filters: <String>[exceptionPauseMode],
        ),
      ),
    ]);
    await sendRequest(ConfigurationDoneArguments());
    return responses[1] as Response; // Return the initialize response.
  }

  /// Send a launchRequest to the server, asking it to start a Flutter app.
  Future<Response> launch({
    String? program,
    List<String>? args,
    List<String>? toolArgs,
    String? cwd,
    bool? noDebug,
    List<String>? additionalProjectPaths,
    bool? debugSdkLibraries,
    bool? debugExternalPackageLibraries,
    bool? evaluateGettersInDebugViews,
    bool? evaluateToStringInDebugViews,
  }) {
    return sendRequest(
      FlutterLaunchRequestArguments(
        noDebug: noDebug,
        program: program,
        cwd: cwd,
        args: args,
        toolArgs: toolArgs,
        additionalProjectPaths: additionalProjectPaths,
        debugSdkLibraries: debugSdkLibraries,
        debugExternalPackageLibraries: debugExternalPackageLibraries,
        evaluateGettersInDebugViews: evaluateGettersInDebugViews,
        evaluateToStringInDebugViews: evaluateToStringInDebugViews,
        // When running out of process, VM Service traffic won't be available
        // to the client-side logger, so force logging on which sends VM Service
        // traffic in a custom event.
        sendLogsToClient: captureVmServiceTraffic,
      ),
      // We can't automatically pick the command when using a custom type
      // (FlutterLaunchRequestArguments).
      overrideCommand: 'launch',
    );
  }

  /// Send an attachRequest to the server, asking it to attach to an already-running Flutter app.
  Future<Response> attach({
    List<String>? toolArgs,
    String? vmServiceUri,
    String? cwd,
    List<String>? additionalProjectPaths,
    bool? debugSdkLibraries,
    bool? debugExternalPackageLibraries,
    bool? evaluateGettersInDebugViews,
    bool? evaluateToStringInDebugViews,
  }) {
    return sendRequest(
      FlutterAttachRequestArguments(
        cwd: cwd,
        toolArgs: toolArgs,
        vmServiceUri: vmServiceUri,
        additionalProjectPaths: additionalProjectPaths,
        debugSdkLibraries: debugSdkLibraries,
        debugExternalPackageLibraries: debugExternalPackageLibraries,
        evaluateGettersInDebugViews: evaluateGettersInDebugViews,
        evaluateToStringInDebugViews: evaluateToStringInDebugViews,
        // When running out of process, VM Service traffic won't be available
        // to the client-side logger, so force logging on which sends VM Service
        // traffic in a custom event.
        sendLogsToClient: captureVmServiceTraffic,
      ),
      // We can't automatically pick the command when using a custom type
      // (FlutterAttachRequestArguments).
      overrideCommand: 'attach',
    );
  }

  /// Sends an arbitrary request to the server.
  ///
  /// Returns a Future that completes when the server returns a corresponding
  /// response.
  Future<Response> sendRequest(Object? arguments,
      {bool allowFailure = false, String? overrideCommand}) {
    final String command = overrideCommand ?? commandTypes[arguments.runtimeType]!;
    final Request request =
        Request(seq: _seq++, command: command, arguments: arguments);
    final Completer<Response> completer = Completer<Response>();
    _pendingRequests[request.seq] =
        _OutgoingRequest(completer, command, allowFailure);
    _channel.sendRequest(request);
    return completer.future;
  }

  /// Returns a Future that completes with the next serviceExtensionAdded
  /// event for [extension].
  Future<Map<String, Object?>> serviceExtensionAdded(String extension) => serviceExtensionAddedEvents.firstWhere(
      (Map<String, Object?> body) => body['extensionRPC'] == extension,
      orElse: () => throw Exception('Did not receive $extension extension added event before stream closed'));

  /// Returns a Future that completes with the next serviceExtensionStateChanged
  /// event for [extension].
  Future<Map<String, Object?>> serviceExtensionStateChanged(String extension) => serviceExtensionStateChangedEvents.firstWhere(
      (Map<String, Object?> body) => body['extension'] == extension,
      orElse: () => throw Exception('Did not receive $extension extension state changed event before stream closed'));

  /// Initializes the debug adapter and launches [program]/[cwd] or calls the
  /// custom [launch] method.
  Future<void> start({
    String? program,
    String? cwd,
    String exceptionPauseMode = 'None',
    Future<Object?> Function()? launch,
  }) {
    return Future.wait(<Future<Object?>>[
      initialize(exceptionPauseMode: exceptionPauseMode),
      launch?.call() ?? this.launch(program: program, cwd: cwd),
    ], eagerError: true);
  }

  Future<void> stop() async {
    _channel.close();
    await _subscription.cancel();
  }

  Future<Response> terminate() => sendRequest(TerminateArguments());

  /// Handles an incoming message from the server, completing the relevant request
  /// of raising the appropriate event.
  Future<void> _handleMessage(Object? message) async {
    if (message is Response) {
      final _OutgoingRequest? pendingRequest = _pendingRequests.remove(message.requestSeq);
      if (pendingRequest == null) {
        return;
      }
      final Completer<Response> completer = pendingRequest.completer;
      if (message.success || pendingRequest.allowFailure) {
        completer.complete(message);
      } else {
        completer.completeError(message);
      }
    } else if (message is Event && !_eventController.isClosed) {
      _eventController.add(message);

      // When we see a terminated event, close the event stream so if any
      // tests are waiting on something that will never come, they fail at
      // a useful location.
      if (message.event == 'terminated') {
        unawaited(_eventController.close());
      }
    }
  }

  /// Creates a [DapTestClient] that connects the server listening on
  /// [host]:[port].
  static Future<DapTestClient> connect(
    DapTestServer server, {
    bool captureVmServiceTraffic = false,
    Logger? logger,
  }) async {
    final ByteStreamServerChannel channel = ByteStreamServerChannel(server.stream, server.sink, logger);
    return DapTestClient._(channel, logger,
        captureVmServiceTraffic: captureVmServiceTraffic);
  }
}

/// Useful events produced by the debug adapter during a debug session.
class TestEvents {
  TestEvents({
    required this.output,
    required this.testNotifications,
  });

  final List<OutputEventBody> output;
  final List<Map<String, Object?>> testNotifications;
}

class _OutgoingRequest {
  _OutgoingRequest(this.completer, this.name, this.allowFailure);

  final Completer<Response> completer;
  final String name;
  final bool allowFailure;
}

/// Additional helper method for tests to simplify interaction with [DapTestClient].
///
/// Unlike the methods on [DapTestClient] these methods might not map directly
/// onto protocol methods. They may call multiple protocol methods and/or
/// simplify assertion specific conditions/results.
extension DapTestClientExtension on DapTestClient {
  /// Collects all output events until the program terminates.
  ///
  /// These results include all events in the order they are received, including
  /// console, stdout and stderr.
  ///
  /// Only one of [start] or [launch] may be provided. Use [start] to customise
  /// the whole start of the session (including initialise) or [launch] to only
  /// customise the [launchRequest].
  Future<List<OutputEventBody>> collectAllOutput({
    String? program,
    String? cwd,
    Future<void> Function()? start,
    Future<Response> Function()? launch,
    bool skipInitialPubGetOutput = true
  }) async {
    assert(
      start == null || launch == null,
      'Only one of "start" or "launch" may be provided',
    );
    final Future<List<OutputEventBody>> outputEventsFuture = outputEvents.toList();

    // Don't await these, in case they don't complete (eg. an error prevents
    // the app from starting).
    if (start != null) {
      unawaited(start());
    } else {
      unawaited(this.start(program: program, cwd: cwd, launch: launch));
    }

    final List<OutputEventBody> output = await outputEventsFuture;

    // TODO(dantup): Integration tests currently trigger "flutter pub get" at
    //   the start due to some timestamp manipulation writing the pubspec.
    //   It may be possible to remove this if
    //   https://github.com/flutter/flutter/pull/91300 lands.
    return skipInitialPubGetOutput
        ? output.skipWhile((OutputEventBody output) => output.output.startsWith('Running "flutter pub get"')).toList()
        : output;
  }

  /// Collects all output and test events until the program terminates.
  ///
  /// These results include all events in the order they are received, including
  /// console, stdout, stderr and test notifications from the test JSON reporter.
  ///
  /// Only one of [start] or [launch] may be provided. Use [start] to customise
  /// the whole start of the session (including initialise) or [launch] to only
  /// customise the [launchRequest].
  Future<TestEvents> collectTestOutput({
    String? program,
    String? cwd,
    Future<Response> Function()? start,
    Future<Object?> Function()? launch,
  }) async {
    assert(
      start == null || launch == null,
      'Only one of "start" or "launch" may be provided',
    );

    final Future<List<OutputEventBody>> outputEventsFuture = outputEvents.toList();
    final Future<List<Map<String, Object?>>> testNotificationEventsFuture = testNotificationEvents.toList();

    if (start != null) {
      await start();
    } else {
      await this.start(program: program, cwd: cwd, launch: launch);
    }

    return TestEvents(
      output: await outputEventsFuture,
      testNotifications: await testNotificationEventsFuture,
    );
  }

  /// Sets a breakpoint at [line] in [file].
  Future<void> setBreakpoint(String filePath, int line) async {
    await sendRequest(
      SetBreakpointsArguments(
        source: Source(path: filePath),
        breakpoints: <SourceBreakpoint>[
          SourceBreakpoint(line: line),
        ],
      ),
    );
  }

  /// Sends a continue request for the given thread.
  ///
  /// Returns a Future that completes when the server returns a corresponding
  /// response.
  Future<Response> continue_(int threadId) =>
      sendRequest(ContinueArguments(threadId: threadId));

  /// Clears breakpoints in [file].
  Future<void> clearBreakpoints(String filePath) async {
    await sendRequest(
      SetBreakpointsArguments(
        source: Source(path: filePath),
        breakpoints: <SourceBreakpoint>[],
      ),
    );
  }

}