test/fuzz: some h1_capture_fuzz_test stability improvements on connec…

test/integration/fake_upstream.cc

@@ -410,7 +410,11 @@ std::string FakeRawConnection::waitForData(uint64_t num_bytes) {
 }
 
 void FakeRawConnection::write(const std::string& data, bool end_stream) {
+  std::unique_lock<std::mutex> lock(lock_);
+  ASSERT_FALSE(disconnected_);
   connection_.dispatcher().post([data, end_stream, this]() -> void {
+    std::unique_lock<std::mutex> lock(lock_);
+    ASSERT_FALSE(disconnected_);
     Buffer::OwnedImpl to_write(data);
     connection_.write(to_write, end_stream);
   });

test/integration/fake_upstream.h

@@ -197,14 +197,18 @@ class FakeConnectionBase : public Network::ConnectionCallbacks {
     connection_.dispatcher().post([this]() -> void { connection_.addConnectionCallbacks(*this); });
   }
   void enableHalfClose(bool enabled);
+  bool connected() const {
+    std::unique_lock<std::mutex> lock(lock_);
+    return !disconnected_;
+  }
 
 protected:
   FakeConnectionBase(QueuedConnectionWrapperPtr connection_wrapper)
       : connection_(connection_wrapper->connection()),
         connection_wrapper_(std::move(connection_wrapper)) {}
 
   Network::Connection& connection_;
-  std::mutex lock_;
+  mutable std::mutex lock_;
   std::condition_variable connection_event_;
   bool disconnected_{};
   bool half_closed_{};

test/integration/h1_capture_fuzz_test.cc

@@ -30,7 +30,7 @@ class H1FuzzIntegrationTest : public HttpIntegrationTest {
       }
       switch (event.event_selector_case()) {
       case test::integration::Event::kDownstreamSendBytes:
-        tcp_client->write(event.downstream_send_bytes());
+        tcp_client->write(event.downstream_send_bytes(), false, false);
         break;
       case test::integration::Event::kDownstreamRecvBytes:
         // TODO(htuch): Should we wait for some data?
@@ -40,11 +40,15 @@ class H1FuzzIntegrationTest : public HttpIntegrationTest {
           fake_upstream_connection = fake_upstreams_[0]->waitForRawConnection(max_wait_ms_);
           // If we timed out, we fail out.
           if (fake_upstream_connection == nullptr) {
-            EXPECT_NE(nullptr, fake_upstream_connection);
             tcp_client->close();
             return;
           }
         }
+        // If we're no longer connected, we're done.
+        if (!fake_upstream_connection->connected()) {
+          tcp_client->close();
+          return;
+        }
         fake_upstream_connection->write(event.upstream_send_bytes());
         break;
       case test::integration::Event::kUpstreamRecvBytes:
@@ -55,7 +59,7 @@ class H1FuzzIntegrationTest : public HttpIntegrationTest {
         break;
       }
     }
-    if (fake_upstream_connection != nullptr) {
+    if (fake_upstream_connection != nullptr && fake_upstream_connection->connected()) {
       fake_upstream_connection->close();
       fake_upstream_connection->waitForDisconnect(true);
     }

test/integration/integration.cc

@@ -132,7 +132,7 @@ IntegrationTcpClient::IntegrationTcpClient(Event::Dispatcher& dispatcher,
   EXPECT_CALL(factory, create_(_, _))
       .WillOnce(Invoke([&](std::function<void()> below_low,
                            std::function<void()> above_high) -> Buffer::Instance* {
-        client_write_buffer_ = new MockWatermarkBuffer(below_low, above_high);
+        client_write_buffer_ = new NiceMock<MockWatermarkBuffer>(below_low, above_high);
         return client_write_buffer_;
       }));
 
@@ -174,11 +174,13 @@ void IntegrationTcpClient::waitForHalfClose() {
 
 void IntegrationTcpClient::readDisable(bool disabled) { connection_->readDisable(disabled); }
 
-void IntegrationTcpClient::write(const std::string& data, bool end_stream) {
+void IntegrationTcpClient::write(const std::string& data, bool end_stream, bool verify) {
   Buffer::OwnedImpl buffer(data);
-  EXPECT_CALL(*client_write_buffer_, move(_));
-  if (!data.empty()) {
-    EXPECT_CALL(*client_write_buffer_, write(_)).Times(AtLeast(1));
+  if (verify) {
+    EXPECT_CALL(*client_write_buffer_, move(_));
+    if (!data.empty()) {
+      EXPECT_CALL(*client_write_buffer_, write(_)).Times(AtLeast(1));
+    }
   }
 
   int bytes_expected = client_write_buffer_->bytes_written() + data.size();

test/integration/integration.h

@@ -80,7 +80,7 @@ class IntegrationTcpClient {
   void waitForDisconnect();
   void waitForHalfClose();
   void readDisable(bool disabled);
-  void write(const std::string& data, bool end_stream = false);
+  void write(const std::string& data, bool end_stream = false, bool verify = true);
   const std::string& data() { return payload_reader_->data(); }
   bool connected() const { return !disconnected_; }