Adding tests

Author: fabianfett

Diff for: Sources/AsyncHTTPClient/RequestBag.swift

@@ -76,11 +76,6 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate>: HTTPScheduledReque
 //        }
     }
 
-    func willBeExecutedOnConnection(_: HTTPConnectionPool.Connection) {
-        // TODO: comment in once we switch to using the Request bag in AHC
-//        self.task.setConnection(connection)
-    }
-
     func requestWasQueued(_ scheduler: HTTPRequestScheduler) {
         self.stateLock.withLock {
             self._state.requestWasQueued(scheduler)

Diff for: Tests/AsyncHTTPClientTests/RequestBagTests.swift

@@ -125,6 +125,57 @@ final class RequestBagTests: XCTestCase {
 
         XCTAssertNoThrow(try bag.task.futureResult.wait(), "The request has succeeded")
     }
+    
+    func testTaskIsFailedIfWritingFails() {
+        struct TestError: Error, Equatable {}
+        
+        let embeddedEventLoop = EmbeddedEventLoop()
+        defer { XCTAssertNoThrow(try embeddedEventLoop.syncShutdownGracefully()) }
+        let logger = Logger(label: "test")
+
+        let requestBody: HTTPClient.Body = .stream(length: 12) { writer -> EventLoopFuture<Void> in
+            
+            writer.write(.byteBuffer(ByteBuffer(bytes: 0...3))).flatMap { _ -> EventLoopFuture<Void> in
+                embeddedEventLoop.makeFailedFuture(TestError())
+            }
+        }
+
+        var maybeRequest: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://swift.org", method: .POST, body: requestBody))
+        guard let request = maybeRequest else { return XCTFail("Expected to have a request") }
+
+        let delegate = UploadCountingDelegate(eventLoop: embeddedEventLoop)
+        let bag = RequestBag(
+            request: request,
+            eventLoopPreference: .delegate(on: embeddedEventLoop),
+            task: .init(eventLoop: embeddedEventLoop, logger: logger),
+            redirectHandler: nil,
+            connectionDeadline: .now() + .seconds(30),
+            idleReadTimeout: nil,
+            delegate: delegate
+        )
+        XCTAssert(bag.task.eventLoop === embeddedEventLoop)
+
+        let executor = MockRequestExecutor()
+
+        XCTAssertTrue(bag.willExecuteRequest(executor), "false indicates request is cancelled")
+
+        XCTAssertEqual(delegate.hitDidSendRequestHead, 0)
+        bag.requestHeadSent(bag.requestHead)
+        XCTAssertEqual(delegate.hitDidSendRequestHead, 1)
+        XCTAssertEqual(delegate.hitDidSendRequestPart, 0)
+        bag.resumeRequestBodyStream()
+        XCTAssertEqual(delegate.hitDidSendRequestPart, 1)
+        XCTAssertEqual(delegate.hitDidReceiveError, 1)
+        XCTAssertEqual(delegate.lastError as? TestError, TestError())
+        
+        XCTAssertTrue(executor.isCancelled)
+
+        XCTAssertThrowsError(try bag.task.futureResult.wait()) {
+            XCTAssertEqual($0 as? TestError, TestError())
+        }
+    }
+
 
     func testCancelFailsTaskBeforeRequestIsSent() {
         let embeddedEventLoop = EmbeddedEventLoop()
@@ -280,10 +331,12 @@ class UploadCountingDelegate: HTTPClientResponseDelegate {
     private(set) var hitDidSendRequestPart = 0
     private(set) var hitDidSendRequest = 0
     private(set) var hitDidReceiveResponse = 0
+    private(set) var hitDidReceiveError = 0
 
     private(set) var receivedHead: HTTPResponseHead?
     private(set) var lastBodyPart: ByteBuffer?
     private(set) var backpressurePromise: EventLoopPromise<Void>?
+    private(set) var lastError: Error?
 
     init(eventLoop: EventLoop) {
         self.eventLoop = eventLoop
@@ -315,6 +368,11 @@ class UploadCountingDelegate: HTTPClientResponseDelegate {
     func didFinishRequest(task: HTTPClient.Task<Void>) throws {
         self.hitDidReceiveResponse += 1
     }
+    
+    func didReceiveError(task: HTTPClient.Task<Void>, _ error: Error) {
+        self.hitDidReceiveError += 1
+        self.lastError = error
+    }
 
     private func createBackpressurePromise() -> EventLoopFuture<Void> {
         assert(self.backpressurePromise == nil)