mirror/mongoose
1
0
Fork 0
mirror of https://github.com/cesanta/mongoose.git synced 2024-11-28 05:39:00 +08:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

Merge pull request #2246 from cesanta/tst

Browse Source 
Improve MQTT tests resiliency
This commit is contained in:
Sergio R. Caprile 2023-06-07 19:01:23 -03:00 committed by GitHub
commit dcfa7ad657
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 70 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
test/unit_test.c
View File

@ -341,8 +341,10 @@ static void test_sntp(void) {
} }
struct mqtt_data { struct mqtt_data {
char *buf; char *topic;
size_t bufsize; char *msg;
size_t topicsize;
size_t msgsize;
int flags; int flags;
}; };
#define flags_subscribed (1 << 0) #define flags_subscribed (1 << 0)
@ -353,10 +355,12 @@ struct mqtt_data {
static void mqtt_cb(struct mg_connection *c, int ev, void *evd, void *fnd) { static void mqtt_cb(struct mg_connection *c, int ev, void *evd, void *fnd) {
struct mqtt_data *test_data = (struct mqtt_data *) fnd; struct mqtt_data *test_data = (struct mqtt_data *) fnd;
char *buf = test_data->buf; char *buf = test_data->msg;
if (ev == MG_EV_MQTT_OPEN) { if (ev == MG_EV_MQTT_OPEN) {
buf[0] = *(int *) evd == 0 ? 'X' : 'Y'; buf[0] = *(int *) evd == 0 ? 'X' : 'Y';
} else if (ev == MG_EV_CLOSE) {
buf[0] = 0;
} else if (ev == MG_EV_MQTT_CMD) { } else if (ev == MG_EV_MQTT_CMD) {
struct mg_mqtt_message *mm = (struct mg_mqtt_message *) evd; struct mg_mqtt_message *mm = (struct mg_mqtt_message *) evd;
if (mm->cmd == MQTT_CMD_SUBACK) { if (mm->cmd == MQTT_CMD_SUBACK) {
@ -372,8 +376,10 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *evd, void *fnd) {
} }
} else if (ev == MG_EV_MQTT_MSG) { } else if (ev == MG_EV_MQTT_MSG) {
struct mg_mqtt_message *mm = (struct mg_mqtt_message *) evd; struct mg_mqtt_message *mm = (struct mg_mqtt_message *) evd;
snprintf(buf + 1, test_data->bufsize, "%.*s/%.*s", (int) mm->topic.len, snprintf(test_data->topic, test_data->topicsize, "%.*s",
mm->topic.ptr, (int) mm->data.len, mm->data.ptr); (int) mm->topic.len, mm->topic.ptr);
snprintf(buf + 1, test_data->msgsize - 2, "%.*s", (int) mm->data.len,
mm->data.ptr);
if (mm->cmd == MQTT_CMD_PUBLISH && c->is_mqtt5) { if (mm->cmd == MQTT_CMD_PUBLISH && c->is_mqtt5) {
size_t pos = 0; size_t pos = 0;
@ -426,8 +432,8 @@ static void construct_props(struct mg_mqtt_prop *props) {
static void test_mqtt_base(void); static void test_mqtt_base(void);
static void test_mqtt_base(void) { static void test_mqtt_base(void) {
char buf[50] = {0}; char buf[10] = {0}; // we won't use it
struct mqtt_data test_data = {buf, 50, 0}; struct mqtt_data test_data = {buf, buf, 10, 10, 0};
struct mg_mgr mgr; struct mg_mgr mgr;
struct mg_connection *c; struct mg_connection *c;
const char *url = "mqtt://broker.hivemq.com:1883"; const char *url = "mqtt://broker.hivemq.com:1883";
@ -445,11 +451,24 @@ static void test_mqtt_base(void) {
ASSERT(mgr.conns == NULL); ASSERT(mgr.conns == NULL);
} }
static void check_mqtt_message(struct mg_mqtt_opts *opts,
struct mqtt_data *data, bool enforce) {
if (opts->topic.len != strlen(data->topic) ||
strcmp(opts->topic.ptr, data->topic)) {
MG_INFO(("TOPIC[%s]", data->topic));
if (enforce) ASSERT(0);
}
if (*data->msg != 'X' || opts->message.len != (strlen(&data->msg[1])) ||
strcmp(opts->message.ptr, &data->msg[1])) {
MG_INFO(("MSG[%s]", data->msg));
if (enforce) ASSERT(0);
}
}
static void test_mqtt_ver(uint8_t mqtt_version) { static void test_mqtt_ver(uint8_t mqtt_version) {
char buf[50] = {0}, client_id[16], will_topic[16]; char tbuf[16], mbuf[50] = {0}, client_id[16], topic[16];
struct mqtt_data test_data = {buf, 50, 0}; struct mqtt_data test_data = {tbuf, mbuf, 16, 50, 0};
struct mg_mgr mgr; struct mg_mgr mgr;
struct mg_str topic = mg_str("x/f12"), data = mg_str("hi");
struct mg_connection *c; struct mg_connection *c;
struct mg_mqtt_opts opts; struct mg_mqtt_opts opts;
struct mg_mqtt_prop properties[4]; struct mg_mqtt_prop properties[4];
@ -460,99 +479,105 @@ static void test_mqtt_ver(uint8_t mqtt_version) {
// Connect with empty client ID, no options, ergo no MQTT != 3.1.1 // Connect with empty client ID, no options, ergo no MQTT != 3.1.1
if (mqtt_version != 4) goto connect_with_options; if (mqtt_version != 4) goto connect_with_options;
c = mg_mqtt_connect(&mgr, url, NULL, mqtt_cb, &test_data); c = mg_mqtt_connect(&mgr, url, NULL, mqtt_cb, &test_data);
for (i = 0; i < 300 && buf[0] == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 300 && mbuf[0] == 0; i++) mg_mgr_poll(&mgr, 10);
if (buf[0] != 'X') MG_INFO(("[%s]", buf)); if (mbuf[0] != 'X') MG_INFO(("[%s]", mbuf));
ASSERT(buf[0] == 'X'); ASSERT(mbuf[0] == 'X');
ASSERT(test_data.flags == 0); ASSERT(test_data.flags == 0);
// Subscribe with QoS1 // Subscribe with QoS1
opts.topic = topic, opts.qos = 1; opts.topic = mg_str(mg_random_str(topic, sizeof(topic)));
opts.qos = 1;
mg_mqtt_sub(c, &opts); mg_mqtt_sub(c, &opts);
for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10);
ASSERT(test_data.flags == flags_subscribed); ASSERT(test_data.flags == flags_subscribed);
test_data.flags = 0; test_data.flags = 0;
// Publish with QoS0 to subscribed topic and check reception // Publish with QoS0 to subscribed topic and check reception
opts.topic = topic, opts.message = data, opts.qos = 0, opts.retain = false; // keep former opts.topic
opts.message = mg_str("hi0"), opts.qos = 0, opts.retain = false;
mg_mqtt_pub(c, &opts); mg_mqtt_pub(c, &opts);
for (i = 0; i < 500 && buf[1] == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 500 && mbuf[1] == 0; i++) mg_mgr_poll(&mgr, 10);
if (strcmp(buf, "Xx/f12/hi") != 0) MG_INFO(("[%s]", buf)); ASSERT(!(test_data.flags & flags_published)); // No PUBACK for QoS0
ASSERT(!(test_data.flags & flags_published)); // No PUBACK for QoS0 check_mqtt_message(&opts, &test_data, false); // We may not get the msg
ASSERT(strcmp(buf, "Xx/f12/hi") == 0); memset(mbuf + 1, 0, sizeof(mbuf) - 1);
memset(buf + 1, 0, sizeof(buf) - 1);
test_data.flags = 0; test_data.flags = 0;
// Publish with QoS1 to subscribed topic and check reception // Publish with QoS1 to subscribed topic and check reception
opts.topic = topic, opts.message = data, opts.qos = 1, opts.retain = false; // keep former opts.topic
opts.message = mg_str("hi1"), opts.qos = 1, opts.retain = false;
retries = 0; // don't do retries for test speed retries = 0; // don't do retries for test speed
do { // retry on failure after an expected timeout do { // retry on failure after an expected timeout
mg_mqtt_pub(c, &opts); mg_mqtt_pub(c, &opts);
for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10);
} while (test_data.flags == 0 && retries--); } while (test_data.flags == 0 && retries--);
ASSERT(test_data.flags == flags_published); ASSERT(test_data.flags == flags_published);
for (i = 0; i < 500 && buf[1] == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 500 && mbuf[1] == 0; i++) mg_mgr_poll(&mgr, 10);
if (strcmp(buf, "Xx/f12/hi") != 0) MG_INFO(("[%s]", buf)); check_mqtt_message(&opts, &test_data, true);
ASSERT(strcmp(buf, "Xx/f12/hi") == 0); memset(mbuf + 1, 0, sizeof(mbuf) - 1);
memset(buf + 1, 0, sizeof(buf) - 1);
test_data.flags = 0; test_data.flags = 0;
// Disconnect ! // Disconnect !
mg_mqtt_disconnect(c, NULL);
for (i = 0; i < 10 && mbuf[0] != 0; i++) mg_mgr_poll(&mgr, 10);
mg_mgr_free(&mgr); mg_mgr_free(&mgr);
ASSERT(mgr.conns == NULL); ASSERT(mgr.conns == NULL);
connect_with_options: connect_with_options:
// (Re-)connect with options: version, clean session, last will, keepalive // (Re-)connect with options: version, clean session, last will, keepalive
// time // time. Don't set retain, some runners are not random
test_data.flags = 0; test_data.flags = 0;
memset(buf, 0, sizeof(buf)); memset(mbuf, 0, sizeof(mbuf));
memset(&opts, 0, sizeof(opts)); memset(&opts, 0, sizeof(opts));
mg_mgr_init(&mgr); mg_mgr_init(&mgr);
opts.clean = true, opts.qos = 1, opts.retain = true, opts.keepalive = 20; opts.clean = true, opts.qos = 1, opts.retain = false, opts.keepalive = 20;
opts.version = mqtt_version; opts.version = mqtt_version;
opts.topic = mg_str(mg_random_str(will_topic, sizeof(will_topic))); opts.topic = mg_str(mg_random_str(topic, sizeof(topic)));
opts.message = mg_str("mg_will_messsage"); opts.message = mg_str("mg_will_messsage");
opts.client_id = mg_str(mg_random_str(client_id, sizeof(client_id))); opts.client_id = mg_str(mg_random_str(client_id, sizeof(client_id)));
c = mg_mqtt_connect(&mgr, url, &opts, mqtt_cb, &test_data); c = mg_mqtt_connect(&mgr, url, &opts, mqtt_cb, &test_data);
for (i = 0; i < 300 && buf[0] == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 300 && mbuf[0] == 0; i++) mg_mgr_poll(&mgr, 10);
if (buf[0] != 'X') MG_INFO(("[%s]", buf)); if (mbuf[0] != 'X') MG_INFO(("[%s]", mbuf));
ASSERT(buf[0] == 'X'); ASSERT(mbuf[0] == 'X');
ASSERT(test_data.flags == 0); ASSERT(test_data.flags == 0);
// Subscribe with QoS2 (reception downgrades to published QoS) // Subscribe with QoS2 (reception downgrades to published QoS)
opts.topic = topic, opts.qos = 2; opts.topic = mg_str(mg_random_str(topic, sizeof(topic)));
opts.qos = 2;
mg_mqtt_sub(c, &opts); mg_mqtt_sub(c, &opts);
for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10);
ASSERT(test_data.flags == flags_subscribed); ASSERT(test_data.flags == flags_subscribed);
test_data.flags = 0; test_data.flags = 0;
// Publish with QoS1 to subscribed topic and check reception // Publish with QoS1 to subscribed topic and check reception
opts.topic = topic, opts.message = data, opts.qos = 1, opts.retain = false; // keep former opts.topic
opts.message = mg_str("hi1"), opts.qos = 1, opts.retain = false;
if (mqtt_version == 5) { if (mqtt_version == 5) {
opts.props = properties; opts.props = properties;
opts.num_props = 4; opts.num_props = 4;
construct_props(properties); construct_props(properties);
} }
retries = 0; // don't do retries for test speed retries = 0; // don't do retries for test speed
do { // retry on failure after an expected timeout do { // retry on failure after an expected timeout
mg_mqtt_pub(c, &opts); mg_mqtt_pub(c, &opts);
for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 500 && test_data.flags == 0; i++) mg_mgr_poll(&mgr, 10);
} while (test_data.flags == 0 && retries--); } while (test_data.flags == 0 && retries--);
ASSERT(test_data.flags == flags_published); ASSERT(test_data.flags == flags_published);
for (i = 0; i < 500 && buf[1] == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 500 && mbuf[1] == 0; i++) mg_mgr_poll(&mgr, 10);
if (strcmp(buf, "Xx/f12/hi") != 0) MG_INFO(("[%s]", buf)); check_mqtt_message(&opts, &test_data, true);
ASSERT(strcmp(buf, "Xx/f12/hi") == 0); memset(mbuf + 1, 0, sizeof(mbuf) - 1);
test_data.flags = 0; test_data.flags = 0;
// Publish with QoS2 to subscribed topic and check (simultaneous) reception // Publish with QoS2 to subscribed topic and check (simultaneous) reception
opts.topic = topic, opts.message = data, opts.qos = 2, opts.retain = false; // keep former opts.topic
opts.message = mg_str("hi2"), opts.qos = 2, opts.retain = false;
if (mqtt_version == 5) { if (mqtt_version == 5) {
opts.props = properties; opts.props = properties;
opts.num_props = 4; opts.num_props = 4;
construct_props(properties); construct_props(properties);
} }
retries = 0; // don't do retries for test speed retries = 0; // don't do retries for test speed
do { // retry on failure after an expected timeout do { // retry on failure after an expected timeout
mg_mqtt_pub(c, &opts); mg_mqtt_pub(c, &opts);
for (i = 0; i < 500 && !(test_data.flags & flags_received); i++) for (i = 0; i < 500 && !(test_data.flags & flags_received); i++)
mg_mgr_poll(&mgr, 10); mg_mgr_poll(&mgr, 10);
@ -565,15 +590,15 @@ connect_with_options:
// TODO(): retry sending PUBREL on failure after an expected timeout // TODO(): retry sending PUBREL on failure after an expected timeout
// or broker sends PUBREC again // or broker sends PUBREC again
ASSERT(test_data.flags & flags_completed); ASSERT(test_data.flags & flags_completed);
for (i = 0; i < 500 && buf[1] == 0; i++) mg_mgr_poll(&mgr, 10); for (i = 0; i < 500 && mbuf[1] == 0; i++) mg_mgr_poll(&mgr, 10);
if (strcmp(buf, "Xx/f12/hi") != 0) MG_INFO(("[%s]", buf)); check_mqtt_message(&opts, &test_data, true);
ASSERT(strcmp(buf, "Xx/f12/hi") == 0);
for (i = 0; i < 500 && !(test_data.flags & flags_released); i++) for (i = 0; i < 500 && !(test_data.flags & flags_released); i++)
mg_mgr_poll(&mgr, 10); mg_mgr_poll(&mgr, 10);
ASSERT(test_data.flags & flags_released); // Mongoose sent PUBCOMP ASSERT(test_data.flags & flags_released); // Mongoose sent PUBCOMP
memset(buf + 1, 0, sizeof(buf) - 1); memset(mbuf + 1, 0, sizeof(mbuf) - 1);
test_data.flags = 0; test_data.flags = 0;
// dirty disconnect
mg_mgr_free(&mgr); mg_mgr_free(&mgr);
ASSERT(mgr.conns == NULL); ASSERT(mgr.conns == NULL);
} }
@ -582,7 +607,6 @@ static void test_mqtt(void) {
test_mqtt_base(); test_mqtt_base();
test_mqtt_ver(4); test_mqtt_ver(4);
test_mqtt_ver(5); test_mqtt_ver(5);
test_mqtt_base();
} }
static void eh1(struct mg_connection *c, int ev, void *ev_data, void *fn_data) { static void eh1(struct mg_connection *c, int ev, void *ev_data, void *fn_data) {