amqtt.client is not working correctly. Using gmqtt instead

bumper/__init__.py

@@ -5,7 +5,8 @@ import socket
 import sys
 
 from bumper.db import revoke_expired_oauths, revoke_expired_tokens
-from bumper.mqttserver import MQTTHelperBot, MQTTServer
+from bumper.mqtt.helper_bot import HelperBot
+from bumper.mqttserver import MQTTServer
 from bumper.util import get_logger, log_to_stdout
 from bumper.web.server import WebServer, WebserverBinding
 from bumper.xmppserver import XMPPServer
@@ -50,7 +51,7 @@ token_validity_seconds = 3600  # 1 hour
 oauth_validity_days = 15
 
 mqtt_server: MQTTServer
-mqtt_helperbot: MQTTHelperBot
+mqtt_helperbot: HelperBot
 web_server: WebServer
 xmpp_server: XMPPServer
 
@@ -103,7 +104,7 @@ async def start():
     global mqtt_server
     mqtt_server = MQTTServer(bumper_listen, mqtt_listen_port)
     global mqtt_helperbot
-    mqtt_helperbot = MQTTHelperBot(bumper_listen, mqtt_listen_port)
+    mqtt_helperbot = HelperBot(bumper_listen, mqtt_listen_port)
     global web_server
     web_server = WebServer(web_server_bindings)
     global xmpp_server

bumper/mqtt/__init__.py

@@ -0,0 +1 @@
+"""Mqtt module."""

bumper/mqtt/helper_bot.py

@@ -0,0 +1,149 @@
+"""Helper bot module."""
+import asyncio
+import json
+import ssl
+from typing import Any, MutableMapping, Union
+
+from cachetools import TTLCache
+from gmqtt import Client, Subscription
+from gmqtt.mqtt.constants import MQTTv311
+
+from bumper.util import get_logger
+
+_LOGGER = get_logger("helperbot")
+
+
+class CommandDto:
+    """Command DTO."""
+
+    def __init__(self, payload_type: str) -> None:
+        self._payload_type = payload_type
+        self._event = asyncio.Event()
+        self._response: Union[str, bytes]
+
+    async def wait_for_response(self) -> Union[str, dict[str, Any]]:
+        """Wait for the response to be received."""
+        await self._event.wait()
+        if self._payload_type == "j":
+            return json.loads(self._response)  # type:ignore[no-any-return]
+
+        return str(self._response)
+
+    def add_response(self, response: Union[str, bytes]) -> None:
+        """Add received response."""
+        self._response = response
+        self._event.set()
+
+
+class HelperBot:
+    """Helper bot, which converts commands from the rest api to mqtt ones."""
+
+    def __init__(self, host: str, port: int, timeout: float = 60):
+        self._commands: MutableMapping[str, CommandDto] = TTLCache(
+            maxsize=timeout * 60, ttl=timeout * 1.1
+        )
+        self._host = host
+        self._port = port
+        self._client_id = "helperbot@bumper/helperbot"
+        self._timeout = timeout
+        self._client = Client("helperbot@bumper/helperbot")
+
+        # pylint: disable=unused-argument
+        async def _on_message(
+            client: Client, topic: str, payload: bytes, qos: int, properties: dict
+        ) -> None:
+            _LOGGER.debug("Got message: topic=%s; payload=%s;", topic, payload.decode())
+            topic_split = topic.split("/")
+            data_decoded = str(payload.decode())
+            if topic_split[10] in self._commands:
+                self._commands[topic_split[10]].add_response(data_decoded)
+
+        self._client.on_message = _on_message
+
+    @property
+    def is_connected(self) -> bool:
+        """Return True if client is connected successfully."""
+        return self._client.is_connected  # type: ignore[no-any-return]
+
+    async def start(self) -> None:
+        """Connect and subscribe helper bot."""
+        try:
+            if self.is_connected:
+                return
+
+            ssl_ctx = ssl.create_default_context()
+            ssl_ctx.check_hostname = False
+            ssl_ctx.verify_mode = ssl.CERT_NONE
+            await self._client.connect(
+                self._host, self._port, ssl=ssl_ctx, version=MQTTv311
+            )
+            self._client.subscribe(
+                Subscription("iot/p2p/+/+/+/+/helperbot/bumper/helperbot/+/+/+")
+            )
+        except Exception:
+            _LOGGER.exception("An exception occurred during startup", exc_info=True)
+            raise
+
+    async def _wait_for_resp(
+        self, command_dto: CommandDto, request_id: str
+    ) -> dict[str, Any]:
+        try:
+            payload = await asyncio.wait_for(
+                command_dto.wait_for_response(), timeout=self._timeout
+            )
+            return {"id": request_id, "ret": "ok", "resp": payload}
+        except asyncio.TimeoutError:
+            _LOGGER.debug("wait_for_resp timeout reached")
+        except asyncio.CancelledError:
+            _LOGGER.debug("wait_for_resp cancelled by asyncio", exc_info=True)
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("An unknown error occurred", exc_info=True)
+
+        return {
+            "id": request_id,
+            "errno": 500,
+            "ret": "fail",
+            "debug": "wait for response timed out",
+        }
+
+    async def send_command(
+        self, cmdjson: dict[str, Any], request_id: str
+    ) -> dict[str, Any]:
+        """Send command over MQTT."""
+        if not self.is_connected:
+            await self.start()
+
+        try:
+            topic = (
+                f"iot/p2p/{cmdjson['cmdName']}/helperbot/bumper/helperbot/{cmdjson['toId']}/"
+                f"{cmdjson['toType']}/{cmdjson['toRes']}/q/{request_id}/{cmdjson['payloadType']}"
+            )
+
+            if cmdjson["payloadType"] == "j":
+                payload = json.dumps(cmdjson["payload"])
+            else:
+                payload = str(cmdjson["payload"])
+
+            command_dto = CommandDto(cmdjson["payloadType"])
+            self._commands[request_id] = command_dto
+
+            _LOGGER.debug("Sending message %s", topic)
+            self._client.publish(topic, payload.encode())
+
+            resp = await self._wait_for_resp(command_dto, request_id)
+            return resp
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Could not send command.", exc_info=True)
+            return {
+                "id": request_id,
+                "errno": 500,
+                "ret": "fail",
+                "debug": "exception occurred please check bumper logs",
+            }
+        finally:
+            self._commands.pop(request_id, None)
+
+    async def disconnect(self) -> None:
+        """Disconnect client."""
+        if self.is_connected:
+            await self._client.disconnect()

bumper/mqttserver.py

@@ -1,18 +1,12 @@
 """Mqtt module."""
 
-import asyncio
-import json
 import os
-from asyncio import Task
+from typing import Any
-from typing import Any, MutableMapping, Optional, Union
 
 import amqtt
 import pkg_resources
 from amqtt.broker import Broker, BrokerContext
-from amqtt.client import MQTTClient
-from amqtt.mqtt.constants import QOS_0
 from amqtt.session import IncomingApplicationMessage, Session
-from cachetools import TTLCache
 from passlib.apps import custom_app_context as pwd_context
 
 import bumper
@@ -32,164 +26,6 @@ helperbotlog = get_logger("helperbot")
 boterrorlog = get_logger("boterror")
 
 
-class CommandDto:
-    """Command DTO."""
-
-    def __init__(self, payload_type: str) -> None:
-        self._payload_type = payload_type
-        self._event = asyncio.Event()
-        self._response: Union[str, bytes]
-
-    async def wait_for_response(self) -> Union[str, dict[str, Any]]:
-        """Wait for the response to be received."""
-        await self._event.wait()
-        if self._payload_type == "j":
-            return json.loads(self._response)  # type:ignore[no-any-return]
-
-        return str(self._response)
-
-    def add_response(self, response: Union[str, bytes]) -> None:
-        """Add received response."""
-        self._response = response
-        self._event.set()
-
-
-class MQTTHelperBot:
-    """Helper bot, which converts commands from the rest api to mqtt ones."""
-
-    def __init__(self, host: str, port: int, timeout: float = 60):
-        self._commands: MutableMapping[str, CommandDto] = TTLCache(
-            maxsize=timeout * 60, ttl=timeout * 1.1
-        )
-        self._host = host
-        self._port = port
-        self._client_id = "helperbot@bumper/helperbot"
-        self._timeout = timeout
-        self._client: Optional[MQTTClient] = None
-        self._new_messages_task: Optional[Task] = None
-
-    @property
-    def is_connected(self) -> bool:
-        """Return True if client is connected successfully."""
-        return (
-            self._client is not None
-            and self._client.session.transitions.state == "connected"
-        )
-
-    async def start(self) -> None:
-        """Connect and subscribe helper bot."""
-        try:
-            if self._client is None:
-                self._client = MQTTClient(
-                    client_id=self._client_id,
-                    config={"check_hostname": False, "reconnect_retries": 20},
-                )
-
-            await self._client.connect(
-                f"mqtts://{self._host}:{self._port}/", cafile=bumper.ca_cert
-            )
-            await self._client.subscribe(
-                [
-                    ("iot/p2p/+/+/+/+/helperbot/bumper/helperbot/+/+/+", QOS_0),
-                ]
-            )
-            self._new_messages_task = asyncio.create_task(
-                self._check_for_new_messages()
-            )
-        except Exception:
-            mqttserverlog.exception(
-                "An exception occurred during startup", exc_info=True
-            )
-            raise
-
-    async def _check_for_new_messages(self) -> None:
-        assert self._client is not None
-        while True:
-            try:
-                message: Optional[
-                    IncomingApplicationMessage
-                ] = await self._client.deliver_message()
-                if message is not None:
-                    topic_split = str(message.topic).split("/")
-                    data_decoded = str(message.data.decode("utf-8"))
-                    if topic_split[10] in self._commands:
-                        self._commands[topic_split[10]].add_response(data_decoded)
-            except asyncio.CancelledError:
-                pass
-            except Exception:  # pylint: disable=broad-except
-                helperbotlog.error(
-                    "An exception occurred during handling new messages", exc_info=True
-                )
-
-    async def _wait_for_resp(
-        self, command_dto: CommandDto, request_id: str
-    ) -> dict[str, Any]:
-        try:
-            payload = await asyncio.wait_for(
-                command_dto.wait_for_response(), timeout=self._timeout
-            )
-            return {"id": request_id, "ret": "ok", "resp": payload}
-        except asyncio.TimeoutError:
-            helperbotlog.debug("wait_for_resp timeout reached")
-        except asyncio.CancelledError:
-            helperbotlog.debug("wait_for_resp cancelled by asyncio", exc_info=True)
-        except Exception:  # pylint: disable=broad-except
-            helperbotlog.exception("An unknown error occurred", exc_info=True)
-
-        return {
-            "id": request_id,
-            "errno": 500,
-            "ret": "fail",
-            "debug": "wait for response timed out",
-        }
-
-    async def send_command(
-        self, cmdjson: dict[str, Any], request_id: str
-    ) -> dict[str, Any]:
-        """Send command over MQTT."""
-        if self._client is None:
-            await self.start()
-            assert self._client is not None
-
-        try:
-            topic = (
-                f"iot/p2p/{cmdjson['cmdName']}/helperbot/bumper/helperbot/{cmdjson['toId']}/"
-                f"{cmdjson['toType']}/{cmdjson['toRes']}/q/{request_id}/{cmdjson['payloadType']}"
-            )
-            command_dto = CommandDto(cmdjson["payloadType"])
-            self._commands[request_id] = command_dto
-
-            if cmdjson["payloadType"] == "j":
-                payload = json.dumps(cmdjson["payload"])
-            else:
-                payload = str(cmdjson["payload"])
-
-            await self._client.publish(topic, payload.encode(), QOS_0)
-
-            resp = await self._wait_for_resp(command_dto, request_id)
-            return resp
-        except Exception:  # pylint: disable=broad-except
-            helperbotlog.exception("Could not send command.", exc_info=True)
-            return {
-                "id": request_id,
-                "errno": 500,
-                "ret": "fail",
-                "debug": "exception occurred please check bumper logs",
-            }
-        finally:
-            self._commands.pop(request_id, None)
-
-    async def disconnect(self) -> None:
-        """Disconnect client."""
-        if self._new_messages_task is not None:
-            self._new_messages_task.cancel()
-            self._new_messages_task = None
-
-        if self._client is not None:
-            await self._client.disconnect()
-            self._client = None
-
-
 class MQTTServer:
     """Mqtt server."""
 

bumper/util.py

@@ -41,7 +41,7 @@ def get_logger(name: str, rotate: RotatingFileHandler = None) -> logging.Logger:
         )  # Ignore this logger
         get_logger("amqtt.broker", rotate)
         get_logger("amqtt.mqtt.protocol", rotate)
-        get_logger("amqtt.client", rotate)
+        get_logger("gmqtt", rotate)
 
     return logger
 

requirements.txt

@@ -2,5 +2,6 @@ aiohttp==3.8.1
 aiohttp-jinja2==1.5
 amqtt==0.10.0
 cachetools==5.0.0
+gmqtt==0.6.11
 Jinja2==3.0.3
 tinydb==4.6.1

tests/conftest.py

@@ -1,10 +1,12 @@
 import asyncio
+import ssl
 
 import pytest
-from amqtt.client import MQTTClient
+from gmqtt import Client
+from gmqtt.mqtt.constants import MQTTv311
 
 import bumper
-from bumper import MQTTServer, WebserverBinding
+from bumper import HelperBot, MQTTServer, WebserverBinding
 from tests import HOST, MQTT_PORT, WEBSERVER_PORT
 
 
@@ -16,26 +18,40 @@ async def mqtt_server():
     while not mqtt_server.state == "started":
         await asyncio.sleep(0.1)
 
-    yield
+    yield mqtt_server
 
     await mqtt_server.shutdown()
 
 
-@pytest.mark.usefixtures("mqtt_server")
 @pytest.fixture
-async def mqtt_client():
+async def mqtt_client(mqtt_server: MQTTServer):
-    client = MQTTClient(
+    assert mqtt_server.state == "started"
-        client_id="helperbot@bumper/test",
-        config={"check_hostname": False, "auto_reconnect": False},
-    )
 
-    await client.connect(f"mqtts://{HOST}:{MQTT_PORT}/", cafile=bumper.ca_cert)
+    client = Client("helperbot@bumper/test")
+    ssl_ctx = ssl.create_default_context()
+    ssl_ctx.check_hostname = False
+    ssl_ctx.verify_mode = ssl.CERT_NONE
+    await client.connect(HOST, MQTT_PORT, ssl=ssl_ctx, version=MQTTv311)
 
     yield client
 
     await client.disconnect()
 
 
+@pytest.fixture
+async def helper_bot(mqtt_server: MQTTServer):
+    assert mqtt_server.state == "started"
+
+    helper_bot = HelperBot(HOST, MQTT_PORT, 0.1)
+    bumper.mqtt_helperbot = helper_bot
+    await helper_bot.start()
+    assert helper_bot.is_connected
+
+    yield helper_bot
+
+    await helper_bot.disconnect()
+
+
 @pytest.fixture
 async def webserver_client(aiohttp_client):
     webserver = bumper.WebServer(WebserverBinding(HOST, WEBSERVER_PORT, False))

tests/test_mqttserver.py

@@ -1,28 +1,27 @@
 import asyncio
 import os
+import ssl
 import time
 
-import pytest
+from gmqtt import Client
-from amqtt.client import MQTTClient
+from gmqtt.mqtt.constants import MQTTv311
-from amqtt.mqtt.constants import QOS_0
 from testfixtures import LogCapture
 
-import bumper
+from bumper import MQTTServer, db
-from bumper import MQTTHelperBot, MQTTServer, ca_cert, db
+from bumper.mqtt.helper_bot import HelperBot
 from tests import HOST, MQTT_PORT
 
 
-@pytest.mark.usefixtures("mqtt_server")
+async def test_helperbot_message(mqtt_client: Client):
-async def test_helperbot_message(mqtt_client: MQTTClient):
     with LogCapture() as l:
 
         # Test broadcast message
-        mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
+        mqtt_helperbot = HelperBot(HOST, MQTT_PORT)
         await mqtt_helperbot.start()
         assert mqtt_helperbot.is_connected
         msg_payload = "<ctl ts='1547822804960' td='DustCaseST' st='0'/>"
         msg_topic_name = "iot/atr/DustCaseST/bot_serial/ls1ok3/wC3g/x"
-        await mqtt_client.publish(msg_topic_name, msg_payload.encode(), QOS_0)
+        mqtt_client.publish(msg_topic_name, msg_payload.encode())
 
         await asyncio.sleep(0.1)
 
@@ -37,12 +36,12 @@ async def test_helperbot_message(mqtt_client: MQTTClient):
         await mqtt_helperbot.disconnect()
 
         # Send command to bot
-        mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
+        mqtt_helperbot = HelperBot(HOST, MQTT_PORT)
         await mqtt_helperbot.start()
         assert mqtt_helperbot.is_connected
         msg_payload = "{}"
         msg_topic_name = "iot/p2p/GetWKVer/helperbot/bumper/helperbot/bot_serial/ls1ok3/wC3g/q/iCmuqp/j"
-        await mqtt_client.publish(msg_topic_name, msg_payload.encode(), QOS_0)
+        mqtt_client.publish(msg_topic_name, msg_payload.encode())
 
         await asyncio.sleep(0.1)
 
@@ -57,12 +56,12 @@ async def test_helperbot_message(mqtt_client: MQTTClient):
         await mqtt_helperbot.disconnect()
 
         # Received response to command
-        mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
+        mqtt_helperbot = HelperBot(HOST, MQTT_PORT)
         await mqtt_helperbot.start()
         assert mqtt_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_client.publish(msg_topic_name, msg_payload.encode(), QOS_0)
+        mqtt_client.publish(msg_topic_name, msg_payload.encode())
 
         await asyncio.sleep(0.1)
 
@@ -77,12 +76,12 @@ async def test_helperbot_message(mqtt_client: MQTTClient):
         await mqtt_helperbot.disconnect()
 
         # Received unknown message
-        mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
+        mqtt_helperbot = HelperBot(HOST, MQTT_PORT)
         await mqtt_helperbot.start()
         assert mqtt_helperbot.is_connected
         msg_payload = "test"
         msg_topic_name = "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/TESTBAD/bumper/helperbot/p/iCmuqp/j"
-        await mqtt_client.publish(msg_topic_name, msg_payload.encode(), QOS_0)
+        mqtt_client.publish(msg_topic_name, msg_payload.encode())
 
         await asyncio.sleep(0.1)
 
@@ -97,12 +96,12 @@ async def test_helperbot_message(mqtt_client: MQTTClient):
         await mqtt_helperbot.disconnect()
 
         # Received error message
-        mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
+        mqtt_helperbot = HelperBot(HOST, MQTT_PORT)
         await mqtt_helperbot.start()
         assert mqtt_helperbot.is_connected
         msg_payload = "<ctl ts='1560904925396' td='errors' old='' new='110'/>"
         msg_topic_name = "iot/atr/errors/bot_serial/ls1ok3/wC3g/x"
-        await mqtt_client.publish(msg_topic_name, msg_payload.encode(), QOS_0)
+        mqtt_client.publish(msg_topic_name, msg_payload.encode())
 
         await asyncio.sleep(0.1)
 
@@ -117,15 +116,7 @@ async def test_helperbot_message(mqtt_client: MQTTClient):
         await mqtt_helperbot.disconnect()
 
 
-@pytest.mark.usefixtures("mqtt_server")
+async def test_helperbot_expire_message(mqtt_client: Client, helper_bot: HelperBot):
-async def test_helperbot_expire_message(mqtt_client: MQTTClient):
-    timeout = 0.1
-    # Test broadcast message
-    mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT, timeout)
-    bumper.mqtt_helperbot = mqtt_helperbot
-    await mqtt_helperbot.start()
-    assert mqtt_helperbot.is_connected
-
     expire_msg_payload = '{"ret":"ok","ver":"0.13.5"}'
     expire_msg_topic_name = "iot/p2p/GetWKVer/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/testgood/j"
     currenttime = time.time()
@@ -136,32 +127,23 @@ async def test_helperbot_expire_message(mqtt_client: MQTTClient):
         "payload": expire_msg_payload,
     }
 
-    mqtt_helperbot._commands[request_id] = data
+    helper_bot._commands[request_id] = data
 
-    assert mqtt_helperbot._commands[request_id] == data
+    assert helper_bot._commands[request_id] == data
 
     await asyncio.sleep(0.1)
     msg_payload = "<ctl ts='1547822804960' td='DustCaseST' st='0'/>"
     msg_topic_name = "iot/atr/DustCaseST/bot_serial/ls1ok3/wC3g/x"
-    await mqtt_client.publish(
+    mqtt_client.publish(
-        msg_topic_name, msg_payload.encode(), QOS_0
+        msg_topic_name, msg_payload.encode()
     )  # Send another message to force get_msg
 
-    await asyncio.sleep(timeout * 2)
+    await asyncio.sleep(0.1 * 2)
 
-    assert mqtt_helperbot._commands.get(request_id, None) == None
+    assert helper_bot._commands.get(request_id, None) == None
-
-    await mqtt_helperbot.disconnect()
 
 
-@pytest.mark.usefixtures("mqtt_server")
+async def test_helperbot_sendcommand(mqtt_client: Client, helper_bot: HelperBot):
-async def test_helperbot_sendcommand(mqtt_client: MQTTClient):
-    timeout = 0.1
-    mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT, timeout)
-    bumper.mqtt_helperbot = mqtt_helperbot
-    await mqtt_helperbot.start()
-    assert mqtt_helperbot.is_connected
-
     cmdjson = {
         "toType": "ls1ok3",
         "payloadType": "j",
@@ -178,7 +160,7 @@ async def test_helperbot_sendcommand(mqtt_client: MQTTClient):
             "realm": "ecouser.net",
         },
     }
-    commandresult = await mqtt_helperbot.send_command(cmdjson, "testfail")
+    commandresult = await helper_bot.send_command(cmdjson, "testfail")
     # Don't send a response, ensure timeout
     assert commandresult == {
         "debug": "wait for response timed out",
@@ -190,9 +172,10 @@ async def test_helperbot_sendcommand(mqtt_client: MQTTClient):
     # 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_client.publish(msg_topic_name, msg_payload.encode(), QOS_0)
+    loop = asyncio.get_event_loop()
+    loop.call_soon(lambda: mqtt_client.publish(msg_topic_name, msg_payload.encode()))
 
-    commandresult = await mqtt_helperbot.send_command(cmdjson, "testgood")
+    commandresult = await helper_bot.send_command(cmdjson, "testgood")
     assert commandresult == {
         "id": "testgood",
         "resp": {"ret": "ok", "ver": "0.13.5"},
@@ -222,9 +205,9 @@ async def test_helperbot_sendcommand(mqtt_client: MQTTClient):
     # Send response beforehand
     msg_payload = "<ctl ret='ok' type='Brush' left='4142' total='18000'/>"
     msg_topic_name = "iot/p2p/GetLifeSpan/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/testx/q"
-    await mqtt_client.publish(msg_topic_name, msg_payload.encode(), QOS_0)
+    mqtt_client.publish(msg_topic_name, msg_payload.encode())
 
-    commandresult = await mqtt_helperbot.send_command(cmdjson, "testx")
+    commandresult = await helper_bot.send_command(cmdjson, "testx")
     assert commandresult == {
         "id": "testx",
         "resp": "<ctl ret='ok' type='Brush' left='4142' total='18000'/>",
@@ -257,9 +240,9 @@ async def test_helperbot_sendcommand(mqtt_client: MQTTClient):
     msg_topic_name = (
         "iot/p2p/getStats/bot_serial/ls1ok3/wC3g/helperbot/bumper/helperbot/p/testj/j"
     )
-    await mqtt_client.publish(msg_topic_name, msg_payload.encode(), QOS_0)
+    mqtt_client.publish(msg_topic_name, msg_payload.encode())
 
-    commandresult = await mqtt_helperbot.send_command(cmdjson, "testj")
+    commandresult = await helper_bot.send_command(cmdjson, "testj")
 
     assert commandresult == {
         "id": "testj",
@@ -287,8 +270,6 @@ async def test_helperbot_sendcommand(mqtt_client: MQTTClient):
         "ret": "ok",
     }
 
-    await mqtt_helperbot.disconnect()
-
 
 async def test_mqttserver():
     if os.path.exists("tests/tmp.db"):
@@ -302,7 +283,7 @@ async def test_mqttserver():
 
     try:
         # Test helperbot connect
-        mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
+        mqtt_helperbot = HelperBot(HOST, MQTT_PORT)
         await mqtt_helperbot.start()
         assert mqtt_helperbot.is_connected
         await mqtt_helperbot.disconnect()
@@ -311,60 +292,35 @@ async def test_mqttserver():
         db.user_add("user_123")  # Add user to db
         db.client_add("user_123", "ecouser.net", "resource_123")  # Add client to db
 
-        client = MQTTClient(
+        ssl_ctx = ssl.create_default_context()
-            client_id="user_123@ecouser.net/resource_123",
+        ssl_ctx.check_hostname = False
-            config={"check_hostname": False},
+        ssl_ctx.verify_mode = ssl.CERT_NONE
-        )
+        client = Client("user_123@ecouser.net/resource_123")
-
+        await client.connect(HOST, MQTT_PORT, ssl=ssl_ctx, version=MQTTv311)
-        await client.connect(
+        assert client.is_connected
-            f"mqtts://{HOST}:{MQTT_PORT}/",
-            cafile=ca_cert,
-        )
-        assert client._connected_state._value == True  # Check client is connected
         await client.disconnect()
-        assert client._connected_state._value == False  # Check client is disconnected
+        assert not client.is_connected
 
         # Test fake_bot connect
-        client = MQTTClient(
+        client = Client("bot_serial@ls1ok3/wC3g")
-            client_id="bot_serial@ls1ok3/wC3g", config={"check_hostname": False}
+        await client.connect(HOST, MQTT_PORT, ssl=ssl_ctx, version=MQTTv311)
-        )
+        assert client.is_connected
-
-        await client.connect(
-            f"mqtts://{HOST}:{MQTT_PORT}/",
-            cafile=ca_cert,
-        )
-        assert client._connected_state._value == True  # Check fake_bot is connected
         await client.disconnect()
 
         # Test file auth client connect
-        client = MQTTClient(
+        client = Client("test-file-auth")
-            client_id="test-file-auth",
+        client.set_auth_credentials("test-client", "abc123!")
-            config={
+        await client.connect(HOST, MQTT_PORT, ssl=ssl_ctx, version=MQTTv311)
-                "check_hostname": False,
+        assert client.is_connected
-                "auto_reconnect": False,
-                "reconnect_retries": 1,
-            },
-        )
-
-        # good user/pass
-        await client.connect(
-            f"mqtts://test-client:abc123!@{HOST}:{MQTT_PORT}/",
-            cafile=ca_cert,
-            cleansession=True,
-        )
-
-        assert client._connected_state._value == True  # Check client is connected
         await client.disconnect()
-        assert client._connected_state._value == False  # Check client is disconnected
+        assert not client.is_connected
 
         # bad password
         with LogCapture() as l:
 
-            await client.connect(
+            client.set_auth_credentials("test-client", "notvalid!")
-                f"mqtts://test-client:notvalid!@{HOST}:{MQTT_PORT}/",
+            await client.connect(HOST, MQTT_PORT, ssl=ssl_ctx, version=MQTTv311)
-                cafile=ca_cert,
+            await client.disconnect()
-                cleansession=True,
-            )
 
             l.check_present(
                 (
@@ -377,11 +333,9 @@ async def test_mqttserver():
             l.clear()
 
             # no username in file
-            await client.connect(
+            client.set_auth_credentials("test-client-noexist", "notvalid!")
-                f"mqtts://test-client-noexist:notvalid!@{HOST}:{MQTT_PORT}/",
+            await client.connect(HOST, MQTT_PORT, ssl=ssl_ctx, version=MQTTv311)
-                cafile=ca_cert,
+            await client.disconnect()
-                cleansession=True,
-            )
 
             l.check_present(
                 (

tests/web/test_server.py

@@ -6,7 +6,7 @@ from unittest import mock
 import pytest
 
 import bumper
-from bumper import MQTTHelperBot, WebServer, WebserverBinding, XMPPServer, db
+from bumper import HelperBot, WebServer, WebserverBinding, XMPPServer, db
 from bumper.models import ERR_TOKEN_INVALID, RETURN_API_SUCCESS
 from tests import HOST, MQTT_PORT, WEBSERVER_PORT
 
@@ -36,7 +36,7 @@ async def test_webserver_no_ssl():
     await webserver.start()
 
 
-@pytest.mark.usefixtures("mqtt_server")
+@pytest.mark.usefixtures("helper_bot")
 async def test_base(webserver_client):
     remove_existing_db()
 
@@ -46,20 +46,13 @@ async def test_base(webserver_client):
     bumper.xmpp_server = xmpp_server
     await xmpp_server.start_async_server()
 
-    # Start Helperbot
-    mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
-    bumper.mqtt_helperbot = mqtt_helperbot
-    await mqtt_helperbot.start()
-
     resp = await webserver_client.get("/")
     assert resp.status == 200
 
-    await mqtt_helperbot.disconnect()
-
     bumper.xmpp_server.disconnect()
 
 
-@pytest.mark.usefixtures("mqtt_server")
+@pytest.mark.usefixtures("helper_bot")
 async def test_restartService(webserver_client):
     remove_existing_db()
 
@@ -69,11 +62,6 @@ async def test_restartService(webserver_client):
     bumper.xmpp_server = xmpp_server
     await xmpp_server.start_async_server()
 
-    # Start Helperbot
-    mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
-    bumper.mqtt_helperbot = mqtt_helperbot
-    await mqtt_helperbot.start()
-
     resp = await webserver_client.get("/restart_Helperbot")
     assert resp.status == 200
 
@@ -83,8 +71,6 @@ async def test_restartService(webserver_client):
     resp = await webserver_client.get("/restart_XMPPServer")
     assert resp.status == 200
 
-    await mqtt_helperbot.disconnect()
-
     xmpp_server.disconnect()
 
 
@@ -718,12 +704,11 @@ async def test_appsvr_api(webserver_client):
     assert jsonresp["ret"] == "ok"
 
 
-async def test_lg_logs(webserver_client):
+async def test_lg_logs(webserver_client, helper_bot: HelperBot):
     remove_existing_db()
     db.bot_add("sn_1234", "did_1234", "ls1ok3", "res_1234", "eco-ng")
     db.bot_set_mqtt("did_1234", True)
     confserver = create_webserver()
-    bumper.mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
 
     # Test return get status
     command_getstatus_resp = {
@@ -731,7 +716,7 @@ async def test_lg_logs(webserver_client):
         "resp": "<ctl ret='ok' status='idle'/>",
         "ret": "ok",
     }
-    bumper.mqtt_helperbot.send_command = mock.MagicMock(
+    helper_bot.send_command = mock.MagicMock(
         return_value=async_return(command_getstatus_resp)
     )
 
@@ -775,10 +760,9 @@ async def test_postLookup(webserver_client):
     assert test_resp["result"] == "ok"
 
 
-async def test_devmgr(webserver_client):
+async def test_devmgr(webserver_client, helper_bot: HelperBot):
     remove_existing_db()
     confserver = create_webserver()
-    bumper.mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
 
     # Test PollSCResult
     postbody = {"td": "PollSCResult"}
@@ -808,7 +792,7 @@ async def test_devmgr(webserver_client):
         "resp": "<ctl ret='ok' status='idle'/>",
         "ret": "ok",
     }
-    bumper.mqtt_helperbot.send_command = mock.MagicMock(
+    helper_bot.send_command = mock.MagicMock(
         return_value=async_return(command_getstatus_resp)
     )
     resp = await webserver_client.post("/api/iot/devmanager.do", json=postbody)
@@ -819,7 +803,7 @@ async def test_devmgr(webserver_client):
 
     # Test return fail timeout
     command_timeout_resp = {"id": "resp_1234", "errno": "timeout", "ret": "fail"}
-    bumper.mqtt_helperbot.send_command = mock.MagicMock(
+    helper_bot.send_command = mock.MagicMock(
         return_value=async_return(command_timeout_resp)
     )
     resp = await webserver_client.post("/api/iot/devmanager.do", json=postbody)
@@ -829,10 +813,9 @@ async def test_devmgr(webserver_client):
     assert test_resp["ret"] == "fail"
 
 
-async def test_dim_devmanager(webserver_client):
+async def test_dim_devmanager(webserver_client, helper_bot: HelperBot):
     remove_existing_db()
     confserver = create_webserver()
-    bumper.mqtt_helperbot = MQTTHelperBot(HOST, MQTT_PORT)
 
     # Test PollSCResult
     postbody = {"td": "PollSCResult"}
@@ -862,7 +845,7 @@ async def test_dim_devmanager(webserver_client):
         "resp": "<ctl ret='ok' status='idle'/>",
         "ret": "ok",
     }
-    bumper.mqtt_helperbot.send_command = mock.MagicMock(
+    helper_bot.send_command = mock.MagicMock(
         return_value=async_return(command_getstatus_resp)
     )
     resp = await webserver_client.post("/api/dim/devmanager.do", json=postbody)
@@ -873,7 +856,7 @@ async def test_dim_devmanager(webserver_client):
 
     # Test return fail timeout
     command_timeout_resp = {"id": "resp_1234", "errno": "timeout", "ret": "fail"}
-    bumper.mqtt_helperbot.send_command = mock.MagicMock(
+    helper_bot.send_command = mock.MagicMock(
         return_value=async_return(command_timeout_resp)
     )
     resp = await webserver_client.post("/api/dim/devmanager.do", json=postbody)
@@ -885,7 +868,7 @@ async def test_dim_devmanager(webserver_client):
 
     # Set bot not on mqtt
     db.bot_set_mqtt("did_1234", False)
-    bumper.mqtt_helperbot.send_command = mock.MagicMock(
+    helper_bot.send_command = mock.MagicMock(
         return_value=async_return(command_getstatus_resp)
     )
     resp = await webserver_client.post("/api/dim/devmanager.do", json=postbody)