From 70fdfdb1b3bcf1c891b0e943ea5d5eebc12a18bd Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Fri, 3 Jan 2020 14:23:37 -0500 Subject: [PATCH] helperbot get_msg - moved get_msg handler from helperbot to broker --- bumper/mqttserver.py | 161 ++++++++++++++++++--------------------- tests/test_mqttserver.py | 89 ++++++++++------------ 2 files changed, 113 insertions(+), 137 deletions(-) diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index e875438..f53d1f8 100644 --- a/bumper/mqttserver.py +++ b/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: - helperbotlog.Error(e) - pass +# except asyncio.CancelledError as e: +# pass - except asyncio.CancelledError as e: - pass - - except hbmqtt.client.ConnectException as e: - 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: - # helperbotlog.debug('VacBot MQTT Response: Topic: %s Payload: %s' % (msg['topic'], msg['payload'])) + if topic[6] == "helperbot" and topic[10] == requestid: 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): - #print(message) - # Look at replacing helperbot with code here + async def on_broker_message_received(self, client_id, message): + self.handle_helperbot_msg(client_id, message) + + 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): diff --git a/tests/test_mqttserver.py b/tests/test_mqttserver.py index f58cddf..70205bc 100644 --- a/tests/test_mqttserver.py +++ b/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 + + await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + 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 + + await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + 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 + + await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + 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 + + await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + 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 + + await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) 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 + + await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + 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 = "" 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,29 +442,23 @@ 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 + + await test_client.Client.connect( + f"mqtts://test-client:notvalid!@{test_client.address[0]}:{test_client.address[1]}/", + cafile=bumper.ca_cert, cleansession=True + ) 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 - ) + 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'), order_matters=False