From c999be7057498268ad16bf31f052f7d7cec56ae6 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sun, 30 Jan 2022 19:00:51 +0100 Subject: [PATCH] migrate from hbmqtt to amqtt --- bumper/mqttserver.py | 23 ++++++++------ bumper/util.py | 6 ++-- requirements.txt | 3 +- tests/__init__.py | 3 -- tests/conftest.py | 2 +- tests/test_confserver.py | 36 +++++++++++++++------ tests/test_mqttserver.py | 67 ++++++++++++++++------------------------ 7 files changed, 70 insertions(+), 70 deletions(-) diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 8b9a90f..79c2d7d 100644 --- a/bumper/mqttserver.py +++ b/bumper/mqttserver.py @@ -5,12 +5,12 @@ import json import os from typing import MutableMapping -import hbmqtt +import amqtt import pkg_resources +from amqtt.broker import Broker +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_0 from cachetools import TTLCache -from hbmqtt.broker import Broker -from hbmqtt.client import MQTTClient -from hbmqtt.mqtt.constants import QOS_0 from passlib.apps import custom_app_context as pwd_context import bumper @@ -160,12 +160,12 @@ class MQTTServer: "allow_anonymous" ] # Set to True to allow anonymous authentication - # The below adds a plugin to the hbmqtt.broker.plugins without having to futz with setup.py - distribution = pkg_resources.Distribution("hbmqtt.broker.plugins") + # The below adds a plugin to the amqtt.broker.plugins without having to futz with setup.py + distribution = pkg_resources.Distribution("amqtt.broker.plugins") bumper_plugin = pkg_resources.EntryPoint.parse( "bumper = bumper.mqttserver:BumperMQTTServer_Plugin", dist=distribution ) - distribution._ep_map = {"hbmqtt.broker.plugins": {"bumper": bumper_plugin}} + distribution._ep_map = {"amqtt.broker.plugins": {"bumper": bumper_plugin}} pkg_resources.working_set.add(distribution) # Initialize bot server @@ -187,10 +187,13 @@ class MQTTServer: "bumper" ], # Bumper plugin provides auth and handling of bots/clients connecting }, - "topic-check": {"enabled": False}, + "topic-check": { + "enabled": True, # Workaround until https://github.com/Yakifo/amqtt/pull/93 is merged + "plugins": [], + }, } - self.broker = hbmqtt.broker.Broker(config=self.default_config) + self.broker = amqtt.broker.Broker(config=self.default_config) except Exception as e: mqttserverlog.exception(f"{e}") @@ -201,7 +204,7 @@ class MQTTServer: try: await self.broker.start() - except hbmqtt.broker.BrokerException as e: + except amqtt.broker.BrokerException as e: mqttserverlog.exception(e) # asyncio.create_task(bumper.shutdown()) pass diff --git a/bumper/util.py b/bumper/util.py index 3ab30ab..5d9c0c5 100644 --- a/bumper/util.py +++ b/bumper/util.py @@ -38,8 +38,8 @@ def get_logger(name: str, rotate: RotatingFileHandler = None) -> logging.Logger: get_logger("passlib", rotate).setLevel( logging.CRITICAL + 1 ) # Ignore this logger - get_logger("hbmqtt.broker", rotate) - get_logger("hbmqtt.mqtt.protocol", rotate) - get_logger("hbmqtt.client", rotate) + get_logger("amqtt.broker", rotate) + get_logger("amqtt.mqtt.protocol", rotate) + get_logger("amqtt.client", rotate) return logger diff --git a/requirements.txt b/requirements.txt index 6ce8577..2e3cc5f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,6 @@ aiohttp==3.8.1 aiohttp-jinja2==1.5 cachetools==5.0.0 -hbmqtt==0.9.5 +amqtt==0.10.0 Jinja2==3.0.3 tinydb==4.6.1 -websockets==8.1 diff --git a/tests/__init__.py b/tests/__init__.py index 9959d47..143d88a 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,5 +1,2 @@ - HOST = "127.0.0.1" MQTT_PORT = 8883 - - diff --git a/tests/conftest.py b/tests/conftest.py index 21a09f1..0e3bd0a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -24,4 +24,4 @@ async def conf_server_client(aiohttp_client): yield client - await client.close() \ No newline at end of file + await client.close() diff --git a/tests/test_confserver.py b/tests/test_confserver.py index 7dfe287..58409dc 100644 --- a/tests/test_confserver.py +++ b/tests/test_confserver.py @@ -105,7 +105,7 @@ async def test_base(conf_server_client): resp = await conf_server_client.get("/") assert resp.status == 200 - mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.Client.disconnect() bumper.xmpp_server.disconnect() @@ -135,7 +135,7 @@ async def test_restartService(conf_server_client): resp = await conf_server_client.get("/restart_XMPPServer") assert resp.status == 200 - mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.Client.disconnect() xmpp_server.disconnect() @@ -155,7 +155,9 @@ async def test_login(conf_server_client): bumper.db = "tests/tmp.db" # Set db location for testing # Test without user - resp = await conf_server_client.get("/v1/private/us/en/dev_1234/ios/1/0/0/user/login") + resp = await conf_server_client.get( + "/v1/private/us/en/dev_1234/ios/1/0/0/user/login" + ) assert resp.status == 200 text = await resp.text() jsonresp = json.loads(text) @@ -168,7 +170,9 @@ async def test_login(conf_server_client): bumper.db = "tests/tmp.db" # Set db location for testing # Test global_e without user - resp = await conf_server_client.get("/v1/private/us/en/dev_1234/global_e/1/0/0/user/login") + resp = await conf_server_client.get( + "/v1/private/us/en/dev_1234/global_e/1/0/0/user/login" + ) assert resp.status == 200 text = await resp.text() jsonresp = json.loads(text) @@ -179,7 +183,9 @@ async def test_login(conf_server_client): # Add a user to db and test with existing users bumper.user_add("testuser") - resp = await conf_server_client.get("/v1/private/us/en/dev_1234/ios/1/0/0/user/login") + resp = await conf_server_client.get( + "/v1/private/us/en/dev_1234/ios/1/0/0/user/login" + ) assert resp.status == 200 text = await resp.text() jsonresp = json.loads(text) @@ -190,7 +196,9 @@ async def test_login(conf_server_client): # Add a bot to db that will be added to user bumper.bot_add("sn_123", "did_123", "dev_123", "res_123", "com_123") - resp = await conf_server_client.get("/v1/private/us/en/dev_1234/ios/1/0/0/user/login") + resp = await conf_server_client.get( + "/v1/private/us/en/dev_1234/ios/1/0/0/user/login" + ) assert resp.status == 200 text = await resp.text() jsonresp = json.loads(text) @@ -209,7 +217,9 @@ async def test_login(conf_server_client): } bumper.bot_full_upsert(newbot) - resp = await conf_server_client.get("/v1/private/us/en/dev_1234/ios/1/0/0/user/login") + resp = await conf_server_client.get( + "/v1/private/us/en/dev_1234/ios/1/0/0/user/login" + ) assert resp.status == 200 text = await resp.text() jsonresp = json.loads(text) @@ -386,7 +396,9 @@ async def test_checkAgreement(conf_server_client): remove_existing_db() bumper.db = "tests/tmp.db" # Set db location for testing - resp = await conf_server_client.get("/v1/private/us/en/dev_1234/ios/1/0/0/user/checkAgreement") + resp = await conf_server_client.get( + "/v1/private/us/en/dev_1234/ios/1/0/0/user/checkAgreement" + ) assert resp.status == 200 text = await resp.text() jsonresp = json.loads(text) @@ -419,7 +431,9 @@ async def test_checkVersion(conf_server_client): remove_existing_db() bumper.db = "tests/tmp.db" # Set db location for testing - resp = await conf_server_client.get("/v1/private/us/en/dev_1234/ios/1/0/0/common/checkVersion") + resp = await conf_server_client.get( + "/v1/private/us/en/dev_1234/ios/1/0/0/common/checkVersion" + ) assert resp.status == 200 text = await resp.text() jsonresp = json.loads(text) @@ -543,7 +557,9 @@ async def test_neng_hasUnreadMessage(conf_server_client): }, "count": 20, } - resp = await conf_server_client.post("/api/neng/message/hasUnreadMsg", json=postbody) + resp = await conf_server_client.post( + "/api/neng/message/hasUnreadMsg", json=postbody + ) assert resp.status == 200 text = await resp.text() jsonresp = json.loads(text) diff --git a/tests/test_mqttserver.py b/tests/test_mqttserver.py index 5157d3c..42b13cd 100644 --- a/tests/test_mqttserver.py +++ b/tests/test_mqttserver.py @@ -2,8 +2,9 @@ import asyncio import os import time -import hbmqtt +import amqtt import pytest +from amqtt.mqtt.constants import QOS_0 from testfixtures import LogCapture import bumper @@ -22,11 +23,9 @@ async def test_helperbot_message(): ) # Check helperbot is connected msg_payload = "" msg_topic_name = "iot/atr/DustCaseST/bot_serial/ls1ok3/wC3g/x" - await mqtt_helperbot.Client.publish( - msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0 - ) + await mqtt_helperbot.Client.publish(msg_topic_name, msg_payload.encode(), QOS_0) - await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + await asyncio.sleep(0.1) l.check_present( ( @@ -36,7 +35,7 @@ async def test_helperbot_message(): ) ) # Check broadcast message was logged l.clear() - mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.Client.disconnect() # Send command to bot mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) @@ -46,11 +45,9 @@ async def test_helperbot_message(): ) # Check helperbot is connected msg_payload = "{}" msg_topic_name = "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 - ) + await mqtt_helperbot.Client.publish(msg_topic_name, msg_payload.encode(), QOS_0) - await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + await asyncio.sleep(0.1) l.check_present( ( @@ -60,7 +57,7 @@ async def test_helperbot_message(): ) ) # Check send command message was logged l.clear() - mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.Client.disconnect() # Received response to command mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) @@ -70,11 +67,9 @@ 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/helperbot/bumper/helperbot/p/iCmuqp/j" - await mqtt_helperbot.Client.publish( - msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0 - ) + await mqtt_helperbot.Client.publish(msg_topic_name, msg_payload.encode(), QOS_0) - await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + await asyncio.sleep(0.1) l.check_present( ( @@ -84,7 +79,7 @@ async def test_helperbot_message(): ) ) # Check received response message was logged l.clear() - mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.Client.disconnect() # Received unknown message mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) @@ -94,11 +89,9 @@ async def test_helperbot_message(): ) # Check helperbot is connected msg_payload = "test" msg_topic_name = "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 - ) + await mqtt_helperbot.Client.publish(msg_topic_name, msg_payload.encode(), QOS_0) - await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + await asyncio.sleep(0.1) l.check_present( ( @@ -108,7 +101,7 @@ async def test_helperbot_message(): ) ) # Check received message was logged l.clear() - mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.Client.disconnect() # Received error message mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) @@ -118,11 +111,9 @@ async def test_helperbot_message(): ) # Check helperbot is connected msg_payload = "" msg_topic_name = "iot/atr/errors/bot_serial/ls1ok3/wC3g/x" - await mqtt_helperbot.Client.publish( - msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0 - ) + await mqtt_helperbot.Client.publish(msg_topic_name, msg_payload.encode(), QOS_0) - await asyncio.wait_for(mqtt_helperbot.Client.deliver_message(), timeout=0.1) + await asyncio.sleep(0.1) l.check_present( ( @@ -132,7 +123,7 @@ async def test_helperbot_message(): ) ) # Check received message was logged l.clear() - mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.Client.disconnect() @pytest.mark.usefixtures("mqtt_server") @@ -164,7 +155,7 @@ async def test_helperbot_expire_message(): msg_payload = "" msg_topic_name = "iot/atr/DustCaseST/bot_serial/ls1ok3/wC3g/x" await mqtt_helperbot.Client.publish( - msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0 + msg_topic_name, msg_payload.encode(), QOS_0 ) # Send another message to force get_msg await asyncio.sleep(timeout * 2) @@ -212,9 +203,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/helperbot/bumper/helperbot/p/testgood/j" - await mqtt_helperbot.Client.publish( - msg_topic_name, msg_payload.encode(), hbmqtt.client.QOS_0 - ) + await mqtt_helperbot.Client.publish(msg_topic_name, msg_payload.encode(), QOS_0) commandresult = await mqtt_helperbot.send_command(cmdjson, "testgood") assert commandresult == { @@ -223,7 +212,7 @@ async def test_helperbot_sendcommand(): "ret": "ok", } - # mqtt_helperbot.Client.disconnect() + # await mqtt_helperbot.Client.disconnect() # Test GetLifeSpan (xml command) cmdjson = { @@ -246,9 +235,7 @@ async def test_helperbot_sendcommand(): # Send response beforehand msg_payload = "" msg_topic_name = "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 - ) + await mqtt_helperbot.Client.publish(msg_topic_name, msg_payload.encode(), QOS_0) commandresult = await mqtt_helperbot.send_command(cmdjson, "testx") assert commandresult == { @@ -283,9 +270,7 @@ async def test_helperbot_sendcommand(): msg_topic_name = ( "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 - ) + await mqtt_helperbot.Client.publish(msg_topic_name, msg_payload.encode(), QOS_0) commandresult = await mqtt_helperbot.send_command(cmdjson, "testj") @@ -315,7 +300,7 @@ async def test_helperbot_sendcommand(): "ret": "ok", } - mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.Client.disconnect() async def test_mqttserver(): @@ -345,7 +330,7 @@ async def test_mqttserver(): test_client = bumper.MQTTHelperBot(HOST, MQTT_PORT) test_client.client_id = "user_123@ecouser.net/resource_123" # await test_client.start_helper_bot() - test_client.Client = hbmqtt.client.MQTTClient( + test_client.Client = amqtt.client.MQTTClient( client_id=test_client.client_id, config={"check_hostname": False} ) @@ -374,7 +359,7 @@ async def test_mqttserver(): test_client = bumper.MQTTHelperBot(HOST, MQTT_PORT) test_client.client_id = "test-file-auth" # await test_client.start_helper_bot() - test_client.Client = hbmqtt.client.MQTTClient( + test_client.Client = amqtt.client.MQTTClient( client_id=test_client.client_id, config={ "check_hostname": False, @@ -445,7 +430,7 @@ async def test_nofileauth_mqttserver(): l.check_present( ( - "hbmqtt.broker.plugins.bumper", + "amqtt.broker.plugins.bumper", "WARNING", "Password file tests/passwd-notfound not found", ),