esp-mqtt/test/mqtt_msg_host_test/main/test_mqtt_msg.cpp

/*
 * SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
 *
 * SPDX-License-Identifier: Unlicense OR CC0-1.0
 */
#include <catch2/catch_test_macros.hpp>
#include <rapidcheck.h>
#include <rapidcheck/catch.h>
#include <cstdint>
#include <cstring>
#include <memory>
#include <random>
#include <string>
#include <vector>

extern "C" {
#include "mqtt5_msg.h"
#include "mqtt_msg.h"

    int platform_random(int max)
    {
        if (max <= 0) {
            return 0;
        }

        static std::mt19937 engine{std::random_device{}()};
        std::uniform_int_distribution<int> distribution(0, max - 1);
        return distribution(engine);
    }
}

namespace
{

constexpr size_t mqtt_fixed_header_size = 5;

struct MessageDeleter {
    void operator()(mqtt_message_t *message) const
    {
        mqtt_msg_destroy(message);
    }
};

using MessagePtr = std::unique_ptr<mqtt_message_t, MessageDeleter>;

MessagePtr make_message(size_t buffer_size)
{
    return MessagePtr(mqtt_msg_create(buffer_size));
}

std::string generated_bytes(int length, int minimum, int maximum)
{
    std::string value(static_cast<size_t>(length), '\0');

    for (char &byte : value) {
        byte = static_cast<char>(*rc::gen::inRange(minimum, maximum));
    }

    return value;
}

} // namespace

TEST_CASE("mqtt_msg_pingreq encodes the fixed MQTT control packet")
{
    MessagePtr message = make_message(mqtt_fixed_header_size);
    REQUIRE(message != nullptr);
    mqtt_message_t *encoded = mqtt_msg_pingreq(message.get());
    REQUIRE(encoded == message.get());
    REQUIRE(encoded->length == 2);
    REQUIRE(mqtt_get_type(encoded->data) == MQTT_MSG_TYPE_PINGREQ);
    REQUIRE(encoded->data[0] == 0xC0);
    REQUIRE(encoded->data[1] == 0x00);
    REQUIRE(mqtt_has_valid_msg_hdr(encoded->data, encoded->length) == 1);
}

TEST_CASE("mqtt_msg_publish rejects an empty topic")
{
    MessagePtr message = make_message(32);
    REQUIRE(message != nullptr);
    uint16_t message_id = 7;
    mqtt_message_t *encoded = mqtt_msg_publish(message.get(), "", "payload", 7, 0, 0, &message_id);
    REQUIRE(encoded == message.get());
    REQUIRE(encoded->length == 0);
}

TEST_CASE("mqtt5 CONNACK parser rejects truncated fixed-width properties")
{
    struct PropertyCase {
        uint8_t id;
        size_t width;
    };
    const PropertyCase property_cases[] = {
        {MQTT5_PROPERTY_RECEIVE_MAXIMUM, 2},
        {MQTT5_PROPERTY_TOPIC_ALIAS_MAXIMIM, 2},
        {MQTT5_PROPERTY_MAXIMUM_PACKET_SIZE, 4},
    };

    for (const PropertyCase &property_case : property_cases) {
        for (size_t available = 0; available < property_case.width; ++available) {
            CAPTURE(property_case.id, available);
            std::vector<uint8_t> packet(6 + available, 0);
            packet[0] = 0x20;
            packet[1] = static_cast<uint8_t>(4 + available);
            packet[4] = static_cast<uint8_t>(1 + available);
            packet[5] = property_case.id;
            mqtt_connect_info_t connection_info = {};
            esp_mqtt5_connection_property_storage_t connection_property = {};
            esp_mqtt5_connection_server_resp_property_t response_property = {};
            int reason_code = 0;
            uint8_t ack_flag = 0;
            mqtt5_user_property_handle_t user_property = nullptr;
            REQUIRE(mqtt5_msg_parse_connack_property(packet.data(), packet.size(), &connection_info,
                                                     &connection_property, &response_property, &reason_code,
                                                     &ack_flag, &user_property) == ESP_FAIL);
        }
    }
}

TEST_CASE("mqtt_msg_publish round-trips an empty payload")
{
    const int qos_values[] = {0, 1};

    for (const int qos : qos_values) {
        const size_t buffer_size = mqtt_fixed_header_size + 2 + 1 + (qos > 0 ? 2 : 0);
        MessagePtr message = make_message(buffer_size);
        REQUIRE(message != nullptr);
        uint16_t message_id = 0;
        mqtt_message_t *encoded = mqtt_msg_publish(message.get(), "t", nullptr, 0, qos, 0, &message_id);
        REQUIRE(encoded == message.get());
        REQUIRE(encoded->length > 0);
        size_t payload_length = encoded->length;
        const char *payload = mqtt_get_publish_data(encoded->data, &payload_length);
        REQUIRE(payload != nullptr);
        REQUIRE(payload_length == 0);

        if (qos == 0) {
            REQUIRE(message_id == 0);
        } else {
            REQUIRE(message_id != 0);
            REQUIRE(mqtt_get_id(encoded->data, encoded->length) == message_id);
        }
    }
}

TEST_CASE("mqtt_msg_publish round-trips a complete packet")
{
    rc::prop("encoded publish recovers topic, payload, flags, and packet id", []() {
        const int qos = *rc::gen::inRange(0, 3);
        const int retain = *rc::gen::inRange(0, 2);
        const std::string topic = generated_bytes(*rc::gen::inRange(1, 65), 1, 127);
        const std::string payload = generated_bytes(*rc::gen::inRange(0, 129), 0, 256);
        const size_t buffer_size = mqtt_fixed_header_size + 2 + topic.size() +
                                   (qos > 0 ? 2 : 0) + payload.size();
        MessagePtr message = make_message(buffer_size);
        RC_ASSERT(message != nullptr);
        uint16_t message_id = 0;
        mqtt_message_t *encoded = mqtt_msg_publish(message.get(), topic.c_str(), payload.data(),
                                                   static_cast<int>(payload.size()), qos, retain, &message_id);
        RC_ASSERT(encoded == message.get());
        RC_ASSERT(encoded->length > 0);
        RC_ASSERT(encoded->length <= buffer_size);
        int fixed_header_size = 0;
        const size_t total_length = mqtt_get_total_length(encoded->data, encoded->length, &fixed_header_size);
        RC_ASSERT(total_length == encoded->length);
        RC_ASSERT(fixed_header_size >= 2);
        RC_ASSERT(mqtt_get_type(encoded->data) == MQTT_MSG_TYPE_PUBLISH);
        RC_ASSERT(mqtt_get_qos(encoded->data) == qos);
        RC_ASSERT(mqtt_get_retain(encoded->data) == retain);
        RC_ASSERT(mqtt_get_dup(encoded->data) == 0);
        RC_ASSERT(mqtt_has_valid_msg_hdr(encoded->data, encoded->length) == 1);
        size_t topic_length = encoded->length;
        const char *encoded_topic = mqtt_get_publish_topic(encoded->data, &topic_length);
        RC_ASSERT(encoded_topic != nullptr);
        RC_ASSERT(topic_length == topic.size());
        RC_ASSERT(std::memcmp(encoded_topic, topic.data(), topic.size()) == 0);
        size_t payload_length = encoded->length;
        const char *encoded_payload = mqtt_get_publish_data(encoded->data, &payload_length);
        RC_ASSERT(encoded_payload != nullptr);
        RC_ASSERT(payload_length == payload.size());
        RC_ASSERT(std::memcmp(encoded_payload, payload.data(), payload.size()) == 0);

        if (qos == 0) {
            RC_ASSERT(message_id == 0);
        } else {
            RC_ASSERT(message_id != 0);
            RC_ASSERT(mqtt_get_id(encoded->data, encoded->length) == message_id);
        }
    });
}

TEST_CASE("mqtt5_msg encodes joined topics without truncation")
{
    auto contains = [](const mqtt_message_t *message, const std::string & expected) {
        const std::string encoded(reinterpret_cast<const char *>(message->data), message->length);
        return encoded.find(expected) != std::string::npos;
    };
    SECTION("response topic with response info") {
        esp_mqtt5_publish_property_config_t property = {};
        property.response_topic = "response";
        MessagePtr message = make_message(64);
        REQUIRE(message != nullptr);
        uint16_t message_id = 1;
        mqtt5_msg_publish(message.get(), "t", "x", 1, 1, 0, &message_id, &property, "info");
        REQUIRE(message->length > 0);
        REQUIRE(contains(message.get(), std::string("\x08\x00\x0d", 3) + "response/info"));
    }
    SECTION("shared subscribe and unsubscribe topics") {
        const std::string expected = std::string("\x00\x10", 2) + "$share/group/a/b";
        const esp_mqtt_topic_t topic = {.filter = "a/b", .qos = 1};
        esp_mqtt5_subscribe_property_config_t subscribe_property = {};
        subscribe_property.is_share_subscribe = true;
        subscribe_property.share_name = "group";
        MessagePtr subscribe = make_message(64);
        REQUIRE(subscribe != nullptr);
        uint16_t message_id = 1;
        mqtt5_msg_subscribe(subscribe.get(), &topic, 1, &message_id, &subscribe_property);
        REQUIRE(subscribe->length > 0);
        REQUIRE(contains(subscribe.get(), expected));
        esp_mqtt5_unsubscribe_property_config_t unsubscribe_property = {};
        unsubscribe_property.is_share_subscribe = true;
        unsubscribe_property.share_name = "group";
        MessagePtr unsubscribe = make_message(64);
        REQUIRE(unsubscribe != nullptr);
        mqtt5_msg_unsubscribe(unsubscribe.get(), "a/b", &message_id, &unsubscribe_property);
        REQUIRE(unsubscribe->length > 0);
        REQUIRE(contains(unsubscribe.get(), expected));
        unsubscribe_property.share_name = nullptr;
        mqtt5_msg_unsubscribe(unsubscribe.get(), "a/b", &message_id, &unsubscribe_property);
        REQUIRE(unsubscribe->length == 0);
    }
}