Add h2 stream integration tests (#502)

Author: fabianfett

Package.swift

@@ -23,7 +23,7 @@ let package = Package(
     dependencies: [
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.34.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.14.1"),
-        .package(url: "https://github.com/apple/swift-nio-http2.git", from: "1.18.2"),
+        .package(url: "https://github.com/apple/swift-nio-http2.git", from: "1.19.0"),
         .package(url: "https://github.com/apple/swift-nio-extras.git", from: "1.10.0"),
         .package(url: "https://github.com/apple/swift-nio-transport-services.git", from: "1.11.0"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.4.0"),

Tests/AsyncHTTPClientTests/HTTP2ClientTests+XCTest.swift

@@ -26,6 +26,8 @@ extension HTTP2ClientTests {
     static var allTests: [(String, (HTTP2ClientTests) -> () throws -> Void)] {
         return [
             ("testSimpleGet", testSimpleGet),
+            ("testStreamRequestBodyWithoutKnowledgeAboutLength", testStreamRequestBodyWithoutKnowledgeAboutLength),
+            ("testStreamRequestBodyWithFalseKnowledgeAboutLength", testStreamRequestBodyWithFalseKnowledgeAboutLength),
             ("testConcurrentRequests", testConcurrentRequests),
             ("testConcurrentRequestsFromDifferentThreads", testConcurrentRequestsFromDifferentThreads),
             ("testConcurrentRequestsWorkWithRequiredEventLoop", testConcurrentRequestsWorkWithRequiredEventLoop),

Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift

@@ -62,6 +62,38 @@ class HTTP2ClientTests: XCTestCase {
         XCTAssertEqual(response?.version, .http2)
     }
 
+    func testStreamRequestBodyWithoutKnowledgeAboutLength() {
+        let bin = HTTPBin(.http2(compress: false)) { _ in HTTPEchoHandler() }
+        defer { XCTAssertNoThrow(try bin.shutdown()) }
+        let client = self.makeDefaultHTTPClient()
+        defer { XCTAssertNoThrow(try client.syncShutdown()) }
+        var response: HTTPClient.Response?
+        let body = HTTPClient.Body.stream(length: nil) { writer in
+            writer.write(.byteBuffer(ByteBuffer(integer: UInt64(0)))).flatMap {
+                writer.write(.byteBuffer(ByteBuffer(integer: UInt64(0))))
+            }
+        }
+        XCTAssertNoThrow(response = try client.post(url: "https://localhost:\(bin.port)", body: body).wait())
+
+        XCTAssertEqual(.ok, response?.status)
+        XCTAssertEqual(response?.version, .http2)
+    }
+
+    func testStreamRequestBodyWithFalseKnowledgeAboutLength() {
+        let bin = HTTPBin(.http2(compress: false)) { _ in HTTPEchoHandler() }
+        defer { XCTAssertNoThrow(try bin.shutdown()) }
+        let client = self.makeDefaultHTTPClient()
+        defer { XCTAssertNoThrow(try client.syncShutdown()) }
+        let body = HTTPClient.Body.stream(length: 12) { writer in
+            writer.write(.byteBuffer(ByteBuffer(integer: UInt64(0)))).flatMap {
+                writer.write(.byteBuffer(ByteBuffer(integer: UInt64(0))))
+            }
+        }
+        XCTAssertThrowsError(try client.post(url: "https://localhost:\(bin.port)", body: body).wait()) {
+            XCTAssertEqual($0 as? HTTPClientError, .bodyLengthMismatch)
+        }
+    }
+
     func testConcurrentRequests() {
         let bin = HTTPBin(.http2(compress: false))
         defer { XCTAssertNoThrow(try bin.shutdown()) }

Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift

@@ -1218,13 +1218,14 @@ class HTTPEchoHandler: ChannelInboundHandler {
     func channelRead(context: ChannelHandlerContext, data: NIOAny) {
         let request = self.unwrapInboundIn(data)
         switch request {
-        case .head:
-            context.writeAndFlush(self.wrapOutboundOut(.head(.init(version: .http1_1, status: .ok))), promise: nil)
+        case .head(let requestHead):
+            context.writeAndFlush(self.wrapOutboundOut(.head(.init(version: .http1_1, status: .ok, headers: requestHead.headers))), promise: nil)
         case .body(let bytes):
             context.writeAndFlush(self.wrapOutboundOut(.body(.byteBuffer(bytes))), promise: nil)
         case .end:
-            context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
-            context.close(promise: nil)
+            context.writeAndFlush(self.wrapOutboundOut(.end(nil))).whenSuccess {
+                context.close(promise: nil)
+            }
         }
     }
 }