From c11a5194d81e16547a5c390ba2e539e08173bfa4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Anarthal=20=28Rub=C3=A9n=20P=C3=A9rez=29?=
 <34971811+anarthal@users.noreply.github.com>
Date: Wed, 14 Jan 2026 21:54:34 +0100
Subject: [PATCH] Reworks test_conn_push and test_conn_push2 (#380)

Adds a test to verify that async_receive (v1) is cancelled on reconnection
Adds a test to verify that a reconnection is triggered when the receive adapter generates an error
Makes the unsubscribe and push adapter tests live only in test_conn_push2, since they test functionality common to async_receive and async_receive2
Migrates all tests to lightweight_test

Entails no functional change
---
 test/test_conn_push.cpp  | 644 +++++++++++++++++++--------------------
 test/test_conn_push2.cpp | 475 ++++++++++++++++-------------
 2 files changed, 582 insertions(+), 537 deletions(-)

diff --git a/test/test_conn_push.cpp b/test/test_conn_push.cpp
index c3f1c339..3b264552 100644
--- a/test/test_conn_push.cpp
+++ b/test/test_conn_push.cpp
@@ -5,255 +5,306 @@
  */
 
 #include <boost/redis/connection.hpp>
+#include <boost/redis/ignore.hpp>
 #include <boost/redis/logger.hpp>
 #include <boost/redis/request.hpp>
+#include <boost/redis/resp3/flat_tree.hpp>
+#include <boost/redis/resp3/node.hpp>
 #include <boost/redis/response.hpp>
 
+#include <boost/asio/error.hpp>
 #include <boost/asio/experimental/channel_error.hpp>
-#include <boost/system/errc.hpp>
-
-#include <string>
-
-#define BOOST_TEST_MODULE conn_push
-#include <boost/test/included/unit_test.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/core/lightweight_test.hpp>
 
 #include "common.hpp"
 
 #include <cstddef>
-#include <iostream>
+#include <functional>
 
 namespace net = boost::asio;
-namespace redis = boost::redis;
-
-using boost::redis::operation;
-using boost::redis::connection;
-using boost::system::error_code;
-using boost::redis::request;
-using boost::redis::response;
-using boost::redis::ignore;
-using boost::redis::ignore_t;
-using boost::system::error_code;
-using boost::redis::logger;
+using namespace boost::redis;
 using namespace std::chrono_literals;
+using boost::system::error_code;
+
+// Focuses on the deprecated async_receive and receive
+// functions. test_conn_push2 covers the newer receive functionality.
 
 namespace {
 
-BOOST_AUTO_TEST_CASE(receives_push_waiting_resps)
+// async_receive is outstanding when a push is received
+void test_async_receive_waiting_for_push()
 {
+   resp3::flat_tree resp;
+   net::io_context ioc;
+   connection conn{ioc};
+   conn.set_receive_response(resp);
+
    request req1;
-   req1.push("HELLO", 3);
    req1.push("PING", "Message1");
+   req1.push("SUBSCRIBE", "test_async_receive_waiting_for_push");
 
    request req2;
-   req2.push("SUBSCRIBE", "channel");
+   req2.push("PING", "Message2");
 
-   request req3;
-   req3.push("PING", "Message2");
-   req3.push("QUIT");
+   bool run_finished = false, push_received = false, exec1_finished = false, exec2_finished = false;
 
-   net::io_context ioc;
-
-   auto conn = std::make_shared<connection>(ioc);
-
-   bool push_received = false, c1_called = false, c2_called = false, c3_called = false;
-
-   auto c3 = [&](error_code ec, std::size_t) {
-      c3_called = true;
-      std::cout << "c3: " << ec.message() << std::endl;
+   auto on_exec2 = [&](error_code ec2, std::size_t) {
+      BOOST_TEST_EQ(ec2, error_code());
+      exec2_finished = true;
+      conn.cancel();
    };
 
-   auto c2 = [&, conn](error_code ec, std::size_t) {
-      c2_called = true;
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req3, ignore, c3);
-   };
+   conn.async_exec(req1, ignore, [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      exec1_finished = true;
+   });
 
-   auto c1 = [&, conn](error_code ec, std::size_t) {
-      c1_called = true;
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c2);
-   };
-
-   conn->async_exec(req1, ignore, c1);
-
-   run(conn, make_test_config(), {});
-
-   conn->async_receive([&, conn](error_code ec, std::size_t) {
-      std::cout << "async_receive" << std::endl;
-      BOOST_TEST(ec == error_code());
+   conn.async_receive([&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_EQ(resp.get_total_msgs(), 1u);
       push_received = true;
-      conn->cancel();
+      conn.async_exec(req2, ignore, on_exec2);
    });
 
-   ioc.run_for(test_timeout);
-
-   BOOST_TEST(push_received);
-   BOOST_TEST(c1_called);
-   BOOST_TEST(c2_called);
-   BOOST_TEST(c3_called);
-}
-
-BOOST_AUTO_TEST_CASE(push_received1)
-{
-   net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
-
-   // Trick: Uses SUBSCRIBE because this command has no response or
-   // better said, its response is a server push, which is what we
-   // want to test. We send two because we want to test both
-   // async_receive and receive.
-   request req;
-   req.push("SUBSCRIBE", "channel1");
-   req.push("SUBSCRIBE", "channel2");
-
-   bool push_received = false, exec_finished = false;
-
-   conn->async_exec(req, ignore, [&, conn](error_code ec, std::size_t) {
-      exec_finished = true;
-      std::cout << "async_exec" << std::endl;
-      BOOST_TEST(ec == error_code());
-   });
-
-   conn->async_receive([&, conn](error_code ec, std::size_t) {
-      push_received = true;
-      std::cout << "(1) async_receive" << std::endl;
-
-      BOOST_TEST(ec == error_code());
-
-      // Receives the second push synchronously.
-      error_code ec2;
-      std::size_t res = 0;
-      res = conn->receive(ec2);
-      BOOST_TEST(!ec2);
-      BOOST_TEST(res != std::size_t(0));
-
-      // Tries to receive a third push synchronously.
-      ec2 = {};
-      res = conn->receive(ec2);
-      BOOST_CHECK_EQUAL(
-         ec2,
-         boost::redis::make_error_code(boost::redis::error::sync_receive_push_failed));
-
-      conn->cancel();
-   });
-
-   run(conn);
-   ioc.run_for(test_timeout);
-
-   BOOST_TEST(exec_finished);
-   BOOST_TEST(push_received);
-}
-
-BOOST_AUTO_TEST_CASE(push_filtered_out)
-{
-   net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
-
-   request req;
-   req.push("HELLO", 3);
-   req.push("PING");
-   req.push("SUBSCRIBE", "channel");
-   req.push("QUIT");
-
-   response<ignore_t, std::string, std::string> resp;
-
-   bool exec_finished = false, push_received = false;
-
-   conn->async_exec(req, resp, [conn, &exec_finished](error_code ec, std::size_t) {
-      exec_finished = true;
-      BOOST_TEST(ec == error_code());
-   });
-
-   conn->async_receive([&, conn](error_code ec, std::size_t) {
-      push_received = true;
-      BOOST_TEST(ec == error_code());
-      conn->cancel(operation::reconnection);
-   });
-
-   run(conn);
-
-   ioc.run_for(test_timeout);
-   BOOST_TEST(exec_finished);
-   BOOST_TEST(push_received);
-
-   BOOST_CHECK_EQUAL(std::get<1>(resp).value(), "PONG");
-   BOOST_CHECK_EQUAL(std::get<2>(resp).value(), "OK");
-}
-
-struct response_error_tag { };
-response_error_tag error_tag_obj;
-
-struct response_error_adapter {
-   void on_init() { }
-   void on_done() { }
-
-   void on_node(
-      boost::redis::resp3::basic_node<std::string_view> const&,
-      boost::system::error_code& ec)
-   {
-      ec = boost::redis::error::incompatible_size;
-   }
-};
-
-auto boost_redis_adapt(response_error_tag&) { return response_error_adapter{}; }
-
-BOOST_AUTO_TEST_CASE(test_push_adapter)
-{
-   net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
-
-   request req;
-   req.push("HELLO", 3);
-   req.push("PING");
-   req.push("SUBSCRIBE", "channel");
-   req.push("PING");
-
-   conn->set_receive_response(error_tag_obj);
-
-   bool push_received = false, exec_finished = false, run_finished = false;
-
-   conn->async_receive([&, conn](error_code ec, std::size_t) {
-      BOOST_CHECK_EQUAL(ec, boost::asio::experimental::error::channel_cancelled);
-      push_received = true;
-   });
-
-   conn->async_exec(req, ignore, [&exec_finished](error_code ec, std::size_t) {
-      BOOST_CHECK_EQUAL(ec, boost::system::errc::errc_t::operation_canceled);
-      exec_finished = true;
-   });
-
-   auto cfg = make_test_config();
-   cfg.reconnect_wait_interval = 0s;
-   conn->async_run(cfg, [&run_finished](error_code ec) {
-      BOOST_CHECK_EQUAL(ec, redis::error::incompatible_size);
+   conn.async_run(make_test_config(), [&](error_code ec) {
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
       run_finished = true;
    });
 
    ioc.run_for(test_timeout);
+
    BOOST_TEST(push_received);
+   BOOST_TEST(exec1_finished);
+   BOOST_TEST(exec2_finished);
+   BOOST_TEST(run_finished);
+}
+
+// A push is already available when async_receive is called
+void test_async_receive_push_available()
+{
+   net::io_context ioc;
+   connection conn{ioc};
+   resp3::flat_tree resp;
+   conn.set_receive_response(resp);
+
+   // SUBSCRIBE doesn't have a response, but causes a push to be delivered.
+   // Add a PING so the overall request has a response.
+   // This ensures that when async_exec completes, the push has been delivered
+   request req;
+   req.push("SUBSCRIBE", "test_async_receive_push_available");
+   req.push("PING", "message");
+
+   bool push_received = false, exec_finished = false, run_finished = false;
+
+   auto on_receive = [&](error_code ec, std::size_t) {
+      push_received = true;
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_EQ(resp.get_total_msgs(), 1u);
+      conn.cancel();
+   };
+
+   conn.async_exec(req, ignore, [&](error_code ec, std::size_t) {
+      exec_finished = true;
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_receive(on_receive);
+   });
+
+   conn.async_run(make_test_config(), [&](error_code ec) {
+      run_finished = true;
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
+   });
+
+   ioc.run_for(test_timeout);
+
+   BOOST_TEST(exec_finished);
+   BOOST_TEST(push_received);
+   BOOST_TEST(run_finished);
+}
+
+// Synchronous receive can be used to try to read a message
+void test_sync_receive()
+{
+   net::io_context ioc;
+   connection conn{ioc};
+   resp3::flat_tree resp;
+   conn.set_receive_response(resp);
+
+   // Subscribing to 2 channels causes 2 pushes to be delivered.
+   // Adding a PING guarantees that after exec finishes, the push has been read
+   request req;
+   req.push("SUBSCRIBE", "test_sync_receive_channel1");
+   req.push("SUBSCRIBE", "test_sync_receive_channel2");
+   req.push("PING", "message");
+
+   bool exec_finished = false, run_finished = false;
+
+   conn.async_exec(req, ignore, [&](error_code ec, std::size_t) {
+      exec_finished = true;
+      BOOST_TEST_EQ(ec, error_code());
+
+      // At this point, the receive response contains all the pushes
+      BOOST_TEST_EQ(resp.get_total_msgs(), 2u);
+
+      // Receive the 1st push synchronously
+      std::size_t push_bytes = conn.receive(ec);
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_GT(push_bytes, 0u);
+
+      // Receive the 2nd push synchronously
+      push_bytes = conn.receive(ec);
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_GT(push_bytes, 0u);
+
+      // There are no more pushes. Trying to receive one more fails
+      push_bytes = conn.receive(ec);
+      BOOST_TEST_EQ(ec, error::sync_receive_push_failed);
+      BOOST_TEST_EQ(push_bytes, 0u);
+
+      conn.cancel();
+   });
+
+   conn.async_run(make_test_config(), [&](error_code ec) {
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
+      run_finished = true;
+   });
+
+   // Trying to receive a push before one is received fails
+   error_code ec;
+   std::size_t push_bytes = conn.receive(ec);
+   BOOST_TEST_EQ(ec, error::sync_receive_push_failed);
+   BOOST_TEST_EQ(push_bytes, 0u);
+
+   ioc.run_for(test_timeout);
+
    BOOST_TEST(exec_finished);
    BOOST_TEST(run_finished);
-
-   // TODO: Reset the ioc reconnect and send a quit to ensure
-   // reconnection is possible after an error.
 }
 
-void launch_push_consumer(std::shared_ptr<connection> conn)
+// async_receive is cancelled every time a reconnection happens,
+// so we can re-establish subscriptions
+struct test_async_receive_cancelled_on_reconnection_impl {
+   net::io_context ioc;
+   connection conn{ioc};
+   resp3::flat_tree resp{};
+   request req_subscribe{}, req_quit{};
+   bool receive_finished = false, quit_finished = false;
+
+   // Subscribe to a channel. This will cause a push to be received
+   void start_subscribe1()
+   {
+      conn.async_exec(req_subscribe, ignore, [this](error_code ec, std::size_t) {
+         BOOST_TEST_EQ(ec, error_code());
+         start_receive1();
+      });
+   }
+
+   // Receive the push triggered by the subscribe
+   void start_receive1()
+   {
+      conn.async_receive([this](error_code ec, std::size_t) {
+         BOOST_TEST_EQ(ec, error_code());
+         BOOST_TEST_EQ(resp.get_total_msgs(), 1u);
+         resp.clear();
+
+         // In parallel, trigger a reconnection and start a receive operation
+         start_receive_reconnection();
+         start_quit();
+      });
+   }
+
+   // The next receive operation will be cancelled by the reconnection
+   void start_receive_reconnection()
+   {
+      conn.async_receive([this](error_code ec, std::size_t) {
+         BOOST_TEST_EQ(ec, net::experimental::channel_errc::channel_cancelled);
+         BOOST_TEST_EQ(resp.get_total_msgs(), 0u);
+         start_subscribe2();
+      });
+   }
+
+   // Trigger a reconnection. This is a "leaf" operation
+   void start_quit()
+   {
+      conn.async_exec(req_quit, ignore, [this](error_code, std::size_t) {
+         quit_finished = true;
+      });
+   }
+
+   // Resubscribe after the reconnection
+   void start_subscribe2()
+   {
+      conn.async_exec(req_subscribe, ignore, [this](error_code ec, std::size_t) {
+         BOOST_TEST_EQ(ec, error_code());
+         start_receive2();
+      });
+   }
+
+   // Receive the push triggered by the 2nd subscribe
+   void start_receive2()
+   {
+      conn.async_receive([this](error_code ec, std::size_t) {
+         BOOST_TEST_EQ(ec, error_code());
+         BOOST_TEST_EQ(resp.get_total_msgs(), 1u);
+         receive_finished = true;
+         conn.cancel();
+      });
+   }
+
+   void run()
+   {
+      req_subscribe.push("SUBSCRIBE", "test_async_receive_cancelled_on_reconnection");
+      req_subscribe.push("PING");
+
+      req_quit.push("QUIT");
+
+      conn.set_receive_response(resp);
+
+      bool run_finished = false;
+
+      start_subscribe1();
+
+      auto cfg = make_test_config();
+      cfg.reconnect_wait_interval = 50ms;  // make the test run faster
+      conn.async_run(cfg, [&](error_code ec) {
+         run_finished = true;
+         BOOST_TEST_EQ(ec, net::error::operation_aborted);
+      });
+
+      ioc.run_for(test_timeout);
+
+      BOOST_TEST(run_finished);
+      BOOST_TEST(receive_finished);
+      BOOST_TEST(quit_finished);
+   }
+};
+
+void test_async_receive_cancelled_on_reconnection()
 {
-   conn->async_receive([conn](error_code ec, std::size_t) {
-      if (ec) {
-         BOOST_TEST(ec == net::experimental::error::channel_cancelled);
-         return;
-      }
-      launch_push_consumer(conn);
-   });
+   test_async_receive_cancelled_on_reconnection_impl{}.run();
 }
 
-BOOST_AUTO_TEST_CASE(many_subscribers)
+// After an async_receive operation finishes, another one can be issued
+void test_consecutive_receives()
 {
-   request req0;
-   req0.get_config().cancel_on_connection_lost = false;
-   req0.push("HELLO", 3);
+   net::io_context ioc;
+   connection conn{ioc};
+   resp3::flat_tree resp;
+   bool push_consumer_finished{false};
+
+   std::function<void()> launch_push_consumer = [&]() {
+      conn.async_receive([&](error_code ec, std::size_t) {
+         if (ec) {
+            BOOST_TEST_EQ(ec, net::experimental::error::channel_cancelled);
+            push_consumer_finished = true;
+            resp.clear();
+            return;
+         }
+         launch_push_consumer();
+      });
+   };
+
+   conn.set_receive_response(resp);
 
    request req1;
    req1.get_config().cancel_on_connection_lost = false;
@@ -263,139 +314,74 @@ BOOST_AUTO_TEST_CASE(many_subscribers)
    req2.get_config().cancel_on_connection_lost = false;
    req2.push("SUBSCRIBE", "channel");
 
-   request req3;
-   req3.get_config().cancel_on_connection_lost = false;
-   req3.push("QUIT");
+   bool exec_finished = false, run_finished = false;
 
-   net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
-
-   bool finished = false;
-
-   auto c11 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->cancel(operation::reconnection);
-      finished = true;
-   };
    auto c10 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req3, ignore, c11);
-   };
-   auto c9 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c10);
-   };
-   auto c8 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req1, ignore, c9);
-   };
-   auto c7 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c8);
-   };
-   auto c6 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c7);
-   };
-   auto c5 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req1, ignore, c6);
-   };
-   auto c4 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c5);
-   };
-   auto c3 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req1, ignore, c4);
-   };
-   auto c2 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c3);
-   };
-   auto c1 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c2);
-   };
-   auto c0 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req1, ignore, c1);
-   };
-
-   conn->async_exec(req0, ignore, c0);
-   launch_push_consumer(conn);
-
-   run(conn, make_test_config(), {});
-
-   ioc.run_for(test_timeout);
-   BOOST_TEST(finished);
-}
-
-BOOST_AUTO_TEST_CASE(test_unsubscribe)
-{
-   net::io_context ioc;
-   connection conn{ioc};
-
-   // Subscribe to 3 channels and 2 patterns. Use CLIENT INFO to verify this took effect
-   request req_subscribe;
-   req_subscribe.push("SUBSCRIBE", "ch1", "ch2", "ch3");
-   req_subscribe.push("PSUBSCRIBE", "ch1*", "ch2*");
-   req_subscribe.push("CLIENT", "INFO");
-
-   // Then, unsubscribe from some of them, and verify again
-   request req_unsubscribe;
-   req_unsubscribe.push("UNSUBSCRIBE", "ch1");
-   req_unsubscribe.push("PUNSUBSCRIBE", "ch2*");
-   req_unsubscribe.push("CLIENT", "INFO");
-
-   // Finally, ping to verify that the connection is still usable
-   request req_ping;
-   req_ping.push("PING", "test_unsubscribe");
-
-   response<std::string> resp_subscribe, resp_unsubscribe, resp_ping;
-
-   bool subscribe_finished = false, unsubscribe_finished = false, ping_finished = false,
-        run_finished = false;
-
-   auto on_ping = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      ping_finished = true;
-      BOOST_TEST(std::get<0>(resp_ping).has_value());
-      BOOST_TEST(std::get<0>(resp_ping).value() == "test_unsubscribe");
+      BOOST_TEST_EQ(ec, error_code());
+      exec_finished = true;
       conn.cancel();
    };
-
-   auto on_unsubscribe = [&](error_code ec, std::size_t) {
-      unsubscribe_finished = true;
-      BOOST_TEST(ec == error_code());
-      BOOST_TEST(std::get<0>(resp_unsubscribe).has_value());
-      BOOST_TEST(find_client_info(std::get<0>(resp_unsubscribe).value(), "sub") == "2");
-      BOOST_TEST(find_client_info(std::get<0>(resp_unsubscribe).value(), "psub") == "1");
-      conn.async_exec(req_ping, resp_ping, on_ping);
+   auto c9 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c10);
+   };
+   auto c8 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req1, ignore, c9);
+   };
+   auto c7 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c8);
+   };
+   auto c6 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c7);
+   };
+   auto c5 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req1, ignore, c6);
+   };
+   auto c4 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c5);
+   };
+   auto c3 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req1, ignore, c4);
+   };
+   auto c2 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c3);
+   };
+   auto c1 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c2);
    };
 
-   auto on_subscribe = [&](error_code ec, std::size_t) {
-      subscribe_finished = true;
-      BOOST_TEST(ec == error_code());
-      BOOST_TEST(std::get<0>(resp_subscribe).has_value());
-      BOOST_TEST(find_client_info(std::get<0>(resp_subscribe).value(), "sub") == "3");
-      BOOST_TEST(find_client_info(std::get<0>(resp_subscribe).value(), "psub") == "2");
-      conn.async_exec(req_unsubscribe, resp_unsubscribe, on_unsubscribe);
-   };
+   conn.async_exec(req1, ignore, c1);
+   launch_push_consumer();
 
-   conn.async_exec(req_subscribe, resp_subscribe, on_subscribe);
-
-   conn.async_run(make_test_config(), [&run_finished](error_code ec) {
-      BOOST_TEST(ec == net::error::operation_aborted);
+   conn.async_run(make_test_config(), [&](error_code ec) {
       run_finished = true;
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
    });
 
    ioc.run_for(test_timeout);
 
-   BOOST_TEST(subscribe_finished);
-   BOOST_TEST(unsubscribe_finished);
-   BOOST_TEST(ping_finished);
+   BOOST_TEST(exec_finished);
    BOOST_TEST(run_finished);
-}
+   BOOST_TEST(push_consumer_finished);
+};
 
 }  // namespace
+
+int main()
+{
+   test_async_receive_waiting_for_push();
+   test_async_receive_push_available();
+   test_sync_receive();
+   test_async_receive_cancelled_on_reconnection();
+   test_consecutive_receives();
+
+   return boost::report_errors();
+}
diff --git a/test/test_conn_push2.cpp b/test/test_conn_push2.cpp
index 9cf638e9..5c260f8c 100644
--- a/test/test_conn_push2.cpp
+++ b/test/test_conn_push2.cpp
@@ -10,218 +10,209 @@
 #include <boost/redis/resp3/flat_tree.hpp>
 #include <boost/redis/response.hpp>
 
+#include <boost/asio/error.hpp>
 #include <boost/asio/experimental/channel_error.hpp>
-#include <boost/system/errc.hpp>
-
-#include <set>
-#include <string>
-#include <string_view>
-
-#define BOOST_TEST_MODULE conn_push
-#include <boost/test/included/unit_test.hpp>
+#include <boost/core/lightweight_test.hpp>
 
 #include "common.hpp"
 
 #include <cstddef>
-#include <iostream>
+#include <functional>
+#include <iterator>
+#include <set>
+#include <string>
+#include <string_view>
 
 namespace net = boost::asio;
-namespace redis = boost::redis;
-
-using boost::redis::operation;
-using boost::redis::connection;
-using boost::system::error_code;
-using boost::redis::request;
-using boost::redis::response;
-using boost::redis::resp3::flat_tree;
-using boost::redis::ignore;
-using boost::redis::ignore_t;
-using boost::system::error_code;
-using boost::redis::logger;
-using boost::redis::resp3::node_view;
-using boost::redis::resp3::type;
+using namespace boost::redis;
 using namespace std::chrono_literals;
+using boost::system::error_code;
+using resp3::flat_tree;
+using resp3::node_view;
+using resp3::type;
+
+// Covers all receive functionality except for the deprecated
+// async_receive and receive functions.
 
 namespace {
 
-BOOST_AUTO_TEST_CASE(receives_push_waiting_resps)
+// async_receive2 is outstanding when a push is received
+void test_async_receive2_waiting_for_push()
 {
+   resp3::flat_tree resp;
+   net::io_context ioc;
+   connection conn{ioc};
+   conn.set_receive_response(resp);
+
    request req1;
-   req1.push("HELLO", 3);
    req1.push("PING", "Message1");
+   req1.push("SUBSCRIBE", "test_async_receive_waiting_for_push");
 
    request req2;
-   req2.push("SUBSCRIBE", "channel");
+   req2.push("PING", "Message2");
 
-   request req3;
-   req3.push("PING", "Message2");
-   req3.push("QUIT");
+   bool run_finished = false, push_received = false, exec1_finished = false, exec2_finished = false;
 
-   net::io_context ioc;
-
-   auto conn = std::make_shared<connection>(ioc);
-
-   bool push_received = false, c1_called = false, c2_called = false, c3_called = false;
-
-   auto c3 = [&](error_code ec, std::size_t) {
-      c3_called = true;
-      std::cout << "c3: " << ec.message() << std::endl;
+   auto on_exec2 = [&](error_code ec2, std::size_t) {
+      BOOST_TEST_EQ(ec2, error_code());
+      exec2_finished = true;
+      conn.cancel();
    };
 
-   auto c2 = [&, conn](error_code ec, std::size_t) {
-      c2_called = true;
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req3, ignore, c3);
-   };
+   conn.async_exec(req1, ignore, [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      exec1_finished = true;
+   });
 
-   auto c1 = [&, conn](error_code ec, std::size_t) {
-      c1_called = true;
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c2);
-   };
-
-   conn->async_exec(req1, ignore, c1);
-
-   run(conn, make_test_config(), {});
-
-   conn->async_receive2([&, conn](error_code ec) {
-      std::cout << "async_receive2" << std::endl;
-      BOOST_TEST(ec == error_code());
+   conn.async_receive2([&](error_code ec) {
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_EQ(resp.get_total_msgs(), 1u);
       push_received = true;
-      conn->cancel();
+      conn.async_exec(req2, ignore, on_exec2);
+   });
+
+   conn.async_run(make_test_config(), [&](error_code ec) {
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
+      run_finished = true;
    });
 
    ioc.run_for(test_timeout);
 
    BOOST_TEST(push_received);
-   BOOST_TEST(c1_called);
-   BOOST_TEST(c2_called);
-   BOOST_TEST(c3_called);
+   BOOST_TEST(exec1_finished);
+   BOOST_TEST(exec2_finished);
+   BOOST_TEST(run_finished);
 }
 
-BOOST_AUTO_TEST_CASE(push_received1)
+// A push is already available when async_receive2 is called
+void test_async_receive2_push_available()
 {
    net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
+   connection conn{ioc};
+   resp3::flat_tree resp;
+   conn.set_receive_response(resp);
 
-   flat_tree resp;
-   conn->set_receive_response(resp);
-
-   // Trick: Uses SUBSCRIBE because this command has no response or
-   // better said, its response is a server push, which is what we
-   // want to test.
+   // SUBSCRIBE doesn't have a response, but causes a push to be delivered.
+   // Add a PING so the overall request has a response.
+   // This ensures that when async_exec completes, the push has been delivered
    request req;
-   req.push("SUBSCRIBE", "channel1");
-   req.push("SUBSCRIBE", "channel2");
+   req.push("SUBSCRIBE", "test_async_receive_push_available");
+   req.push("PING", "message");
 
-   bool push_received = false, exec_finished = false;
+   bool push_received = false, exec_finished = false, run_finished = false;
 
-   conn->async_exec(req, ignore, [&, conn](error_code ec, std::size_t) {
-      exec_finished = true;
-      std::cout << "async_exec" << std::endl;
-      BOOST_TEST(ec == error_code());
-   });
-
-   conn->async_receive2([&, conn](error_code ec) {
+   auto on_receive = [&](error_code ec, std::size_t) {
       push_received = true;
-      std::cout << "async_receive2" << std::endl;
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_EQ(resp.get_total_msgs(), 1u);
+      conn.cancel();
+   };
 
-      BOOST_TEST(ec == error_code());
-
-      BOOST_CHECK_EQUAL(resp.get_total_msgs(), 2u);
-
-      conn->cancel();
+   conn.async_exec(req, ignore, [&](error_code ec, std::size_t) {
+      exec_finished = true;
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_receive(on_receive);
+   });
+
+   conn.async_run(make_test_config(), [&](error_code ec) {
+      run_finished = true;
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
    });
 
-   run(conn);
    ioc.run_for(test_timeout);
 
    BOOST_TEST(exec_finished);
    BOOST_TEST(push_received);
+   BOOST_TEST(run_finished);
 }
 
-BOOST_AUTO_TEST_CASE(push_filtered_out)
+// A push may be interleaved between regular responses.
+// It is handed to the receive adapter (filtered out).
+void test_exec_push_interleaved()
 {
    net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
+   connection conn{ioc};
+   resp3::flat_tree receive_resp;
+   conn.set_receive_response(receive_resp);
 
    request req;
-   req.push("HELLO", 3);
-   req.push("PING");
-   req.push("SUBSCRIBE", "channel");
-   req.push("QUIT");
+   req.push("PING", "msg1");
+   req.push("SUBSCRIBE", "test_exec_push_interleaved");
+   req.push("PING", "msg2");
 
-   response<ignore_t, std::string, std::string> resp;
+   response<std::string, std::string> resp;
 
-   bool exec_finished = false, push_received = false;
+   bool exec_finished = false, push_received = false, run_finished = false;
 
-   conn->async_exec(req, resp, [conn, &exec_finished](error_code ec, std::size_t) {
+   conn.async_exec(req, resp, [&](error_code ec, std::size_t) {
       exec_finished = true;
-      BOOST_TEST(ec == error_code());
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_EQ(std::get<0>(resp).value(), "msg1");
+      BOOST_TEST_EQ(std::get<1>(resp).value(), "msg2");
+      conn.cancel();
    });
 
-   conn->async_receive2([&, conn](error_code ec) {
+   conn.async_receive2([&](error_code ec) {
       push_received = true;
-      BOOST_TEST(ec == error_code());
-      conn->cancel(operation::reconnection);
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_EQ(receive_resp.get_total_msgs(), 1u);
    });
 
-   run(conn);
+   conn.async_run(make_test_config(), [&](error_code ec) {
+      run_finished = true;
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
+   });
 
    ioc.run_for(test_timeout);
+
    BOOST_TEST(exec_finished);
    BOOST_TEST(push_received);
-
-   BOOST_CHECK_EQUAL(std::get<1>(resp).value(), "PONG");
-   BOOST_CHECK_EQUAL(std::get<2>(resp).value(), "OK");
+   BOOST_TEST(run_finished);
 }
 
+// An adapter that always errors
 struct response_error_tag { };
 response_error_tag error_tag_obj;
 
 struct response_error_adapter {
    void on_init() { }
    void on_done() { }
-
-   void on_node(
-      boost::redis::resp3::basic_node<std::string_view> const&,
-      boost::system::error_code& ec)
-   {
-      ec = boost::redis::error::incompatible_size;
-   }
+   void on_node(node_view const&, error_code& ec) { ec = error::incompatible_size; }
 };
 
 auto boost_redis_adapt(response_error_tag&) { return response_error_adapter{}; }
 
-BOOST_AUTO_TEST_CASE(test_push_adapter)
+// If the push adapter returns an error, the connection is torn down
+void test_push_adapter_error()
 {
    net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
+   connection conn{ioc};
+   conn.set_receive_response(error_tag_obj);
 
    request req;
-   req.push("HELLO", 3);
    req.push("PING");
    req.push("SUBSCRIBE", "channel");
    req.push("PING");
 
-   conn->set_receive_response(error_tag_obj);
-
    bool push_received = false, exec_finished = false, run_finished = false;
 
-   conn->async_receive2([&, conn](error_code ec) {
-      BOOST_CHECK_EQUAL(ec, boost::asio::experimental::error::channel_cancelled);
+   // async_receive2 is cancelled every reconnection cycle
+   conn.async_receive2([&](error_code ec) {
+      BOOST_TEST_EQ(ec, net::experimental::error::channel_cancelled);
       push_received = true;
    });
 
-   conn->async_exec(req, ignore, [&exec_finished](error_code ec, std::size_t) {
-      BOOST_CHECK_EQUAL(ec, boost::system::errc::errc_t::operation_canceled);
+   // The request is cancelled because the PING response isn't processed
+   // by the time the error is generated
+   conn.async_exec(req, ignore, [&exec_finished](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
       exec_finished = true;
    });
 
    auto cfg = make_test_config();
-   cfg.reconnect_wait_interval = 0s;
-   conn->async_run(cfg, [&run_finished](error_code ec) {
-      BOOST_CHECK_EQUAL(ec, redis::error::incompatible_size);
+   cfg.reconnect_wait_interval = 0s;  // so we can validate the generated error
+   conn.async_run(cfg, [&run_finished](error_code ec) {
+      BOOST_TEST_EQ(ec, error::incompatible_size);
       run_finished = true;
    });
 
@@ -229,27 +220,83 @@ BOOST_AUTO_TEST_CASE(test_push_adapter)
    BOOST_TEST(push_received);
    BOOST_TEST(exec_finished);
    BOOST_TEST(run_finished);
-
-   // TODO: Reset the ioc reconnect and send a quit to ensure
-   // reconnection is possible after an error.
 }
 
-void launch_push_consumer(std::shared_ptr<connection> conn)
+// A push response error triggers a reconnection
+void test_push_adapter_error_reconnection()
 {
-   conn->async_receive2([conn](error_code ec) {
-      if (ec) {
-         BOOST_TEST(ec == net::experimental::error::channel_cancelled);
-         return;
-      }
-      launch_push_consumer(conn);
+   net::io_context ioc;
+   connection conn{ioc};
+   conn.set_receive_response(error_tag_obj);
+
+   request req;
+   req.push("PING");
+   req.push("SUBSCRIBE", "channel");
+   req.push("PING");
+
+   request req2;
+   req2.push("PING", "msg2");
+   req2.get_config().cancel_if_unresponded = false;
+
+   response<std::string> resp;
+
+   bool push_received = false, exec_finished = false, run_finished = false;
+
+   // async_receive2 is cancelled every reconnection cycle
+   conn.async_receive2([&](error_code ec) {
+      BOOST_TEST_EQ(ec, net::experimental::error::channel_cancelled);
+      push_received = true;
    });
+
+   auto on_exec2 = [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, error_code());
+      BOOST_TEST_EQ(std::get<0>(resp).value(), "msg2");
+      exec_finished = true;
+      conn.cancel();
+   };
+
+   // The request is cancelled because the PING response isn't processed
+   // by the time the error is generated
+   conn.async_exec(req, ignore, [&](error_code ec, std::size_t) {
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
+      conn.async_exec(req2, resp, on_exec2);
+   });
+
+   auto cfg = make_test_config();
+   cfg.reconnect_wait_interval = 50ms;  // make the test run faster
+   conn.async_run(cfg, [&run_finished](error_code ec) {
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
+      run_finished = true;
+   });
+
+   ioc.run_for(test_timeout);
+
+   BOOST_TEST(push_received);
+   BOOST_TEST(exec_finished);
+   BOOST_TEST(run_finished);
 }
 
-BOOST_AUTO_TEST_CASE(many_subscribers)
+// After an async_receive2 operation finishes, another one can be issued
+void test_consecutive_receives()
 {
-   request req0;
-   req0.get_config().cancel_on_connection_lost = false;
-   req0.push("HELLO", 3);
+   net::io_context ioc;
+   connection conn{ioc};
+   resp3::flat_tree resp;
+   bool push_consumer_finished{false};
+
+   std::function<void()> launch_push_consumer = [&]() {
+      conn.async_receive2([&](error_code ec) {
+         if (ec) {
+            BOOST_TEST_EQ(ec, net::experimental::error::channel_cancelled);
+            push_consumer_finished = true;
+            resp.clear();
+            return;
+         }
+         launch_push_consumer();
+      });
+   };
+
+   conn.set_receive_response(resp);
 
    request req1;
    req1.get_config().cancel_on_connection_lost = false;
@@ -259,75 +306,67 @@ BOOST_AUTO_TEST_CASE(many_subscribers)
    req2.get_config().cancel_on_connection_lost = false;
    req2.push("SUBSCRIBE", "channel");
 
-   request req3;
-   req3.get_config().cancel_on_connection_lost = false;
-   req3.push("QUIT");
+   bool exec_finished = false, run_finished = false;
 
-   net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
-
-   bool finished = false;
-
-   auto c11 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->cancel(operation::reconnection);
-      finished = true;
-   };
    auto c10 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req3, ignore, c11);
+      BOOST_TEST_EQ(ec, error_code());
+      exec_finished = true;
+      conn.cancel();
    };
    auto c9 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c10);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c10);
    };
    auto c8 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req1, ignore, c9);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req1, ignore, c9);
    };
    auto c7 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c8);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c8);
    };
    auto c6 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c7);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c7);
    };
    auto c5 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req1, ignore, c6);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req1, ignore, c6);
    };
    auto c4 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c5);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c5);
    };
    auto c3 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req1, ignore, c4);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req1, ignore, c4);
    };
    auto c2 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c3);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c3);
    };
    auto c1 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req2, ignore, c2);
-   };
-   auto c0 = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
-      conn->async_exec(req1, ignore, c1);
+      BOOST_TEST_EQ(ec, error_code());
+      conn.async_exec(req2, ignore, c2);
    };
 
-   conn->async_exec(req0, ignore, c0);
-   launch_push_consumer(conn);
+   conn.async_exec(req1, ignore, c1);
+   launch_push_consumer();
 
-   run(conn, make_test_config(), {});
+   conn.async_run(make_test_config(), [&](error_code ec) {
+      run_finished = true;
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
+   });
 
    ioc.run_for(test_timeout);
-   BOOST_TEST(finished);
+
+   BOOST_TEST(exec_finished);
+   BOOST_TEST(run_finished);
+   BOOST_TEST(push_consumer_finished);
 }
 
-BOOST_AUTO_TEST_CASE(test_unsubscribe)
+// UNSUBSCRIBE and PUNSUBSCRIBE work
+void test_unsubscribe()
 {
    net::io_context ioc;
    connection conn{ioc};
@@ -354,35 +393,35 @@ BOOST_AUTO_TEST_CASE(test_unsubscribe)
         run_finished = false;
 
    auto on_ping = [&](error_code ec, std::size_t) {
-      BOOST_TEST(ec == error_code());
+      BOOST_TEST_EQ(ec, error_code());
       ping_finished = true;
       BOOST_TEST(std::get<0>(resp_ping).has_value());
-      BOOST_TEST(std::get<0>(resp_ping).value() == "test_unsubscribe");
+      BOOST_TEST_EQ(std::get<0>(resp_ping).value(), "test_unsubscribe");
       conn.cancel();
    };
 
    auto on_unsubscribe = [&](error_code ec, std::size_t) {
       unsubscribe_finished = true;
-      BOOST_TEST(ec == error_code());
+      BOOST_TEST_EQ(ec, error_code());
       BOOST_TEST(std::get<0>(resp_unsubscribe).has_value());
-      BOOST_TEST(find_client_info(std::get<0>(resp_unsubscribe).value(), "sub") == "2");
-      BOOST_TEST(find_client_info(std::get<0>(resp_unsubscribe).value(), "psub") == "1");
+      BOOST_TEST_EQ(find_client_info(std::get<0>(resp_unsubscribe).value(), "sub"), "2");
+      BOOST_TEST_EQ(find_client_info(std::get<0>(resp_unsubscribe).value(), "psub"), "1");
       conn.async_exec(req_ping, resp_ping, on_ping);
    };
 
    auto on_subscribe = [&](error_code ec, std::size_t) {
       subscribe_finished = true;
-      BOOST_TEST(ec == error_code());
+      BOOST_TEST_EQ(ec, error_code());
       BOOST_TEST(std::get<0>(resp_subscribe).has_value());
-      BOOST_TEST(find_client_info(std::get<0>(resp_subscribe).value(), "sub") == "3");
-      BOOST_TEST(find_client_info(std::get<0>(resp_subscribe).value(), "psub") == "2");
+      BOOST_TEST_EQ(find_client_info(std::get<0>(resp_subscribe).value(), "sub"), "3");
+      BOOST_TEST_EQ(find_client_info(std::get<0>(resp_subscribe).value(), "psub"), "2");
       conn.async_exec(req_unsubscribe, resp_unsubscribe, on_unsubscribe);
    };
 
    conn.async_exec(req_subscribe, resp_subscribe, on_subscribe);
 
    conn.async_run(make_test_config(), [&run_finished](error_code ec) {
-      BOOST_TEST(ec == net::error::operation_aborted);
+      BOOST_TEST_EQ(ec, net::error::operation_aborted);
       run_finished = true;
    });
 
@@ -394,12 +433,12 @@ BOOST_AUTO_TEST_CASE(test_unsubscribe)
    BOOST_TEST(run_finished);
 }
 
-class test_pubsub_state_restoration_ {
+struct test_pubsub_state_restoration_impl {
    net::io_context ioc;
    connection conn{ioc};
-   request req;
-   response<std::string> resp_str;
-   flat_tree resp_push;
+   request req{};
+   response<std::string> resp_str{};
+   flat_tree resp_push{};
    bool exec_finished = false;
 
    void check_subscriptions()
@@ -408,13 +447,13 @@ class test_pubsub_state_restoration_ {
       std::set<std::string_view> seen_channels, seen_patterns;
       for (auto it = resp_push.get_view().begin(); it != resp_push.get_view().end();) {
          // The root element should be a push
-         BOOST_TEST_REQUIRE(it->data_type == type::push);
-         BOOST_TEST_REQUIRE(it->aggregate_size >= 2u);
-         BOOST_TEST_REQUIRE((++it != resp_push.get_view().end()));
+         BOOST_TEST_EQ(it->data_type, type::push);
+         BOOST_TEST_GE(it->aggregate_size, 2u);
+         BOOST_TEST(++it != resp_push.get_view().end());
 
          // The next element should be the message type
          std::string_view msg_type = it->value;
-         BOOST_TEST_REQUIRE((++it != resp_push.get_view().end()));
+         BOOST_TEST(++it != resp_push.get_view().end());
 
          // The next element is the channel or pattern
          if (msg_type == "subscribe")
@@ -430,8 +469,16 @@ class test_pubsub_state_restoration_ {
       const std::string_view expected_channels[] = {"ch1", "ch3", "ch5"};
       const std::string_view expected_patterns[] = {"ch1*", "ch3*", "ch4*", "ch8*"};
 
-      BOOST_TEST(seen_channels == expected_channels, boost::test_tools::per_element());
-      BOOST_TEST(seen_patterns == expected_patterns, boost::test_tools::per_element());
+      BOOST_TEST_ALL_EQ(
+         seen_channels.begin(),
+         seen_channels.end(),
+         std::begin(expected_channels),
+         std::end(expected_channels));
+      BOOST_TEST_ALL_EQ(
+         seen_patterns.begin(),
+         seen_patterns.end(),
+         std::begin(expected_patterns),
+         std::end(expected_patterns));
    }
 
    void sub1()
@@ -441,7 +488,7 @@ class test_pubsub_state_restoration_ {
       req.subscribe({"ch1", "ch2", "ch3"});              // active: 1, 2, 3
       req.psubscribe({"ch1*", "ch2*", "ch3*", "ch4*"});  // active: 1, 2, 3, 4
       conn.async_exec(req, ignore, [this](error_code ec, std::size_t) {
-         BOOST_TEST(ec == error_code());
+         BOOST_TEST_EQ(ec, error_code());
          unsub();
       });
    }
@@ -454,7 +501,7 @@ class test_pubsub_state_restoration_ {
       req.unsubscribe({"ch2", "ch1", "ch5"});      // active: 3
       req.punsubscribe({"ch2*", "ch4*", "ch9*"});  // active: 1, 3
       conn.async_exec(req, ignore, [this](error_code ec, std::size_t) {
-         BOOST_TEST(ec == error_code());
+         BOOST_TEST_EQ(ec, error_code());
          sub2();
       });
    }
@@ -476,12 +523,12 @@ class test_pubsub_state_restoration_ {
       req.push("CLIENT", "INFO");
 
       conn.async_exec(req, resp_str, [this](error_code ec, std::size_t) {
-         BOOST_TEST(ec == error_code());
+         BOOST_TEST_EQ(ec, error_code());
 
          // We are subscribed to 4 channels and 5 patterns
          BOOST_TEST(std::get<0>(resp_str).has_value());
-         BOOST_TEST(find_client_info(std::get<0>(resp_str).value(), "sub") == "4");
-         BOOST_TEST(find_client_info(std::get<0>(resp_str).value(), "psub") == "5");
+         BOOST_TEST_EQ(find_client_info(std::get<0>(resp_str).value(), "sub"), "4");
+         BOOST_TEST_EQ(find_client_info(std::get<0>(resp_str).value(), "psub"), "5");
 
          resp_push.clear();
 
@@ -507,12 +554,12 @@ class test_pubsub_state_restoration_ {
       req.get_config().cancel_if_unresponded = false;
 
       conn.async_exec(req, resp_str, [this](error_code ec, std::size_t) {
-         BOOST_TEST(ec == error_code());
+         BOOST_TEST_EQ(ec, error_code());
 
          // We are subscribed to 3 channels and 4 patterns (1 of each didn't survive reconnection)
          BOOST_TEST(std::get<0>(resp_str).has_value());
-         BOOST_TEST(find_client_info(std::get<0>(resp_str).value(), "sub") == "3");
-         BOOST_TEST(find_client_info(std::get<0>(resp_str).value(), "psub") == "4");
+         BOOST_TEST_EQ(find_client_info(std::get<0>(resp_str).value(), "sub"), "3");
+         BOOST_TEST_EQ(find_client_info(std::get<0>(resp_str).value(), "psub"), "4");
 
          // We have received pushes confirming it
          check_subscriptions();
@@ -522,7 +569,6 @@ class test_pubsub_state_restoration_ {
       });
    }
 
-public:
    void run()
    {
       conn.set_receive_response(resp_push);
@@ -533,7 +579,7 @@ public:
       // Start running
       bool run_finished = false;
       conn.async_run(make_test_config(), [&run_finished](error_code ec) {
-         BOOST_TEST(ec == net::error::operation_aborted);
+         BOOST_TEST_EQ(ec, net::error::operation_aborted);
          run_finished = true;
       });
 
@@ -544,7 +590,20 @@ public:
       BOOST_TEST(run_finished);
    }
 };
-
-BOOST_AUTO_TEST_CASE(test_pubsub_state_restoration) { test_pubsub_state_restoration_().run(); }
+void test_pubsub_state_restoration() { test_pubsub_state_restoration_impl{}.run(); }
 
 }  // namespace
+
+int main()
+{
+   test_async_receive2_waiting_for_push();
+   test_async_receive2_push_available();
+   test_exec_push_interleaved();
+   test_push_adapter_error();
+   test_push_adapter_error_reconnection();
+   test_consecutive_receives();
+   test_unsubscribe();
+   test_pubsub_state_restoration();
+
+   return boost::report_errors();
+}