From cbbdf13260baf849e082d05227a7b8c1f6231ab4 Mon Sep 17 00:00:00 2001
From: Hans-Joachim Krauch <achim-k@users.noreply.github.com>
Date: Mon, 14 Aug 2023 18:07:45 -0300
Subject: [PATCH] Reduce smoke test flakiness (#258)

### Public-Facing Changes

Reduce smoke test flakiness

### Description
Attempt to fix flaky smoke tests:

**ROS 1**
- Increased timeouts when waiting for channel to be communicated by the
server
- The ROS 1 server does not listen on graph changes, but instead
periodically queries the ROS master for changes regarding available
topics or services. The max. period between two consecutive ROS master
queries is 5 seconds (default), hence we have to use the same timeout in
the test.

**ROS 2**
- Use std::atomic for message handler that may be called concurrently
- Other minor improvements
---
 ros1_foxglove_bridge/tests/smoke_test.cpp |  4 ++--
 ros2_foxglove_bridge/tests/smoke_test.cpp | 14 +++++++++-----
 2 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/ros1_foxglove_bridge/tests/smoke_test.cpp b/ros1_foxglove_bridge/tests/smoke_test.cpp
index 72afd91..44163c6 100644
--- a/ros1_foxglove_bridge/tests/smoke_test.cpp
+++ b/ros1_foxglove_bridge/tests/smoke_test.cpp
@@ -84,7 +84,7 @@ TEST(SmokeTest, testSubscription) {
     auto client = std::make_shared<foxglove::Client<websocketpp::config::asio_client>>();
     auto channelFuture = foxglove::waitForChannel(client, topic_name);
     ASSERT_EQ(std::future_status::ready, client->connect(URI).wait_for(ONE_SECOND));
-    ASSERT_EQ(std::future_status::ready, channelFuture.wait_for(ONE_SECOND));
+    ASSERT_EQ(std::future_status::ready, channelFuture.wait_for(DEFAULT_TIMEOUT));
     const foxglove::Channel channel = channelFuture.get();
     const foxglove::SubscriptionId subscriptionId = 1;
 
@@ -124,7 +124,7 @@ TEST(SmokeTest, testSubscriptionParallel) {
   for (auto client : clients) {
     auto channelFuture = foxglove::waitForChannel(client, topic_name);
     ASSERT_EQ(std::future_status::ready, client->connect(URI).wait_for(ONE_SECOND));
-    ASSERT_EQ(std::future_status::ready, channelFuture.wait_for(ONE_SECOND));
+    ASSERT_EQ(std::future_status::ready, channelFuture.wait_for(DEFAULT_TIMEOUT));
     const foxglove::Channel channel = channelFuture.get();
     client->subscribe({{subscriptionId, channel.id}});
   }
diff --git a/ros2_foxglove_bridge/tests/smoke_test.cpp b/ros2_foxglove_bridge/tests/smoke_test.cpp
index 17f639e..4480f78 100644
--- a/ros2_foxglove_bridge/tests/smoke_test.cpp
+++ b/ros2_foxglove_bridge/tests/smoke_test.cpp
@@ -484,9 +484,9 @@ TEST_F(ServiceTest, testCallServiceParallel) {
 
   auto serviceFuture = foxglove::waitForService(*clients.begin(), SERVICE_NAME);
   for (auto client : clients) {
-    ASSERT_EQ(std::future_status::ready, client->connect(URI).wait_for(std::chrono::seconds(5)));
+    ASSERT_EQ(std::future_status::ready, client->connect(URI).wait_for(ONE_SECOND));
   }
-  ASSERT_EQ(std::future_status::ready, serviceFuture.wait_for(std::chrono::seconds(5)));
+  ASSERT_EQ(std::future_status::ready, serviceFuture.wait_for(DEFAULT_TIMEOUT));
   const foxglove::Service service = serviceFuture.get();
 
   std_srvs::srv::SetBool::Request requestMsg;
@@ -508,7 +508,7 @@ TEST_F(ServiceTest, testCallServiceParallel) {
   }
 
   for (auto& future : futures) {
-    ASSERT_EQ(std::future_status::ready, future.wait_for(std::chrono::seconds(5)));
+    ASSERT_EQ(std::future_status::ready, future.wait_for(DEFAULT_TIMEOUT));
     foxglove::ServiceResponse response;
     EXPECT_NO_THROW(response = future.get());
     EXPECT_EQ(response.serviceId, request.serviceId);
@@ -558,7 +558,7 @@ TEST(SmokeTest, receiveMessagesOfMultipleTransientLocalPublishers) {
 
   // Set up binary message handler to resolve the promise when all nPub message have been received
   std::promise<void> promise;
-  size_t nReceivedMessages = 0;
+  std::atomic<size_t> nReceivedMessages = 0;
   client->setBinaryMessageHandler([&promise, &nReceivedMessages](const uint8_t*, size_t) {
     if (++nReceivedMessages == nPubs) {
       promise.set_value();
@@ -567,8 +567,12 @@ TEST(SmokeTest, receiveMessagesOfMultipleTransientLocalPublishers) {
 
   // Subscribe to the channel and confirm that the promise resolves
   client->subscribe({{subscriptionId, channel.id}});
-  ASSERT_EQ(std::future_status::ready, promise.get_future().wait_for(ONE_SECOND));
+  EXPECT_EQ(std::future_status::ready, promise.get_future().wait_for(DEFAULT_TIMEOUT));
+  EXPECT_EQ(nReceivedMessages, nPubs);
+  client->unsubscribe({subscriptionId});
 
+  pubs.clear();
+  executor.remove_node(node);
   executor.cancel();
   spinnerThread.join();
 }