From ea8369ae98722ed3114e62bd2bb7937831749ced Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 6 May 2024 00:19:00 +0800 Subject: [PATCH 1/4] feat(core): mqtt client subscription support wildcard --- include/neuron/connection/mqtt_client.h | 14 +++ src/connection/mqtt_client.c | 117 ++++++++++++++++++++- tests/ut/CMakeLists.txt | 8 ++ tests/ut/mqtt_client_test.cc | 129 ++++++++++++++++++++++++ 4 files changed, 265 insertions(+), 3 deletions(-) create mode 100644 tests/ut/mqtt_client_test.cc diff --git a/include/neuron/connection/mqtt_client.h b/include/neuron/connection/mqtt_client.h index faec6e75..91641db5 100644 --- a/include/neuron/connection/mqtt_client.h +++ b/include/neuron/connection/mqtt_client.h @@ -53,6 +53,20 @@ typedef enum { typedef struct neu_mqtt_client_s neu_mqtt_client_t; +/** + * Check that `topic_filter` is a valid MQTT topic filter. + */ +bool neu_mqtt_topic_filter_is_valid(const char *topic_filter); + +/** + * Check the given `topic_filter` matches the given `topic_name`. + * + * @Precondition `topic_filter` must be a valid MQTT topic filter. + * @Precondition `topic_name` must be a valid MQTT topic name. + */ +bool neu_mqtt_topic_filter_is_match(const char *topic_filter, + const char *topic_name); + typedef void (*neu_mqtt_client_connection_cb_t)(void *data); typedef void (*neu_mqtt_client_publish_cb_t)(int errcode, neu_mqtt_qos_e qos, char *topic, uint8_t *payload, diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c index 5ef4bc09..75e3fe9b 100644 --- a/src/connection/mqtt_client.c +++ b/src/connection/mqtt_client.c @@ -33,6 +33,7 @@ #include #include "connection/mqtt_client.h" +#include "errcodes.h" #include "event/event.h" #include "utils/asprintf.h" #include "utils/time.h" @@ -179,6 +180,93 @@ static inline uint8_t neu_mqtt_version_to_nng_mqtt_version(neu_mqtt_version_e v) } } +bool neu_mqtt_topic_filter_is_valid(const char *topic_filter) +{ + if (NULL == topic_filter || '\0' == *topic_filter || + strlen(topic_filter) > 65535) { + return false; + } + + char c; + while ((c = *topic_filter++)) { + switch (c) { + case '#': + // '#' must be last character + return '\0' == *topic_filter; + case '+': + // '+' must occupy an entire level + if (*topic_filter && '/' != *topic_filter) { + return false; + } + break; + default: + for (; '/' != c; c = *topic_filter++) { + if ('#' == c || '+' == c) { + return false; + } else if ('\0' == c) { + return true; + } + } + } + } + + return true; +} + +static bool topic_filter_is_match(const char *topic_filter, + const char *topic_name, + uint32_t topic_name_len) +{ + // The Server MUST NOT match Topic Filters starting with a wildcard + // character (# or +) with Topic Names beginning with a $ character + // [MQTT-4.7.2-1]. + if ('$' == topic_name[0]) { + if ('$' != topic_filter[0]) { + return false; + } + } else if (0 == strcmp("#", topic_filter)) { + return true; + } + + uint32_t i = 0; + while (true) { + switch (*topic_filter) { + case '#': + return true; + case '+': + while (i < topic_name_len && '/' != topic_name[i]) { + ++i; + }; + ++topic_filter; + break; + case '/': + if (i < topic_name_len && '/' == topic_name[i]) { + ++i; + } else if ('#' != topic_filter[1]) { // '#' include parent level + return false; + } + ++topic_filter; + break; + case '\0': + return i == topic_name_len; + default: + for (; *topic_filter && '/' != *topic_filter; ++topic_filter, ++i) { + if (!(i < topic_name_len && *topic_filter == topic_name[i])) { + return false; + } + } + } + } + + return true; +} + +bool neu_mqtt_topic_filter_is_match(const char *topic_filter, + const char *topic_name) +{ + return topic_filter_is_match(topic_filter, topic_name, strlen(topic_name)); +} + static inline task_t *task_new(neu_mqtt_client_t *client) { task_t *task = calloc(1, sizeof(*task)); @@ -405,6 +493,26 @@ static inline void subscriptions_free(subscription_t *subscriptions) } } +static inline subscription_t * +subscription_find_match(subscription_t *subscriptions, const char *topic_name, + uint32_t topic_name_len) +{ + subscription_t *sub = NULL, *tmp = NULL; + HASH_FIND(hh, subscriptions, topic_name, topic_name_len, sub); + if (NULL == sub) { + // subscription wildcard matching + // TODO: linear scanning is enough for current usages, + // but may pose performance issues in the future + HASH_ITER(hh, subscriptions, sub, tmp) + { + if (topic_filter_is_match(sub->topic, topic_name, topic_name_len)) { + return sub; + } + } + } + return sub; +} + static int resub_cb(void *data) { neu_mqtt_client_t *client = data; @@ -517,10 +625,9 @@ static void recv_cb(void *arg) log(debug, "recv [%.*s, QoS%d] %" PRIu32 " bytes", (unsigned) topic_len, topic, (int) qos, payload_len); - // find matching subscription - // NOTE: does not support wildcards nng_mtx_lock(client->mtx); - HASH_FIND(hh, client->subscriptions, topic, topic_len, subscription); + subscription = + subscription_find_match(client->subscriptions, topic, topic_len); if (NULL != subscription) { task_t *task = client_alloc_task(client); if (NULL != task) { @@ -1375,6 +1482,10 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos, int rv = 0; subscription_t *subscription = NULL; + if (!neu_mqtt_topic_filter_is_valid(topic)) { + return NEU_ERR_MQTT_SUBSCRIBE_FAILURE; + } + nng_mtx_lock(client->mtx); if (NULL == client->recv_aio) { if ((rv = nng_aio_alloc(&client->recv_aio, recv_cb, client)) != 0) { diff --git a/tests/ut/CMakeLists.txt b/tests/ut/CMakeLists.txt index f4feb6a2..88340072 100644 --- a/tests/ut/CMakeLists.txt +++ b/tests/ut/CMakeLists.txt @@ -64,6 +64,13 @@ target_include_directories(rolling_counter_test PRIVATE ) target_link_libraries(rolling_counter_test neuron-base gtest_main gtest) +add_executable(mqtt_client_test mqtt_client_test.cc) +target_include_directories(mqtt_client_test PRIVATE + ${CMAKE_SOURCE_DIR}/src + ${CMAKE_SOURCE_DIR}/include +) +target_link_libraries(mqtt_client_test neuron-base gtest_main gtest) + include(GoogleTest) gtest_discover_tests(json_test) gtest_discover_tests(http_test) @@ -73,3 +80,4 @@ gtest_discover_tests(tag_sort_test) gtest_discover_tests(modbus_test) gtest_discover_tests(async_queue_test) gtest_discover_tests(rolling_counter_test) +gtest_discover_tests(mqtt_client_test) diff --git a/tests/ut/mqtt_client_test.cc b/tests/ut/mqtt_client_test.cc new file mode 100644 index 00000000..0f7fab21 --- /dev/null +++ b/tests/ut/mqtt_client_test.cc @@ -0,0 +1,129 @@ +#include + +#include "connection/mqtt_client.h" +#include "utils/log.h" + +zlog_category_t *neuron = NULL; + +TEST(MQTTClientTest, test_topic_filter_is_valid) +{ + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid(NULL)); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("")); + + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/tennis#")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/tennis/#/ranking")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("#")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/#")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis/player/#")); + + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport+")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/+player")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+/+")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/+/player")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+/tennis/#")); + + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("$SYS")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("$SYS/#")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis/player")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_valid("sport/tennis/player/ranking")); +} + +TEST(MQTTClientTest, test_topic_filter_is_match) +{ + const char *filter = "#"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$sport")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player")); + + filter = "sport/tennis/player/#"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/play")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/")); + ASSERT_EQ( + true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match( + filter, "sport/tennis/player/ranking/score/wimbledon")); + + filter = "+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$finance")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "/finance")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "finance")); + + filter = "/+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "finance")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "/finance")); + + filter = "+/+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$finance")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "finance")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "/finance")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + + filter = "sport/+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/")); + ASSERT_EQ(false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/ranking")); + + filter = "sport/tennis/+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ( + false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player1")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player2")); + + filter = "sport/+/player"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/")); + ASSERT_EQ(false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/ranking")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player")); + ASSERT_EQ(false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/")); + ASSERT_EQ( + false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking")); + + filter = "+/tennis/#"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/baseball")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player")); + ASSERT_EQ( + true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking")); +} + +int main(int argc, char **argv) +{ + zlog_init("./config/dev.conf"); + neuron = zlog_get_category("neuron"); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From f2d851a5fa181a05a04d32896bbc97e59d309bcf Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 6 May 2024 16:21:40 +0800 Subject: [PATCH 2/4] fix(azure): clean configuration keys --- plugins/mqtt/azure-iot.json | 18 ++--------- plugins/mqtt/azure_iot_plugin.c | 54 +++++++++++++++++++++++++++------ plugins/mqtt/mqtt_config.c | 14 ++++----- plugins/mqtt/mqtt_config.h | 8 ++--- 4 files changed, 57 insertions(+), 37 deletions(-) diff --git a/plugins/mqtt/azure-iot.json b/plugins/mqtt/azure-iot.json index d2176cf9..14fd6ecd 100644 --- a/plugins/mqtt/azure-iot.json +++ b/plugins/mqtt/azure-iot.json @@ -1,5 +1,5 @@ { - "client-id": { + "device-id": { "name": "Device ID", "name_zh": "设备 ID", "description": "Azure IoT device ID, used as MQTT client id", @@ -84,7 +84,7 @@ ] } }, - "password": { + "sas": { "name": "SAS Token", "name_zh": "SAS 令牌", "description": "Azure IoT Hub shared access signature (SAS) token", @@ -99,20 +99,6 @@ "length": 255 } }, - "ssl": { - "name": "SSL", - "name_zh": "SSL", - "description": "Enable SSL connection", - "description_zh": "是否启用 SSL 连接", - "attribute": "optional", - "type": "bool", - "default": true, - "condition": { - "field": "invisible", - "value": true - }, - "valid": {} - }, "ca": { "name": "CA", "name_zh": "CA 证书", diff --git a/plugins/mqtt/azure_iot_plugin.c b/plugins/mqtt/azure_iot_plugin.c index 767bd005..c72724a3 100644 --- a/plugins/mqtt/azure_iot_plugin.c +++ b/plugins/mqtt/azure_iot_plugin.c @@ -28,6 +28,41 @@ #define AUTH_SAS 0 #define AUTH_CERT 1 +static int parse_ssl_params(neu_plugin_t *plugin, const char *setting, + neu_json_elem_t *ca, neu_json_elem_t *cert, + neu_json_elem_t *key) +{ + // ca, required + int ret = neu_parse_param(setting, NULL, 1, ca); + if (0 != ret) { + plog_notice(plugin, "setting no ca"); + return -1; + } + + if (0 != decode_b64_param(plugin, ca)) { + return -1; + } + + // cert, required + ret = neu_parse_param(setting, NULL, 1, cert); + if (0 != ret) { + plog_notice(plugin, "setting no cert"); + return -1; + } + + if (0 != decode_b64_param(plugin, cert)) { + return -1; + } + + // key, required + ret = parse_b64_param(plugin, setting, key); + if (0 != ret) { + return -1; + } + + return 0; +} + static int azure_parse_config(neu_plugin_t *plugin, const char *setting, mqtt_config_t *config) { @@ -38,17 +73,15 @@ static int azure_parse_config(neu_plugin_t *plugin, const char *setting, char * write_resp_topic = NULL; char * username = NULL; - neu_json_elem_t client_id = { .name = "client-id", .t = NEU_JSON_STR }; + neu_json_elem_t client_id = { .name = "device-id", .t = NEU_JSON_STR }; neu_json_elem_t qos = { .name = "qos", .t = NEU_JSON_INT }; neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT }; neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR }; neu_json_elem_t auth = { .name = "auth", .t = NEU_JSON_INT }; - neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR }; - neu_json_elem_t ssl = { .name = "ssl", .t = NEU_JSON_BOOL }; + neu_json_elem_t password = { .name = "sas", .t = NEU_JSON_STR }; neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR }; neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR }; neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR }; - neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR }; if (NULL == setting || NULL == config) { plog_error(plugin, "invalid argument, null pointer"); @@ -104,11 +137,9 @@ static int azure_parse_config(neu_plugin_t *plugin, const char *setting, password.v.val_str = strdup(""); } - ssl.v.val_bool = true; if (AUTH_CERT == auth.v.val_int) { - ret = - parse_ssl_params(plugin, setting, &ssl, &ca, &cert, &key, &keypass); - if (0 != ret || NULL == ca.v.val_str || NULL == cert.v.val_str) { + ret = parse_ssl_params(plugin, setting, &ca, &cert, &key); + if (0 != ret) { plog_error(plugin, "setting certificates fail"); goto error; } @@ -200,7 +231,6 @@ error: free(ca.v.val_str); free(cert.v.val_str); free(key.v.val_str); - free(keypass.v.val_str); return -1; } @@ -308,7 +338,11 @@ static int azure_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, break; } case NEU_REQ_SUBSCRIBE_GROUP: - case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: + case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: { + neu_req_subscribe_t *sub_info = data; + free(sub_info->params); + break; + } case NEU_REQ_UNSUBSCRIBE_GROUP: case NEU_REQ_UPDATE_GROUP: case NEU_REQ_DEL_GROUP: diff --git a/plugins/mqtt/mqtt_config.c b/plugins/mqtt/mqtt_config.c index 1f38943e..0be384a6 100644 --- a/plugins/mqtt/mqtt_config.c +++ b/plugins/mqtt/mqtt_config.c @@ -26,7 +26,7 @@ #define MB 1000000 -static inline int decode_b64_param(neu_plugin_t *plugin, neu_json_elem_t *el) +int decode_b64_param(neu_plugin_t *plugin, neu_json_elem_t *el) { int len = 0; char *s = (char *) neu_decode64(&len, el->v.val_str); @@ -47,8 +47,8 @@ static inline int decode_b64_param(neu_plugin_t *plugin, neu_json_elem_t *el) return 0; } -static inline int parse_b64_param(neu_plugin_t *plugin, const char *setting, - neu_json_elem_t *el) +int parse_b64_param(neu_plugin_t *plugin, const char *setting, + neu_json_elem_t *el) { if (0 != neu_parse_param(setting, NULL, 1, el)) { plog_error(plugin, "setting no %s", el->name); @@ -64,10 +64,10 @@ static inline int parse_b64_param(neu_plugin_t *plugin, const char *setting, return 0; } -int parse_ssl_params(neu_plugin_t *plugin, const char *setting, - neu_json_elem_t *ssl, neu_json_elem_t *ca, - neu_json_elem_t *cert, neu_json_elem_t *key, - neu_json_elem_t *keypass) +static int parse_ssl_params(neu_plugin_t *plugin, const char *setting, + neu_json_elem_t *ssl, neu_json_elem_t *ca, + neu_json_elem_t *cert, neu_json_elem_t *key, + neu_json_elem_t *keypass) { // ssl, optional int ret = neu_parse_param(setting, NULL, 1, ssl); diff --git a/plugins/mqtt/mqtt_config.h b/plugins/mqtt/mqtt_config.h index 066c18bc..e1cbccf1 100644 --- a/plugins/mqtt/mqtt_config.h +++ b/plugins/mqtt/mqtt_config.h @@ -70,10 +70,10 @@ typedef struct { // for backward compatibility } mqtt_config_t; -int parse_ssl_params(neu_plugin_t *plugin, const char *setting, - neu_json_elem_t *ssl, neu_json_elem_t *ca, - neu_json_elem_t *cert, neu_json_elem_t *key, - neu_json_elem_t *keypass); +int decode_b64_param(neu_plugin_t *plugin, neu_json_elem_t *el); +int parse_b64_param(neu_plugin_t *plugin, const char *setting, + neu_json_elem_t *el); + int mqtt_config_parse(neu_plugin_t *plugin, const char *setting, mqtt_config_t *config); void mqtt_config_fini(mqtt_config_t *config); From c39747b5854c817cc17170ae84ed541748dfb813 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 6 May 2024 18:58:32 +0800 Subject: [PATCH 3/4] test(azure): add ft --- tests/ft/app/test_azure.py | 494 +++++++++++++++++++++++++++++++++++++ 1 file changed, 494 insertions(+) create mode 100644 tests/ft/app/test_azure.py diff --git a/tests/ft/app/test_azure.py b/tests/ft/app/test_azure.py new file mode 100644 index 00000000..f9567670 --- /dev/null +++ b/tests/ft/app/test_azure.py @@ -0,0 +1,494 @@ +import json +import time + +import pytest + +import neuron.api as api +import neuron.error as error +import neuron.config as config +from neuron.common import ( + case_time, + class_setup_and_teardown, + compare_float, + description, + process, + random_port, +) +from neuron.mqtt import Mock + + +NODE = "azure" +DRIVER = "modbus" +GROUP = "group" +INTERVAL = 100 +TAG0 = { + "name": "tag0", + "address": "1!400001.", + "attribute": config.NEU_TAG_ATTRIBUTE_RW, + "type": config.NEU_TYPE_INT16, +} +TAGS = [ + { + "name": "tag1", + "address": "1!000001", + "attribute": config.NEU_TAG_ATTRIBUTE_RW, + "type": config.NEU_TYPE_BIT, + }, + { + "name": "tag2", + "address": "1!400017", + "attribute": config.NEU_TAG_ATTRIBUTE_RW, + "type": config.NEU_TYPE_FLOAT, + }, + { + "name": "tag3", + "address": "1!400020.10", + "attribute": config.NEU_TAG_ATTRIBUTE_RW, + "type": config.NEU_TYPE_STRING, + }, + { + "name": "tag4", + "address": "1!400040.4", + "attribute": config.NEU_TAG_ATTRIBUTE_RW, + "type": config.NEU_TYPE_BYTES, + }, +] +FORMAT_VALUES = 0 +FORMAT_TAGS = 1 +HUB = "127.0.0.1" +DEVICE_ID = "client-005" +USER = f"{HUB}/{DEVICE_ID}/?api-version=2021-04-12" +SAS = "neuron" +PASSWORD = SAS +D2C_TOPIC = f"devices/{DEVICE_ID}/messages/events/" +C2D_TOPIC = f"devices/{DEVICE_ID}/messages/devicebound/write" + + +@pytest.fixture(autouse=True, scope="class") +def azure_node(class_setup_and_teardown): + api.add_node_check(NODE, "Azure IoT") + yield + api.del_node(NODE) + + +@pytest.fixture(autouse=True, scope="class") +def modbus_node(azure_node): + try: + api.add_node_check(DRIVER, "Modbus TCP") + api.add_group_check(DRIVER, GROUP, INTERVAL) + api.add_tags_check(DRIVER, GROUP, tags=[TAG0]) + api.subscribe_group_check(NODE, DRIVER, GROUP) + port = random_port() + p = process.start_simulator( + ["./modbus_simulator", "tcp", f"{port}", "ip_v4"] + ) + response = api.modbus_tcp_node_setting(DRIVER, port) + assert 200 == response.status_code + assert error.NEU_ERR_SUCCESS == response.json()["error"] + yield + finally: + api.del_node(DRIVER) + process.stop_simulator(p) + + +@pytest.fixture(autouse=True, scope="class") +def mocker(): + mock = Mock(user=USER, password=PASSWORD) + yield mock + mock.done() + + +def reqresp(mocker, resp_topic, req_topic, payload, qos=0, timeout=1): + try: + mocker.client.subscribe(resp_topic) + start = time.time_ns() + # try request until response received + while time.time_ns() - start <= timeout * 1_000_000_000: + mocker.pub(req_topic, payload, qos=qos) + msg = mocker.client.listen(resp_topic, timeout=0.1) + if msg: + msg = json.loads(msg) + if "uuid" in msg: + return msg + finally: + mocker.client.unsubscribe(resp_topic) + + +@pytest.fixture(scope="function") +def conf_base(mocker): + return { + "device-id": DEVICE_ID, + "qos": 0, + "format": FORMAT_VALUES, + "host": HUB, + "auth": 0, + "sas": SAS, + } + + +@pytest.fixture(scope="function") +def conf_fmt_tags(conf_base): + return { + **conf_base, + "format": FORMAT_TAGS, + } + + +@pytest.fixture(scope="function") +def conf_ssl(conf_base, mocker): + return { + **conf_base, + "auth": 1, + "ca": mocker.ca_base64, + "cert": mocker.client_cert_base64, + "key": mocker.client_key_base64, + } + + +@pytest.fixture(scope="function") +def conf_no_id(conf_base): + del conf_base["device-id"] + + +@pytest.fixture(scope="function") +def conf_bad_id(conf_base): + return { + **conf_base, + "device-id": "", + } + + +@pytest.fixture(scope="function") +def conf_bad_qos(conf_base): + return { + **conf_base, + "qos": 2, + } + + +@pytest.fixture(scope="function") +def conf_bad_fmt(conf_base): + return { + **conf_base, + "format": -1, + } + + +@pytest.fixture(scope="function") +def conf_ssl_bad_ca(conf_ssl): + return { + **conf_ssl, + "ca": "", + } + + +@pytest.fixture(scope="function") +def conf_ssl_bad_cert(conf_ssl): + return { + **conf_ssl, + "cert": "(@$&*)", + } + + +@pytest.fixture(scope="function") +def conf_ssl_no_cert(conf_base, mocker): + return { + **conf_base, + "port": mocker.ssl_port, + "auth": 1, + "ca": mocker.ca_base64, + } + + +@pytest.fixture(scope="function") +def conf_ssl_no_key(conf_ssl, mocker): + conf = { + **conf_ssl, + } + del conf["key"] + return conf + + +@pytest.fixture(scope="function") +def conf_ssl_bad_key(conf_ssl): + return { + **conf_ssl, + "key": "", + } + + +class TestAzure: + @description( + given="Azure node and {conf}", + when="config node", + then="should fail", + ) + @pytest.mark.parametrize( + "conf", + [ + "conf_no_id", + "conf_bad_id", + "conf_bad_qos", + "conf_bad_fmt", + "conf_ssl_bad_ca", + "conf_ssl_no_cert", + "conf_ssl_bad_cert", + "conf_ssl_no_key", + "conf_ssl_bad_key", + ], + ) + def test_azure_conf(self, request, conf): + conf = request.getfixturevalue(conf) + response = api.node_setting(NODE, conf) + assert 400 == response.status_code + assert error.NEU_ERR_NODE_SETTING_INVALID == response.json()["error"] + + @description( + given="Azure node and {conf}", + when="after configuration", + then="should connect to the broker", + ) + @pytest.mark.parametrize( + "conf", + [ + "conf_base", + "conf_ssl", + ], + ) + def test_azure_connect(self, request, mocker, conf): + conf = request.getfixturevalue(conf) + api.node_setting_check(NODE, conf) + api.node_ctl(NODE, config.NEU_CTL_START) + mocker.assert_client_connected() + + @description( + given="Azure node", + when="stop/start node", + then="should disconnect/connect broker", + ) + def test_azure_start_stop(self, request, mocker, conf_base): + api.node_setting_check(NODE, conf_base) + api.node_ctl_check(NODE, config.NEU_CTL_STOP) + mocker.assert_client_disconnected() + api.node_ctl_check(NODE, config.NEU_CTL_START) + mocker.assert_client_connected() + + @description( + given="Azure node and {conf}", + when="after configuration", + then="broker should receive data on upload topic", + ) + @pytest.mark.parametrize("conf", ["conf_base", "conf_fmt_tags"]) + def test_azure_upload(self, mocker, request, conf): + fmt_tags = conf == "conf_fmt_tags" + conf = request.getfixturevalue(conf) + api.node_setting_check(NODE, conf) + api.node_ctl(NODE, config.NEU_CTL_START) + + msg = mocker.get(D2C_TOPIC, timeout=3) + assert msg is not None + assert DRIVER == msg["node"] + assert GROUP == msg["group"] + if fmt_tags: + assert TAG0["name"] in [tag["name"] for tag in msg["tags"]] + else: + assert ( + TAG0["name"] in msg["values"] or TAG0["name"] in msg["errors"] + ) + + try: + api.add_tags_check(DRIVER, GROUP, TAGS) + msg = mocker.get(D2C_TOPIC, timeout=3) + assert msg is not None + assert DRIVER == msg["node"] + assert GROUP == msg["group"] + if fmt_tags: + len(TAGS) + 1 == len(msg["tags"]) + else: + len(TAGS) + 1 == len(msg["values"]) + len(msg["errors"]) + finally: + api.del_tags(DRIVER, GROUP, [tag["name"] for tag in TAGS]) + + @description( + given="Azure node", + when="after renaming subscribed driver", + then="broker should receive data on upload topic", + ) + def test_azure_upload_when_driver_rename(self, mocker, conf_fmt_tags): + api.node_setting_check(NODE, conf_fmt_tags) + api.node_ctl(NODE, config.NEU_CTL_START) + + new_name = DRIVER + "-new" + api.update_node_check(DRIVER, new_name) + + msg = mocker.get(D2C_TOPIC, timeout=3) + try: + assert msg is not None + assert new_name == msg["node"] + assert GROUP == msg["group"] + assert TAG0["name"] in [tag["name"] for tag in msg["tags"]] + finally: + api.update_node_check(new_name, DRIVER) + + @description( + given="Azure node", + when="after renaming subscribed driver group", + then="broker should receive data on upload topic", + ) + def test_azure_upload_when_group_rename(self, mocker, conf_fmt_tags): + api.node_setting_check(NODE, conf_fmt_tags) + api.node_ctl(NODE, config.NEU_CTL_START) + + new_name = GROUP + "-new" + api.update_group_check(DRIVER, GROUP, new_name) + + msg = mocker.get(D2C_TOPIC, timeout=3) + try: + assert msg is not None + assert DRIVER == msg["node"] + assert new_name == msg["group"] + assert TAG0["name"] in [tag["name"] for tag in msg["tags"]] + finally: + api.update_group_check(DRIVER, new_name, GROUP) + + @description( + given="Azure node", + when="delete subscribed driver group and add back group", + then="broker should receive data on upload topic", + ) + def test_azure_upload_when_group_deleted(self, mocker, conf_fmt_tags): + api.node_setting_check(NODE, conf_fmt_tags) + api.node_ctl(NODE, config.NEU_CTL_START) + + # delete driver group + api.del_group_check(DRIVER, GROUP) + # add back driver group + api.add_group_check(DRIVER, GROUP, INTERVAL) + api.add_tags_check(DRIVER, GROUP, tags=[TAG0]) + # add back subscription on new topic + api.subscribe_group_check(NODE, DRIVER, GROUP) + + msg = mocker.get(D2C_TOPIC, timeout=3) + assert msg is not None + assert DRIVER == msg["node"] + assert GROUP == msg["group"] + assert TAG0["name"] in [tag["name"] for tag in msg["tags"]] + + @description( + given="Azure node", + when="send bad write req", + then="should fail", + ) + @pytest.mark.parametrize( + "req", + [ + { + "tag": TAG0["name"], + }, + { + "tags": [ + { + "tag": TAG0["name"], + }, + ], + }, + ], + ) + def test_azure_write_bad_req(self, mocker, conf_base, req): + api.node_setting_check(NODE, conf_base) + api.node_ctl(NODE, config.NEU_CTL_START) + mocker.assert_client_connected() + + req = { + "uuid": "cd32be1b-c8b1-3257-94af-77f847b1ed3e", + "node": DRIVER, + "group": GROUP, + **req, + } + mocker.pub(C2D_TOPIC, req) + del req["uuid"] + mocker.pub(C2D_TOPIC, req) + + @description( + given="Azure node", + when="write a driver tag", + then="should success", + ) + def test_azure_write_tag(self, mocker, conf_base): + api.node_setting_check(NODE, conf_base) + api.node_ctl(NODE, config.NEU_CTL_START) + mocker.assert_client_connected() + + req = { + "uuid": "cd32be1b-c8b1-3257-94af-77f847b1ed3e", + "node": DRIVER, + "group": GROUP, + "tag": TAG0["name"], + "value": 1234, + } + msg = reqresp(mocker, D2C_TOPIC, C2D_TOPIC, req, timeout=3) + assert msg is not None + assert req["uuid"] == msg["uuid"] + assert error.NEU_ERR_SUCCESS == msg["error"] + + time.sleep(INTERVAL * 5 / 1000) + assert 1234 == api.read_tag(DRIVER, GROUP, TAG0["name"]) + + @description( + given="Azure node", + when="write driver many tags", + then="should success", + ) + def test_azure_write_tags(self, mocker, conf_base): + api.node_setting_check(NODE, conf_base) + api.node_ctl(NODE, config.NEU_CTL_START) + mocker.assert_client_connected() + + req = { + "uuid": "cd32be1b-c8b1-3257-94af-77f847b1ed3e", + "node": DRIVER, + "group": GROUP, + "tags": [ + { + "tag": TAG0["name"], + "value": 1234, + }, + { + "tag": "tag1", + "value": True, + }, + { + "tag": "tag2", + "value": 3.14, + }, + { + "tag": "tag3", + "value": "hello", + }, + { + "tag": "tag4", + "value": [1, 2, 3, 4], + }, + ], + } + + try: + api.add_tags_check(DRIVER, GROUP, TAGS) + msg = reqresp(mocker, D2C_TOPIC, C2D_TOPIC, req, timeout=3) + assert msg is not None + assert req["uuid"] == msg["uuid"] + assert error.NEU_ERR_SUCCESS == msg["error"] + + time.sleep(INTERVAL * 5 / 1000) + resp = api.read_tags(DRIVER, GROUP) + assert 200 == resp.status_code + resp_tags = sorted(resp.json()["tags"], key=lambda x: x["name"]) + print(resp_tags) + req_tags = req["tags"] + assert req_tags[0]["value"] == resp_tags[0]["value"] + assert req_tags[1]["value"] == resp_tags[1]["value"] + assert compare_float(req_tags[2]["value"], resp_tags[2]["value"]) + assert req_tags[3]["value"] == resp_tags[3]["value"] + assert req_tags[4]["value"] == resp_tags[4]["value"] + finally: + api.del_tags(DRIVER, GROUP, [tag["name"] for tag in TAGS]) From 68ca03f19771607610bfebef6921a6aeff07d679 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 7 May 2024 11:22:31 +0800 Subject: [PATCH 4/4] ci: Azure IoT plugin codecov report --- .github/workflows/check.yml | 7 +++++-- create_cov_report.sh | 4 ++++ 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index fc707f0f..a7487af0 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -81,7 +81,7 @@ jobs: container: ghcr.io/neugates/build:x86_64-main strategy: matrix: - plugin: [core, ekuiper, modbus, mqtt, metrics] + plugin: [core, ekuiper, modbus, mqtt, azure, metrics] steps: - uses: actions/checkout@v4 @@ -90,7 +90,7 @@ jobs: - name: setup dependencies run: | - if [ "${{ matrix.plugin }}" = "mqtt" ]; then + if [ "${{ matrix.plugin }}" = "mqtt" ] || [ "${{ matrix.plugin }}" = "azure" ]; then sudo apt-get install -y mosquitto fi pip install -U pytest requests prometheus-client gmqtt pynng sniffio @@ -109,6 +109,7 @@ jobs: pytest -s -v tests/ft \ --ignore=tests/ft/app/test_ekuiper.py \ --ignore=tests/ft/app/test_mqtt.py \ + --ignore=tests/ft/app/test_azure.py \ --ignore=tests/ft/driver/test_modbus.py \ --ignore=tests/ft/metrics/test_metrics.py elif [ "${{ matrix.plugin }}" = "ekuiper" ]; then @@ -117,6 +118,8 @@ jobs: pytest -s -v tests/ft/driver/"test_modbus.py" elif [ "${{ matrix.plugin }}" = "mqtt" ]; then pytest -s -v tests/ft/app/"test_mqtt.py" + elif [ "${{ matrix.plugin }}" = "azure" ]; then + pytest -s -v tests/ft/app/"test_azure.py" else pytest -s -v tests/ft/metrics/"test_metrics.py" fi diff --git a/create_cov_report.sh b/create_cov_report.sh index f8817b3d..a1e49f77 100755 --- a/create_cov_report.sh +++ b/create_cov_report.sh @@ -39,6 +39,10 @@ case $1 in lcov -c -d build/plugins/mqtt/CMakeFiles/plugin-mqtt.dir -o cov_report/cov-mqtt.info cd cov_report gen_trace -a cov-mqtt.info -o cov-mqtt.info;; + (azure) + lcov -c -d build/plugins/mqtt/CMakeFiles/plugin-azure-iot.dir -o cov_report/cov-azure.info + cd cov_report + gen_trace -a cov-azure.info -o cov-azure.info;; (core) cd cov_report gen_trace -o cov-core.info;;