Run the request body tests against the browser client. (#711)

Author: brianquinlan

pkgs/http_client_conformance_tests/lib/http_client_conformance_tests.dart

@@ -5,12 +5,14 @@
 import 'package:http/http.dart';
 
 import 'src/redirect_tests.dart';
+import 'src/request_body_streamed_tests.dart';
 import 'src/request_body_tests.dart';
 import 'src/request_headers_tests.dart';
 import 'src/response_body_tests.dart';
 import 'src/response_headers_tests.dart';
 
 export 'src/redirect_tests.dart' show testRedirect;
+export 'src/request_body_streamed_tests.dart' show testRequestBodyStreamed;
 export 'src/request_body_tests.dart' show testRequestBody;
 export 'src/request_headers_tests.dart' show testRequestHeaders;
 export 'src/response_body_tests.dart' show testResponseBody;
@@ -32,7 +34,8 @@ void testAll(Client client,
     {bool canStreamRequestBody = true,
     bool canStreamResponseBody = true,
     bool redirectAlwaysAllowed = false}) {
-  testRequestBody(client, canStreamRequestBody: canStreamRequestBody);
+  testRequestBody(client);
+  testRequestBodyStreamed(client, canStreamRequestBody: canStreamRequestBody);
   testResponseBody(client, canStreamResponseBody: canStreamResponseBody);
   testRequestHeaders(client);
   testResponseHeaders(client);

pkgs/http_client_conformance_tests/lib/src/redirect_tests.dart

@@ -2,6 +2,7 @@
 // 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 'package:async/async.dart';
 import 'package:http/http.dart';
 import 'package:stream_channel/stream_channel.dart';
 import 'package:test/test.dart';
@@ -12,22 +13,24 @@ import 'package:test/test.dart';
 /// to limit redirects will be skipped.
 void testRedirect(Client client, {bool redirectAlwaysAllowed = false}) async {
   group('redirects', () {
-    late String host;
-    late StreamChannel<Object?> httpServerChannel;
+    late final String host;
+    late final StreamChannel<Object?> httpServerChannel;
+    late final StreamQueue<Object?> httpServerQueue;
 
-    setUp(() async {
+    setUpAll(() async {
       httpServerChannel = spawnHybridUri('../lib/src/redirect_server.dart');
-      host = 'localhost:${await httpServerChannel.stream.first as int}';
+      httpServerQueue = StreamQueue(httpServerChannel.stream);
+      host = 'localhost:${await httpServerQueue.next}';
     });
-    tearDown(() => httpServerChannel.sink.add(null));
+    tearDownAll(() => httpServerChannel.sink.add(null));
 
     test('disallow redirect', () async {
       final request = Request('GET', Uri.http(host, '/1'))
         ..followRedirects = false;
       final response = await client.send(request);
       expect(response.statusCode, 302);
       expect(response.isRedirect, true);
-    }, skip: redirectAlwaysAllowed ? 'redirects always allowed' : '');
+    }, skip: redirectAlwaysAllowed ? 'redirects always allowed' : false);
 
     test('allow redirect', () async {
       final request = Request('GET', Uri.http(host, '/1'))
@@ -56,7 +59,7 @@ void testRedirect(Client client, {bool redirectAlwaysAllowed = false}) async {
       final response = await client.send(request);
       expect(response.statusCode, 200);
       expect(response.isRedirect, false);
-    }, skip: redirectAlwaysAllowed ? 'redirects always allowed' : '');
+    }, skip: redirectAlwaysAllowed ? 'redirects always allowed' : false);
 
     test('too many redirects', () async {
       final request = Request('GET', Uri.http(host, '/6'))
@@ -66,7 +69,7 @@ void testRedirect(Client client, {bool redirectAlwaysAllowed = false}) async {
           client.send(request),
           throwsA(isA<ClientException>()
               .having((e) => e.message, 'message', 'Redirect limit exceeded')));
-    }, skip: redirectAlwaysAllowed ? 'redirects always allowed' : '');
+    }, skip: redirectAlwaysAllowed ? 'redirects always allowed' : false);
 
     test(
       'loop',

pkgs/http_client_conformance_tests/lib/src/request_body_server.dart

@@ -0,0 +1,47 @@
+// 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 'dart:io';
+
+import 'package:stream_channel/stream_channel.dart';
+
+/// Starts an HTTP server that captures the content type header and request
+/// body.
+///
+/// Channel protocol:
+///    On Startup:
+///     - send port
+///    On Request Received:
+///     - send "Content-Type" header
+///     - send request body
+///    When Receive Anything:
+///     - exit
+void hybridMain(StreamChannel<Object?> channel) async {
+  late HttpServer server;
+
+  server = (await HttpServer.bind('localhost', 0))
+    ..listen((request) async {
+      request.response.headers.set('Access-Control-Allow-Origin', '*');
+      if (request.method == 'OPTIONS') {
+        // Handle a CORS preflight request:
+        // https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS#preflighted_requests
+        request.response.headers
+          ..set('Access-Control-Allow-Methods', 'POST, DELETE')
+          ..set('Access-Control-Allow-Headers', 'Content-Type');
+      } else {
+        channel.sink.add(request.headers[HttpHeaders.contentTypeHeader]);
+        final serverReceivedBody =
+            await const Utf8Decoder().bind(request).fold('', (p, e) => '$p$e');
+        channel.sink.add(serverReceivedBody);
+      }
+      unawaited(request.response.close());
+    });
+
+  channel.sink.add(server.port);
+  await channel
+      .stream.first; // Any writes indicates that the server should exit.
+  unawaited(server.close());
+}

pkgs/http_client_conformance_tests/lib/src/request_body_streamed_server.dart

@@ -0,0 +1,42 @@
+// 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 'dart:io';
+
+import 'package:stream_channel/stream_channel.dart';
+
+/// Starts an HTTP server that absorbes a request stream of integers and
+/// signals the client to quit after 1000 have been received.
+///
+/// Channel protocol:
+///    On Startup:
+///     - send port
+///    On Integer == 1000 received:
+///     - send 1000
+///    When Receive Anything:
+///     - exit
+void hybridMain(StreamChannel<Object?> channel) async {
+  late HttpServer server;
+
+  server = (await HttpServer.bind('localhost', 0))
+    ..listen((request) async {
+      request.response.headers.set('Access-Control-Allow-Origin', '*');
+      await const LineSplitter()
+          .bind(const Utf8Decoder().bind(request))
+          .forEach((s) {
+        final lastReceived = int.parse(s.trim());
+        if (lastReceived == 1000) {
+          channel.sink.add(lastReceived);
+        }
+      });
+      unawaited(request.response.close());
+    });
+
+  channel.sink.add(server.port);
+  await channel
+      .stream.first; // Any writes indicates that the server should exit.
+  unawaited(server.close());
+}

pkgs/http_client_conformance_tests/lib/src/request_body_streamed_tests.dart

@@ -0,0 +1,63 @@
+// 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:http/http.dart';
+import 'package:stream_channel/stream_channel.dart';
+import 'package:test/test.dart';
+
+/// Tests that the [Client] correctly implements streamed request body
+/// uploading.
+///
+/// If [canStreamRequestBody] is `false` then tests that assume that the
+/// [Client] supports sending HTTP requests with unbounded body sizes will be
+/// skipped.
+void testRequestBodyStreamed(Client client,
+    {bool canStreamRequestBody = true}) {
+  group('streamed requests', () {
+    late String host;
+    late StreamChannel<Object?> httpServerChannel;
+    late StreamQueue<Object?> httpServerQueue;
+
+    setUp(() async {
+      httpServerChannel =
+          spawnHybridUri('../lib/src/request_body_streamed_server.dart');
+      httpServerQueue = StreamQueue(httpServerChannel.stream);
+      host = 'localhost:${await httpServerQueue.next}';
+    });
+    tearDown(() => httpServerChannel.sink.add(null));
+
+    test('client.send() with StreamedRequest', () async {
+      // The client continuously streams data to the server until
+      // instructed to stop (by setting `clientWriting` to `false`).
+      // The server sets `serverWriting` to `false` after it has
+      // already received some data.
+      //
+      // This ensures that the client supports streamed data sends.
+      var lastReceived = 0;
+
+      Stream<String> count() async* {
+        var i = 0;
+        unawaited(
+            httpServerQueue.next.then((value) => lastReceived = value as int));
+        do {
+          yield '${i++}\n';
+          // Let the event loop run.
+          await Future<void>.delayed(const Duration());
+        } while (lastReceived < 1000);
+      }
+
+      final request = StreamedRequest('POST', Uri.http(host, ''));
+      const Utf8Encoder()
+          .bind(count())
+          .listen(request.sink.add, onDone: request.sink.close);
+      await client.send(request);
+
+      expect(lastReceived, greaterThanOrEqualTo(1000));
+    });
+  }, skip: canStreamRequestBody ? false : 'does not stream request bodies');
+}