bumper/mqttserver.py

@@ -27,7 +27,7 @@ class MQTTHelperBot:
 
     def __init__(self, address):
         self.address = address
-        self.client_id = "helper1@bumper/helper1"
+        self.client_id = "helperbot@bumper/helperbot"
         self.command_responses = []
 
     async def start_helper_bot(self):
@@ -44,89 +44,26 @@ class MQTTHelperBot:
             )
             await self.Client.subscribe(
                 [
-                    ("iot/p2p/+/+/+/+/helper1/bumper/helper1/+/+/+", QOS_0),
+                    ("iot/p2p/+/+/+/+/helperbot/bumper/helperbot/+/+/+", QOS_0),
                     ("iot/p2p/+", QOS_0),
                     ("iot/atr/+", QOS_0),
                 ]
             )
 
-            asyncio.create_task(self.get_msg())
+#        except ConnectionRefusedError as e:
+#            helperbotlog.Error(e)
+#            pass
 
-        except ConnectionRefusedError as e:
+#        except asyncio.CancelledError as e:
-            helperbotlog.Error(e)
+#            pass
-            pass
 
-        except asyncio.CancelledError as e:
+#        except hbmqtt.client.ConnectException as e:
-            pass
+#            helperbotlog.Error(e)
-
+#            pass
-        except hbmqtt.client.ConnectException as e:
-            helperbotlog.Error(e)
-            pass
 
         except Exception as e:
             helperbotlog.exception("{}".format(e))
 
-    async def get_msg(self):
-        while True:
-            message = await self.Client.deliver_message()
-
-            if str(message.topic).split("/")[6] == "helper1":
-                # Response to command
-                helperbotlog.debug(
-                    "Received Response - Topic: {} - Message: {}".format(
-                        message.topic, str(message.data.decode("utf-8"))
-                    )
-                )
-                self.command_responses.append(
-                    {
-                        "time": time.time(),
-                        "topic": message.topic,
-                        "payload": str(message.data.decode("utf-8")),
-                    }
-                )
-            elif str(message.topic).split("/")[3] == "helper1":
-                # Helperbot sending command
-                helperbotlog.debug(
-                    "Send Command - Topic: {} - Message: {}".format(
-                        message.topic, str(message.data.decode("utf-8"))
-                    )
-                )
-            elif str(message.topic).split("/")[1] == "atr":
-                # Broadcast message received on atr
-                if str(message.topic).split("/")[2] == "errors":
-                    boterrorlog.error(
-                        "Received Error - Topic: {} - Message: {}".format(
-                            message.topic, str(message.data.decode("utf-8"))
-                        )
-                    )
-                else:
-                    helperbotlog.debug(
-                        "Received Broadcast - Topic: {} - Message: {}".format(
-                            message.topic, str(message.data.decode("utf-8"))
-                        )
-                    )
-
-            else:
-                helperbotlog.debug(
-                    "Received Message - Topic: {} - Message: {}".format(
-                        message.topic, str(message.data.decode("utf-8"))
-                    )
-                )
-
-            # Cleanup "expired messages" > 60 seconds from time
-            for msg in self.command_responses:
-                expire_time = (
-                    datetime.fromtimestamp(msg["time"])
-                    + timedelta(seconds=self.expire_msg_seconds)
-                ).timestamp()
-                if time.time() > expire_time:
-                    helperbotlog.debug(
-                        "Pruning Message Due To Expiration - Message Topic: {}".format(
-                            msg["topic"]
-                        )
-                    )
-                    self.command_responses.remove(msg)
-
     async def wait_for_resp(self, requestid):
         try:
 
@@ -139,8 +76,7 @@ class MQTTHelperBot:
                 if len(self.command_responses) > 0:
                     for msg in self.command_responses:
                         topic = str(msg["topic"]).split("/")
-                        if topic[6] == "helper1" and topic[10] == requestid:
+                        if topic[6] == "helperbot" and topic[10] == requestid:
-                            # helperbotlog.debug('VacBot MQTT Response: Topic: %s Payload: %s' % (msg['topic'], msg['payload']))
                             if topic[11] == "j":
                                 resppayload = json.loads(msg["payload"])
                             else:
@@ -175,7 +111,7 @@ class MQTTHelperBot:
     async def send_command(self, cmdjson, requestid):
         if not self.Client._handler.writer is None:
             try:
-                ttopic = "iot/p2p/{}/helper1/bumper/helper1/{}/{}/{}/q/{}/{}".format(
+                ttopic = "iot/p2p/{}/helperbot/bumper/helperbot/{}/{}/{}/q/{}/{}".format(
                     cmdjson["cmdName"],
                     cmdjson["toId"],
                     cmdjson["toType"],
@@ -298,13 +234,6 @@ class BumperMQTTServer_Plugin:
 
     async def authenticate(self, *args, **kwargs):
         authenticated = False
-        if not self.auth_config:
-            # auth config section not found
-            self.context.logger.warning(
-                "'auth' section not found in context configuration"
-            )
-            return False
-
         
         try:
             session = kwargs.get("session", None)
@@ -334,7 +263,7 @@ class BumperMQTTServer_Plugin:
                     realm = tmpclientdetail[0]
                     resource = tmpclientdetail[1]
 
-                    if userid == "helper1":
+                    if userid == "helperbot":
                         mqttserverlog.info(f"Bumper Authentication Success - Helperbot: {client_id}")
                         authenticated = True
                     else:
@@ -413,9 +342,67 @@ class BumperMQTTServer_Plugin:
             bumper.client_set_mqtt(client["resource"], True)
             return
 
-    #async def on_broker_message_received(self, client_id, message):
+    async def on_broker_message_received(self, client_id, message):
-        #print(message)
+        self.handle_helperbot_msg(client_id, message)
-        # Look at replacing helperbot with code here
+        
+    def handle_helperbot_msg(self, client_id, message):
+
+            if str(message.topic).split("/")[6] == "helperbot":
+                # Response to command
+                helperbotlog.debug(
+                    "Received Response - Topic: {} - Message: {}".format(
+                        message.topic, str(message.data.decode("utf-8"))
+                    )
+                )
+                bumper.mqtt_helperbot.command_responses.append(
+                    {
+                        "time": time.time(),
+                        "topic": message.topic,
+                        "payload": str(message.data.decode("utf-8")),
+                    }
+                )
+            elif str(message.topic).split("/")[3] == "helperbot":
+                # Helperbot sending command
+                helperbotlog.debug(
+                    "Send Command - Topic: {} - Message: {}".format(
+                        message.topic, str(message.data.decode("utf-8"))
+                    )
+                )
+            elif str(message.topic).split("/")[1] == "atr":
+                # Broadcast message received on atr
+                if str(message.topic).split("/")[2] == "errors":
+                    boterrorlog.error(
+                        "Received Error - Topic: {} - Message: {}".format(
+                            message.topic, str(message.data.decode("utf-8"))
+                        )
+                    )
+                else:
+                    helperbotlog.debug(
+                        "Received Broadcast - Topic: {} - Message: {}".format(
+                            message.topic, str(message.data.decode("utf-8"))
+                        )
+                    )
+
+            else:
+                helperbotlog.debug(
+                    "Received Message - Topic: {} - Message: {}".format(
+                        message.topic, str(message.data.decode("utf-8"))
+                    )
+                )
+
+            # Cleanup "expired messages" > 60 seconds from time
+            for msg in bumper.mqtt_helperbot.command_responses:
+                expire_time = (
+                    datetime.fromtimestamp(msg["time"])
+                    + timedelta(seconds=bumper.mqtt_helperbot.expire_msg_seconds)
+                ).timestamp()
+                if time.time() > expire_time:
+                    helperbotlog.debug(
+                        "Pruning Message Due To Expiration - Message Topic: {}".format(
+                            msg["topic"]
+                        )
+                    )
+                    bumper.mqtt_helperbot.command_responses.remove(msg)
 
 
     async def on_broker_client_disconnected(self, client_id):

tests/test_mqttserver.py

@@ -31,10 +31,9 @@ async def test_helperbot_message():
         await mqtt_helperbot.Client.publish(
             msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
         )
-        try:
+
         await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1)
-        except asyncio.TimeoutError:
+
-            pass
         l.check_present(
             (
                 "helperbot",
@@ -53,20 +52,19 @@ async def test_helperbot_message():
         )  # Check helperbot is connected
         msg_payload = "{}"
         msg_topic_name = (
-            "iot/p2p/GetWKVer/helper1/bumper/helper1/bot_serial/ls1ok3/wC3g/q/iCmuqp/j"
+            "iot/p2p/GetWKVer/helperbot/bumper/helperbot/bot_serial/ls1ok3/wC3g/q/iCmuqp/j"
         )
         await mqtt_helperbot.Client.publish(
             msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
         )
-        try:
+        
         await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1)
-        except asyncio.TimeoutError:
+
-            pass
         l.check_present(
             (
                 "helperbot",
                 "DEBUG",
-                "Send Command - Topic: iot/p2p/GetWKVer/helper1/bumper/helper1/bot_serial/ls1ok3/wC3g/q/iCmuqp/j - Message: {}",
+                "Send Command - Topic: iot/p2p/GetWKVer/helperbot/bumper/helperbot/bot_serial/ls1ok3/wC3g/q/iCmuqp/j - Message: {}",
             )
         )  # Check send command message was logged
         l.clear()
@@ -80,20 +78,19 @@ async def test_helperbot_message():
         )  # Check helperbot is connected
         msg_payload = '{"ret":"ok","ver":"0.13.5"}'
         msg_topic_name = (
-            "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helper1/bumper/helper1/p/iCmuqp/j"
+            "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/iCmuqp/j"
         )
         await mqtt_helperbot.Client.publish(
             msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
         )
-        try:
+
         await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1)
-        except asyncio.TimeoutError:
+
-            pass
         l.check_present(
             (
                 "helperbot",
                 "DEBUG",
-                'Received Response - Topic: iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helper1/bumper/helper1/p/iCmuqp/j - Message: {"ret":"ok","ver":"0.13.5"}',
+                'Received Response - Topic: iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/iCmuqp/j - Message: {"ret":"ok","ver":"0.13.5"}',
             )
         )  # Check received response message was logged
         l.clear()
@@ -107,21 +104,20 @@ async def test_helperbot_message():
         )  # Check helperbot is connected
         msg_payload = "test"
         msg_topic_name = (
-            "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/TESTBAD/bumper/helper1/p/iCmuqp/j"
+            "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/TESTBAD/bumper/helperbot/p/iCmuqp/j"
         )
         await mqtt_helperbot.Client.publish(
             msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
         )
-        try:
+
         await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1)
-        except asyncio.TimeoutError:
+
-            pass
 
         l.check_present(
             (
                 "helperbot",
                 "DEBUG",
-                "Received Message - Topic: iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/TESTBAD/bumper/helper1/p/iCmuqp/j - Message: test",
+                "Received Message - Topic: iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/TESTBAD/bumper/helperbot/p/iCmuqp/j - Message: test",
             )
         )  # Check received message was logged
         l.clear()
@@ -138,10 +134,8 @@ async def test_helperbot_message():
         await mqtt_helperbot.Client.publish(
             msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
         )
-        try:
+            
         await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1)
-        except asyncio.TimeoutError:
-            pass
 
         l.check_present(
             (
@@ -165,13 +159,14 @@ async def test_helperbot_expire_message():
 
         # Test broadcast message
         mqtt_helperbot = bumper.MQTTHelperBot(mqtt_address)
+        bumper.mqtt_helperbot = mqtt_helperbot
         await mqtt_helperbot.start_helper_bot()
         assert (
             mqtt_helperbot.Client._connected_state._value == True
         )  # Check helperbot is connected
 
         expire_msg_payload = '{"ret":"ok","ver":"0.13.5"}'
-        expire_msg_topic_name = "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helper1/bumper/helper1/p/testgood/j"
+        expire_msg_topic_name = "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/testgood/j"
         currenttime = time.time()
         mqtt_helperbot.command_responses.append(
             {
@@ -197,10 +192,9 @@ async def test_helperbot_expire_message():
             msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
         )  # Send another message to force get_msg
 
-        try:
+
         await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1)
-        except asyncio.TimeoutError:
+
-            pass
 
         assert {
             "time": currenttime,
@@ -229,6 +223,7 @@ async def test_helperbot_sendcommand():
     await mqtt_server.broker_coro()
 
     mqtt_helperbot = bumper.MQTTHelperBot(mqtt_address)
+    bumper.mqtt_helperbot = mqtt_helperbot
     await mqtt_helperbot.start_helper_bot()
     assert (
         mqtt_helperbot.Client._connected_state._value == True
@@ -268,7 +263,7 @@ async def test_helperbot_sendcommand():
     # Send response beforehand
     msg_payload = '{"ret":"ok","ver":"0.13.5"}'
     msg_topic_name = (
-        "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helper1/bumper/helper1/p/testgood/j"
+        "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/testgood/j"
     )
     await mqtt_helperbot.Client.publish(
         msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
@@ -307,7 +302,7 @@ async def test_helperbot_sendcommand():
     # Send response beforehand
     msg_payload = "<ctl ret='ok' type='Brush' left='4142' total='18000'/>"
     msg_topic_name = (
-        "iot/p2p/GetLifeSpan/bot_serial/ls1ok3/wC3g/helper1/bumper/helper1/p/testx/q"
+        "iot/p2p/GetLifeSpan/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/testx/q"
     )
     await mqtt_helperbot.Client.publish(
         msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
@@ -352,7 +347,7 @@ async def test_helperbot_sendcommand():
     msg_payload = '{"body":{"code":0,"data":{"area":0,"cid":"111","start":"1569378657","time":6,"type":"auto"},"msg":"ok"},"header":{"fwVer":"1.6.4","hwVer":"0.1.1","pri":1,"ts":"1569380074036","tzm":480,"ver":"0.0.1"}}'
     
     msg_topic_name = (
-        "iot/p2p/getStats/bot_serial/ls1ok3/wC3g/helper1/bumper/helper1/p/testj/j"
+        "iot/p2p/getStats/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/testj/j"
     )
     await mqtt_helperbot.Client.publish(
         msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0
@@ -447,28 +442,22 @@ async def test_mqttserver():
     
     # bad password
     with LogCapture() as l:
-        try:
+        
         await test_client.Client.connect(
             f"mqtts://test-client:notvalid!@{test_client.address[0]}:{test_client.address[1]}/",
             cafile=bumper.ca_cert, cleansession=True
         )
 
-        except Exception as ae:
-            pass
-
         l.check_present(
                 ("mqttserver", "INFO", "File Authentication Failed - Username: test-client - ClientID: test-file-auth"),
                 order_matters=False
             )
     # no username in file    
-        try:
         await test_client.Client.connect(
             f"mqtts://test-client-noexist:notvalid!@{test_client.address[0]}:{test_client.address[1]}/",
             cafile=bumper.ca_cert, cleansession=True
         )
 
-        except Exception as ae:
-            pass    
 
         l.check_present(
             ("mqttserver", "INFO", 'File Authentication Failed - No Entry for Username: test-client-noexist - ClientID: test-file-auth'),