Connect_op basic unit tests

Summary:
related to T12015
- added basic unit tests (without auth for a smaller diff)
- further simplified tests in publish_receive.cpp

Reviewers: ivica

Reviewed By: ivica

Subscribers: miljen, iljazovic

Differential Revision: https://repo.mireo.local/D27417

include/async_mqtt5/impl/connect_op.hpp

@@ -214,7 +214,7 @@ public:
 		if (ec)
 			return complete(ec);
 
-		_buffer_ptr = std::make_unique<std::string>(min_packet_sz, 0);
+		_buffer_ptr = std::make_unique<std::string>(min_packet_sz, char(0));
 
 		auto buff = asio::buffer(_buffer_ptr->data(), min_packet_sz);
 		asio::async_read(
@@ -240,7 +240,7 @@ public:
 		);
 
 		if (!varlen)
-			complete(asio::error::try_again);
+			return complete(asio::error::try_again);
 
 		auto varlen_sz = std::distance(_buffer_ptr->cbegin() + 1, varlen_ptr);
 		auto remain_len = *varlen -

test/integration/malformed_packet.cpp

@@ -8,6 +8,8 @@
 
 using namespace async_mqtt5;
 
+BOOST_AUTO_TEST_SUITE(malformed_packet/* , *boost::unit_test::disabled()*/)
+
 struct shared_test_data {
 	const std::string connect = encoders::encode_connect(
 		"", std::nullopt, std::nullopt, 10, false, {}, std::nullopt
@@ -32,9 +34,7 @@ struct shared_test_data {
 
 };
 
-BOOST_FIXTURE_TEST_SUITE(malformed_packet, shared_test_data/* , *boost::unit_test::disabled()*/)
-
-BOOST_AUTO_TEST_CASE(test_malformed_publish) {
+BOOST_FIXTURE_TEST_CASE(test_malformed_publish, shared_test_data) {
 	using test::after;
 	using std::chrono_literals::operator ""ms;
 
@@ -79,7 +79,7 @@ BOOST_AUTO_TEST_CASE(test_malformed_publish) {
 	BOOST_CHECK(broker.received_all_expected());
 }
 
-BOOST_AUTO_TEST_CASE(malformed_puback) {
+BOOST_FIXTURE_TEST_CASE(malformed_puback, shared_test_data) {
 	using test::after;
 	using std::chrono_literals::operator ""ms;
 
@@ -151,7 +151,7 @@ BOOST_AUTO_TEST_CASE(malformed_puback) {
 }
 
 
-BOOST_AUTO_TEST_CASE(malformed_pubrel) {
+BOOST_FIXTURE_TEST_CASE(malformed_pubrel, shared_test_data) {
 	using test::after;
 	using std::chrono_literals::operator ""ms;
 
@@ -219,7 +219,7 @@ BOOST_AUTO_TEST_CASE(malformed_pubrel) {
 	BOOST_CHECK(broker.received_all_expected());
 }
 
-BOOST_AUTO_TEST_CASE(malformed_pubrec) {
+BOOST_FIXTURE_TEST_CASE(malformed_pubrec, shared_test_data) {
 	using test::after;
 	using std::chrono_literals::operator ""ms;
 
@@ -286,7 +286,7 @@ BOOST_AUTO_TEST_CASE(malformed_pubrec) {
 }
 
 
-BOOST_AUTO_TEST_CASE(malformed_pubcomp) {
+BOOST_FIXTURE_TEST_CASE(malformed_pubcomp, shared_test_data) {
 	using test::after;
 	using std::chrono_literals::operator ""ms;
 

test/integration/publish_receive.cpp

@@ -14,147 +14,43 @@ using namespace async_mqtt5;
 
 BOOST_AUTO_TEST_SUITE(publish_receive/*, *boost::unit_test::disabled()*/)
 
-template <qos_e qos>
-void receive_publish() {
-	using test::after;
-	using std::chrono_literals::operator ""ms;
-
-	constexpr int expected_handlers_called = 1;
-	int handlers_called = 0;
-
-	std::string topic = "topic";
-	std::string payload = "payload";
-
-	// packets
-	auto connect = encoders::encode_connect(
-		"", std::nullopt, std::nullopt, 10, false, {}, std::nullopt
-	);
-	auto connack = encoders::encode_connack(
-		false, reason_codes::success.value(), {}
-	);
-
-	auto publish = encoders::encode_publish(
-		1, topic, payload, qos, retain_e::no, dup_e::no, {}
-	);
-
-	auto puback = encoders::encode_puback(1, reason_codes::success.value(), {});
-
-	auto pubrec = encoders::encode_pubrec(1, reason_codes::success.value(), {});
-	auto pubrel = encoders::encode_pubrel(1, reason_codes::success.value(), {});
-	auto pubcomp = encoders::encode_pubcomp(1, reason_codes::success.value(), {});
-
-	test::msg_exchange broker_side;
-	error_code success {};
-
-	broker_side
-		.expect(connect)
-			.complete_with(success, after(0ms))
-			.reply_with(connack, after(0ms))
-		.send(publish, after(10ms));
-
-	if constexpr (qos == qos_e::at_least_once) {
-		broker_side
-			.expect(puback)
-				.complete_with(success, after(1ms));
-	} else if constexpr (qos == qos_e::exactly_once) {
-		broker_side
-			.expect(pubrec)
-				.complete_with(success, after(1ms))
-				.reply_with(pubrel, after(2ms))
-			.expect(pubcomp)
-				.complete_with(success, after(1ms));
-	}
-
-	asio::io_context ioc;
-	auto executor = ioc.get_executor();
-	auto& broker = asio::make_service<test::test_broker>(
-		ioc, executor, std::move(broker_side)
-	);
-
-	using client_type = mqtt_client<test::test_stream>;
-	client_type c(executor, "");
-	c.brokers("127.0.0.1")
-		.async_run(asio::detached);
-
-	c.async_receive(
-		[&](
-			error_code ec,
-			std::string rec_topic, std::string rec_payload,
-			publish_props
-		) {
-			++handlers_called;
-
-			BOOST_CHECK_MESSAGE(!ec, ec.message());
-			BOOST_CHECK_EQUAL(topic, rec_topic);
-			BOOST_CHECK_EQUAL(payload,  rec_payload);
-
-			c.cancel();
-		}
-	);
-
-	ioc.run_for(std::chrono::seconds(10));
-	BOOST_CHECK_EQUAL(handlers_called, expected_handlers_called);
-	BOOST_CHECK(broker.received_all_expected());
-}
-
-BOOST_AUTO_TEST_CASE(test_receive_publish_qos0) {
-	receive_publish<qos_e::at_most_once>();
-}
-
-BOOST_AUTO_TEST_CASE(test_receive_publish_qos1) {
-	receive_publish<qos_e::at_least_once>();
-}
-
-BOOST_AUTO_TEST_CASE(test_receive_publish_qos2) {
-	receive_publish<qos_e::exactly_once>();
-}
-
-BOOST_AUTO_TEST_CASE(test_waiting_on_pubrel) {
-	using test::after;
-	using std::chrono_literals::operator ""ms;
-
-	constexpr int expected_handlers_called = 1;
-	int handlers_called = 0;
-
-	std::string topic = "topic";
-	std::string payload = "payload";
-
-	// packets
-	auto connect = encoders::encode_connect(
-		"", std::nullopt, std::nullopt, 10, false, {}, std::nullopt
-	);
-	auto connack = encoders::encode_connack(
-		false, reason_codes::success.value(), {}
-	);
-
-	auto publish = encoders::encode_publish(
-		1, topic, payload, qos_e::exactly_once, retain_e::yes, dup_e::no, {}
-	);
-
-	auto pubrec = encoders::encode_pubrec(1, reason_codes::success.value(), {});
-	auto pubrel = encoders::encode_pubrel(1, reason_codes::success.value(), {});
-	auto pubcomp = encoders::encode_pubcomp(1, reason_codes::success.value(), {});
-
-	test::msg_exchange broker_side;
+struct shared_test_data {
 	error_code success {};
 	error_code fail = asio::error::not_connected;
 
-	broker_side
-		.expect(connect)
-			.complete_with(success, after(0ms))
-			.reply_with(connack, after(0ms))
-		.send(publish, after(10ms))
-		.expect(pubrec)
-			.complete_with(success, after(1ms))
-			.reply_with(pubrel, after(2ms))
-		.expect(pubcomp)
-			.complete_with(fail, after(1ms))
-		.expect(connect)
-			.complete_with(success, after(0ms))
-			.reply_with(connack, after(0ms))
-		.send(pubrel, after(10ms))
-		.expect(pubcomp)
-			.complete_with(success, after(1ms));
+	const std::string connect = encoders::encode_connect(
+		"", std::nullopt, std::nullopt, 10, false, {}, std::nullopt
+	);
+	const std::string connack = encoders::encode_connack(
+		true, reason_codes::success.value(), {}
+	);
+
+	const std::string topic = "topic";
+	const std::string payload = "payload";
+
+	const std::string publish_qos0 = encoders::encode_publish(
+		0, topic, payload, qos_e::at_most_once, retain_e::no, dup_e::no, {}
+	);
+	const std::string publish_qos1 = encoders::encode_publish(
+		1, topic, payload, qos_e::at_least_once, retain_e::no, dup_e::no, {}
+	);
+	const std::string publish_qos2 = encoders::encode_publish(
+		1, topic, payload, qos_e::exactly_once, retain_e::no, dup_e::no, {}
+	);
+
+	const std::string puback = encoders::encode_puback(1, uint8_t(0x00), {});
+
+	const std::string pubrec = encoders::encode_pubrec(1, uint8_t(0x00), {});
+	const std::string pubrel = encoders::encode_pubrel(1, uint8_t(0x00), {});
+	const std::string pubcomp = encoders::encode_pubcomp(1, uint8_t(0x00), {});
+};
+
+using test::after;
+using std::chrono_literals::operator ""ms;
+
+void run_test(test::msg_exchange broker_side) {
+	constexpr int expected_handlers_called = 1;
+	int handlers_called = 0;
 
 	asio::io_context ioc;
 	auto executor = ioc.get_executor();
@@ -176,8 +72,8 @@ BOOST_AUTO_TEST_CASE(test_waiting_on_pubrel) {
 			++handlers_called;
 
 			BOOST_CHECK_MESSAGE(!ec, ec.message());
-			BOOST_CHECK_EQUAL(topic, rec_topic);
-			BOOST_CHECK_EQUAL(payload,  rec_payload);
+			BOOST_CHECK_EQUAL(shared_test_data().topic, rec_topic);
+			BOOST_CHECK_EQUAL(shared_test_data().payload,  rec_payload);
 
 			c.cancel();
 		}
@@ -188,4 +84,97 @@ BOOST_AUTO_TEST_CASE(test_waiting_on_pubrel) {
 	BOOST_CHECK(broker.received_all_expected());
 }
 
+
+BOOST_FIXTURE_TEST_CASE(test_receive_publish_qos0, shared_test_data) {
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(0ms))
+			.reply_with(connack, after(0ms))
+		.send(publish_qos0, after(10ms));
+
+	run_test(std::move(broker_side));
+}
+
+BOOST_FIXTURE_TEST_CASE(test_receive_publish_qos1, shared_test_data) {
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(0ms))
+			.reply_with(connack, after(0ms))
+		.send(publish_qos1, after(10ms))
+		.expect(puback)
+			.complete_with(success, after(1ms));
+
+	run_test(std::move(broker_side));
+}
+
+BOOST_FIXTURE_TEST_CASE(test_receive_publish_qos2, shared_test_data) {
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(0ms))
+			.reply_with(connack, after(0ms))
+		.send(publish_qos2, after(10ms))
+		.expect(pubrec)
+			.complete_with(success, after(1ms))
+		.reply_with(pubrel, after(2ms))
+			.expect(pubcomp)
+		.complete_with(success, after(1ms));
+
+	run_test(std::move(broker_side));
+}
+
+
+BOOST_FIXTURE_TEST_CASE(fail_to_send_pubrec, shared_test_data) {
+	auto publish_dup = encoders::encode_publish(
+		1, topic, payload, qos_e::exactly_once, retain_e::no, dup_e::yes, {}
+	);
+
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(0ms))
+			.reply_with(connack, after(0ms))
+		.send(publish_qos2, after(3ms))
+		// write completed, but the broker did not actually
+		// receive pubrec, it will resend publish again
+		.expect(pubrec)
+			.complete_with(success, after(1ms))
+			.reply_with(fail, after(3ms))
+		.expect(connect)
+			.complete_with(success, after(0ms))
+			.reply_with(connack, after(0ms))
+		.send(publish_dup, after(100ms))
+		.expect(pubrec)
+			.complete_with(success, after(1ms))
+			.reply_with(pubrel, after(2ms))
+		.expect(pubcomp)
+			.complete_with(success, after(1ms));
+
+	run_test(std::move(broker_side));
+}
+
+BOOST_FIXTURE_TEST_CASE(fail_to_send_pubcomp, shared_test_data) {
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(0ms))
+			.reply_with(connack, after(0ms))
+		.send(publish_qos2, after(10ms))
+		.expect(pubrec)
+			.complete_with(success, after(1ms))
+			.reply_with(pubrel, after(2ms))
+		.expect(pubcomp)
+			.complete_with(fail, after(1ms))
+		.expect(connect)
+			.complete_with(success, after(0ms))
+			.reply_with(connack, after(0ms))
+		.send(pubrel, after(10ms))
+		.expect(pubcomp)
+			.complete_with(success, after(1ms));
+
+	run_test(std::move(broker_side));
+}
+
 BOOST_AUTO_TEST_SUITE_END();

test/unit/connect_op.cpp

@@ -0,0 +1,225 @@
+#include <boost/test/unit_test.hpp>
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/steady_timer.hpp>
+
+#include <boost/asio/ip/tcp.hpp>
+
+#include <async_mqtt5/detail/internal_types.hpp>
+
+#include <async_mqtt5/impl/connect_op.hpp>
+
+#include <async_mqtt5/types.hpp>
+
+#include "test_common/test_stream.hpp"
+
+using namespace async_mqtt5;
+
+BOOST_AUTO_TEST_SUITE(connect_op/*, *boost::unit_test::disabled()*/)
+
+struct shared_test_data {
+	error_code success{};
+	error_code fail = asio::error::not_connected;
+
+	const std::string connect = encoders::encode_connect(
+		"", std::nullopt, std::nullopt, 10, false, {}, std::nullopt
+	);
+	const std::string connack = encoders::encode_connack(
+		true, reason_codes::success.value(), {}
+	);
+};
+
+using test::after;
+using std::chrono_literals::operator ""ms;
+
+void run_unit_test(
+	test::msg_exchange broker_side,
+	asio::any_completion_handler<void(error_code)> op_handler
+) {
+	constexpr int expected_handlers_called = 1;
+	int handlers_called = 0;
+
+	asio::io_context ioc;
+	auto executor = ioc.get_executor();
+	auto& broker = asio::make_service<test::test_broker>(
+		ioc, executor, std::move(broker_side)
+	);
+	test::test_stream stream(executor);
+
+	detail::mqtt_ctx mqtt_ctx;
+	authority_path ap;
+	auto eps = asio::ip::tcp::resolver(executor).resolve("127.0.0.1", "");
+
+	auto handler = [&handlers_called, h = std::move(op_handler)](error_code ec) mutable {
+		handlers_called++;
+		std::move(h)(ec);
+	};
+
+	detail::connect_op<test::test_stream>(
+		stream, std::move(handler), mqtt_ctx
+	).perform(eps, ap);
+
+	ioc.run_for(std::chrono::seconds(1));
+	BOOST_CHECK_EQUAL(handlers_called, expected_handlers_called);
+	BOOST_CHECK(broker.received_all_expected());
+}
+
+BOOST_FIXTURE_TEST_CASE(successfully_connect, shared_test_data) {
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(2ms))
+			.reply_with(connack, after(4ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == success);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+BOOST_FIXTURE_TEST_CASE(connection_refused, shared_test_data) {
+	auto refuse_connack = encoders::encode_connack(
+		true, reason_codes::server_unavailable.value(), {}
+	);
+
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(2ms))
+			.reply_with(refuse_connack, after(4ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == asio::error::connection_refused);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+
+BOOST_FIXTURE_TEST_CASE(access_denied, shared_test_data) {
+	auto denied_connack = encoders::encode_connack(
+		true, reason_codes::bad_username_or_password.value(), {}
+	);
+
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(2ms))
+			.reply_with(denied_connack, after(4ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == asio::error::access_denied);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+BOOST_FIXTURE_TEST_CASE(fail_to_send_connect, shared_test_data) {
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+		.complete_with(fail, after(2ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == fail);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+
+BOOST_FIXTURE_TEST_CASE(receive_wrong_packet, shared_test_data) {
+	// packets
+	auto unexpected_packet = encoders::encode_puback(1, uint8_t(0x00), {});
+
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(2ms))
+			.reply_with(unexpected_packet, after(3ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == asio::error::try_again);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+BOOST_FIXTURE_TEST_CASE(malformed_connack_varlen, shared_test_data) {
+	// packets
+	auto malformed_connack = std::string({ 0x20, -1 /* 0xFF */, -1, -1, -1 });
+
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(2ms))
+			.reply_with(malformed_connack, after(3ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == asio::error::try_again);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+BOOST_FIXTURE_TEST_CASE(malformed_connack_rc, shared_test_data) {
+	// packets
+	auto malformed_connack = encoders::encode_connack(true, uint8_t(0x04), {});
+
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+		.complete_with(success, after(2ms))
+		.reply_with(malformed_connack, after(3ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == client::error::malformed_packet);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+BOOST_FIXTURE_TEST_CASE(fail_reading_connack_payload, shared_test_data) {
+	// packets
+	connack_props cprops;
+	cprops[prop::reason_string] = std::string(256, 'a');
+
+	auto big_connack = encoders::encode_connack(
+		true, uint8_t(0x00), cprops
+	);
+
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(2ms))
+		.send(big_connack.substr(0, 5), after(5ms))
+		.send(fail, after(7ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == fail);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+BOOST_FIXTURE_TEST_CASE(receive_unexpected_auth, shared_test_data) {
+	auth_props aprops;
+	aprops[prop::authentication_method] = "method";
+	aprops[prop::authentication_data] = "data";
+	auto auth = encoders::encode_auth(uint8_t(0x19), aprops);
+
+	test::msg_exchange broker_side;
+	broker_side
+		.expect(connect)
+			.complete_with(success, after(2ms))
+		.send(auth, after(5ms));
+
+	auto handler = [&](error_code ec) {
+		BOOST_CHECK(ec == client::error::malformed_packet);
+	};
+
+	run_unit_test(std::move(broker_side), std::move(handler));
+}
+
+BOOST_AUTO_TEST_SUITE_END();

test/unit/disconnect_op.cpp

@@ -1,7 +1,5 @@
 #include <boost/test/unit_test.hpp>
 
-#include <boost/asio/bind_cancellation_slot.hpp>
-#include <boost/asio/cancellation_signal.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/steady_timer.hpp>