Fix race conditions in test_driver for tool tests (#88322)

packages/flutter_tools/test/integration.shard/debugger_stepping_test.dart

@@ -4,8 +4,6 @@
 
 // @dart = 2.8
 
-import 'dart:io';
-
 import 'package:file/file.dart';
 
 import '../src/common.dart';
@@ -32,8 +30,7 @@ void main() {
 
     await _flutter.run(withDebugger: true, startPaused: true);
     await _flutter.addBreakpoint(_project.breakpointUri, _project.breakpointLine);
-    await _flutter.resume();
-    await _flutter.waitForPause(); // Now we should be on the breakpoint.
+    await _flutter.resume(waitForNextPause: true); // Now we should be on the breakpoint.
 
     expect((await _flutter.getSourceLocation()).line, equals(_project.breakpointLine));
 
@@ -53,5 +50,5 @@ void main() {
     }
 
     await _flutter.stop();
-  }, skip: Platform.isWindows); // Skipping for https://github.com/flutter/flutter/issues/87481
+  });
 }

packages/flutter_tools/test/integration.shard/test_driver.dart

@@ -235,8 +235,10 @@ abstract class FlutterTestDriver {
   /// an hour before setting the breakpoint. Would the code still eventually hit
   /// the breakpoint and stop?
   Future<void> breakAt(Uri uri, int line) async {
+    final String isolateId = await _getFlutterIsolateId();
+    final Future<Event> event = subscribeToPauseEvent(isolateId);
     await addBreakpoint(uri, line);
-    await waitForPause();
+    await waitForPauseEvent(isolateId, event);
   }
 
   Future<void> addBreakpoint(Uri uri, int line) async {
@@ -248,44 +250,64 @@ abstract class FlutterTestDriver {
     );
   }
 
-  // This method isn't racy. If the isolate is already paused,
-  // it will immediately return.
-  Future<Isolate> waitForPause() => waitForDebugEvent('Pause');
-  Future<Isolate> waitForResume() => waitForDebugEvent('Resume');
+  Future<Event> subscribeToPauseEvent(String isolateId) => subscribeToDebugEvent('Pause', isolateId);
+  Future<Event> subscribeToResumeEvent(String isolateId) => subscribeToDebugEvent('Resume', isolateId);
+
+  Future<Isolate> waitForPauseEvent(String isolateId, Future<Event> event) =>
+      waitForDebugEvent('Pause', isolateId, event);
+  Future<Isolate> waitForResumeEvent(String isolateId, Future<Event> event) =>
+      waitForDebugEvent('Resume', isolateId, event);
+
+  Future<Isolate> waitForPause() async => subscribeAndWaitForDebugEvent('Pause', await _getFlutterIsolateId());
+  Future<Isolate> waitForResume() async => subscribeAndWaitForDebugEvent('Resume', await _getFlutterIsolateId());
 
-  Future<Isolate> waitForDebugEvent(String kind) async {
-    return _timeoutWithMessages<Isolate>(
-      () async {
-        final String flutterIsolate = await _getFlutterIsolateId();
-        final Completer<Event> pauseEvent = Completer<Event>();
 
-        // Start listening for events containing 'kind'.
-        final StreamSubscription<Event> pauseSubscription = _vmService.onDebugEvent
+  Future<Isolate> subscribeAndWaitForDebugEvent(String kind, String isolateId) {
+    final Future<Event> event = subscribeToDebugEvent(kind, isolateId);
+    return waitForDebugEvent(kind, isolateId, event);
+  }
+
+  /// Subscribes to debug events containing [kind].
+  ///
+  /// Returns a future that completes when the [kind] event is received.
+  ///
+  /// Note that this method should be called before the command that triggers
+  /// the event to subscribe to the event in time, for example:
+  ///
+  /// ```
+  ///  var event = subscribeToDebugEvent('Pause', id); // Subscribe to 'pause' events.
+  ///  ...                                             // Code that pauses the app.
+  ///  await waitForDebugEvent('Pause', id, event);    // Isolate is paused now.
+  /// ```
+  Future<Event> subscribeToDebugEvent(String kind, String isolateId) {
+    _debugPrint('Start listening for $kind events');
+
+    return _vmService.onDebugEvent
       .where((Event event) {
-            return event.isolate.id == flutterIsolate
+        return event.isolate.id == isolateId
             && event.kind.startsWith(kind);
-          })
-          .listen((Event event) {
-            if (!pauseEvent.isCompleted) {
-              pauseEvent.complete(event);
+      }).first;
   }
-          });
 
-        // But also check if the isolate was already at the stae we need (only after we've
-        // set up the subscription) to avoid races. If it was paused, we don't need to wait
-        // for the event.
-        final Isolate isolate = await _vmService.getIsolate(flutterIsolate);
+  /// Wait for the [event] if needed.
+  ///
+  /// Return immediately if the isolate is already in the desired state.
+  Future<Isolate> waitForDebugEvent(String kind, String isolateId, Future<Event> event) {
+    return _timeoutWithMessages<Isolate>(
+      () async {
+        // But also check if the isolate was already at the state we need (only after we've
+        // set up the subscription) to avoid races. If it already in the desired state, we
+        // don't need to wait for the event.
+        final Isolate isolate = await _vmService.getIsolate(isolateId);
         if (isolate.pauseEvent.kind.startsWith(kind)) {
           _debugPrint('Isolate was already at "$kind" (${isolate.pauseEvent.kind}).');
+          event.ignore();
         } else {
           _debugPrint('Waiting for "$kind" event to arrive...');
-          await pauseEvent.future;
+          await event;
         }
 
-        // Cancel the subscription on either of the above.
-        await pauseSubscription.cancel();
-
-        return getFlutterIsolate();
+        return _vmService.getIsolate(isolateId);
       },
       task: 'Waiting for isolate to $kind',
     );
@@ -311,12 +333,17 @@ abstract class FlutterTestDriver {
 
   Future<Isolate> _resume(String step, bool waitForNextPause) async {
     assert(waitForNextPause != null);
+    final String isolateId = await _getFlutterIsolateId();
+
+    final Future<Event> resume = subscribeToResumeEvent(isolateId);
+    final Future<Event> pause = subscribeToPauseEvent(isolateId);
+
     await _timeoutWithMessages<dynamic>(
-      () async => _vmService.resume(await _getFlutterIsolateId(), step: step),
+      () async => _vmService.resume(isolateId, step: step),
       task: 'Resuming isolate (step=$step)',
     );
-    await waitForResume();
-    return waitForNextPause ? waitForPause() : null;
+    await waitForResumeEvent(isolateId, resume);
+    return waitForNextPause? waitForPauseEvent(isolateId, pause): null;
   }
 
   Future<ObjRef> evaluateInFrame(String expression) async {

packages/flutter_tools/test/web.shard/debugger_stepping_web_test.dart

@@ -29,9 +29,7 @@ void main() {
       withDebugger: true, startPaused: true, chrome: true,
       additionalCommandArgs: <String>['--verbose', '--web-renderer=html']);
     await flutter.addBreakpoint(_project.breakpointUri, _project.breakpointLine);
-    await flutter.resume();
-    await flutter.waitForPause(); // Now we should be on the breakpoint.
-
+    await flutter.resume(waitForNextPause: true); // Now we should be on the breakpoint.
     expect((await flutter.getSourceLocation()).line, equals(_project.breakpointLine));
 
     // Issue 5 steps, ensuring that we end up on the annotated lines each time.

packages/flutter_tools/test/web.shard/expression_evaluation_web_test.dart

@@ -127,8 +127,7 @@ void main() {
         project.breakpointAppUri,
         project.breakpointLine,
       );
-      await flutter.resume();
-      return flutter.waitForPause();
+      return flutter.resume(waitForNextPause: true);
     }
 
     Future<void> startPaused({bool expressionEvaluation}) {

packages/flutter_tools/test/web.shard/vm_service_web_test.dart

@@ -62,7 +62,7 @@ void main() {
         validateFlutterVersion(client1),
         validateFlutterVersion(client2)]
       );
-    }, skip: true); // DDS failure: https://github.com/dart-lang/sdk/issues/45569
+    }, skip: true); // DDS failure: https://github.com/dart-lang/sdk/issues/46696
   });
 
   group('Clients of flutter run on web with DDS disabled', () {