Merge pull request #1977 from eeff/main

test(azure): add ft

.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

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;;

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,

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 证书",

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:

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);

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);

src/connection/mqtt_client.c

@@ -33,6 +33,7 @@
 #include <nng/supplemental/util/platform.h>
 
 #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) {

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])

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)

tests/ut/mqtt_client_test.cc

@@ -0,0 +1,129 @@
+#include <gtest/gtest.h>
+
+#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();
+}