diff --git a/bumper/__init__.py b/bumper/__init__.py index bcbf2a6..1fb27fc 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -125,19 +125,19 @@ async def start(): # Start MQTT Server # await start otherwise we get an error connecting the helper bot - await asyncio.create_task(mqtt_server.broker_coro()) + await asyncio.create_task(mqtt_server.start()) # Start MQTT Helperbot - asyncio.create_task(mqtt_helperbot.start_helper_bot()) + asyncio.create_task(mqtt_helperbot.start()) # Start XMPP Server asyncio.create_task(xmpp_server.start_async_server()) # Wait for helperbot to connect first - while mqtt_helperbot.Client is None: + while mqtt_helperbot.client is None: await asyncio.sleep(0.1) - while not mqtt_helperbot.Client.session.transitions.state == "connected": + while not mqtt_helperbot.client.session.transitions.state == "connected": await asyncio.sleep(0.1) # Start web servers @@ -177,7 +177,7 @@ async def shutdown(): await asyncio.sleep(0.1) if mqtt_server.broker.transitions.state == "started": await mqtt_server.broker.shutdown() - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() if xmpp_server.server: if xmpp_server.server._serving: xmpp_server.server.close() diff --git a/bumper/confserver.py b/bumper/confserver.py index b5859b3..37d9cd9 100644 --- a/bumper/confserver.py +++ b/bumper/confserver.py @@ -184,7 +184,7 @@ class ConfServer: bots = bumper.db_get().table("bots").all() clients = bumper.db_get().table("clients").all() - helperbot = bumper.mqtt_helperbot.Client.session.transitions.state + helperbot = bumper.mqtt_helperbot.client.session.transitions.state mqttserver = bumper.mqtt_server.broker xmppserver = bumper.xmpp_server mq_sessions = [] @@ -284,8 +284,8 @@ class ConfServer: async def restart_Helper(self): - await bumper.mqtt_helperbot.Client.disconnect() - asyncio.create_task(bumper.mqtt_helperbot.start_helper_bot()) + await bumper.mqtt_helperbot.client.disconnect() + asyncio.create_task(bumper.mqtt_helperbot.start()) async def restart_MQTT(self): @@ -307,7 +307,7 @@ class ConfServer: aloop = asyncio.get_event_loop() aloop.call_later( - 1.5, lambda: asyncio.create_task(bumper.mqtt_server.broker_coro()) + 1.5, lambda: asyncio.create_task(bumper.mqtt_server.start()) ) # In 1.5 seconds start broker async def restart_XMPP(self): diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 79c2d7d..6cb5534 100644 --- a/bumper/mqttserver.py +++ b/bumper/mqttserver.py @@ -3,13 +3,14 @@ import asyncio import json import os -from typing import MutableMapping +from typing import Any, Dict, MutableMapping, Optional, Union import amqtt import pkg_resources -from amqtt.broker import Broker +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 @@ -25,31 +26,30 @@ class CommandDto: def __init__(self, payload_type: str) -> None: self._payload_type = payload_type self._event = asyncio.Event() - self._response = None + self._response: Union[str, bytes] - async def wait_for_response(self): + async def wait_for_response(self) -> Union[str, Dict[str, Any]]: await self._event.wait() if self._payload_type == "j": return json.loads(self._response) else: return str(self._response) - def add_response(self, response): + def add_response(self, response: Union[str, bytes]) -> None: self._response = response self._event.set() class MQTTHelperBot: - Client = None - 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._client_id = "helperbot@bumper/helperbot" self._timeout = timeout + self._client: Optional[MQTTClient] = None @property def commands(self) -> MutableMapping[str, CommandDto]: @@ -59,18 +59,22 @@ class MQTTHelperBot: def timeout(self) -> float: return self._timeout - async def start_helper_bot(self): + @property + def client(self) -> MQTTClient: + return self._client + + async def start(self) -> None: try: - if self.Client is None: - self.Client = MQTTClient( - client_id=self.client_id, + if self._client is None: + self._client = MQTTClient( + client_id=self._client_id, config={"check_hostname": False, "reconnect_retries": 20}, ) - await self.Client.connect( + await self._client.connect( f"mqtts://{self._host}:{self._port}/", cafile=bumper.ca_cert ) - await self.Client.subscribe( + await self._client.subscribe( [ ("iot/p2p/+/+/+/+/helperbot/bumper/helperbot/+/+/+", QOS_0), ("iot/p2p/+", QOS_0), @@ -79,8 +83,11 @@ class MQTTHelperBot: ) except Exception as e: helperbotlog.exception(f"{e}") + raise e - async def _wait_for_resp(self, command_dto: CommandDto, request_id: str): + 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 @@ -100,65 +107,58 @@ class MQTTHelperBot: "debug": "wait for response timed out", } - async def send_command(self, cmdjson, requestid): - if not self.Client._handler.writer is None: - try: - topic = "iot/p2p/{}/helperbot/bumper/helperbot/{}/{}/{}/q/{}/{}".format( - cmdjson["cmdName"], - cmdjson["toId"], - cmdjson["toType"], - cmdjson["toRes"], - requestid, - cmdjson["payloadType"], - ) - command_dto = CommandDto(cmdjson["payloadType"]) - self.commands[requestid] = command_dto + async def send_command( + self, cmdjson: Dict[str, Any], request_id: str + ) -> Dict[str, Any]: + if self.client is None: + await self.start() + assert self.client is not None - if cmdjson["payloadType"] == "j": - payload = json.dumps(cmdjson["payload"]) - else: - payload = str(cmdjson["payload"]) + try: + topic = "iot/p2p/{}/helperbot/bumper/helperbot/{}/{}/{}/q/{}/{}".format( + cmdjson["cmdName"], + cmdjson["toId"], + cmdjson["toType"], + cmdjson["toRes"], + request_id, + cmdjson["payloadType"], + ) + command_dto = CommandDto(cmdjson["payloadType"]) + self.commands[request_id] = command_dto - await self.Client.publish(topic, payload.encode(), QOS_0) + if cmdjson["payloadType"] == "j": + payload = json.dumps(cmdjson["payload"]) + else: + payload = str(cmdjson["payload"]) - resp = await self._wait_for_resp(command_dto, requestid) - return resp - except Exception as e: - helperbotlog.exception(f"{e}") - return { - "id": requestid, - "errno": 500, - "ret": "fail", - "debug": "exception occurred please check bumper logs", - } - finally: - self.commands.pop(requestid, None) + await self.client.publish(topic, payload.encode(), QOS_0) + + resp = await self._wait_for_resp(command_dto, request_id) + return resp + except Exception as e: + helperbotlog.exception(f"{e}") + return { + "id": request_id, + "errno": 500, + "ret": "fail", + "debug": "exception occurred please check bumper logs", + } + finally: + self.commands.pop(request_id, None) class MQTTServer: - default_config = None - broker = None - - def __init__(self, host: str, port: int, **kwargs): + def __init__(self, host: str, port: int, **kwargs: Dict[str, Any]) -> None: try: self._host = host self._port = port - # Default config opts - passwd_file = os.path.join(os.path.join(bumper.data_dir, "passwd")) # For file auth, set user:hash in passwd file see # (https://hbmqtt.readthedocs.io/en/latest/references/hbmqtt.html#configuration-example) - - allow_anon = False - - for key, value in kwargs.items(): - if key == "password_file": - passwd_file = kwargs["password_file"] - - elif key == "allow_anonymous": - allow_anon = kwargs[ - "allow_anonymous" - ] # Set to True to allow anonymous authentication + passwd_file = kwargs.get( + "password_file", os.path.join(os.path.join(bumper.data_dir, "passwd")) + ) + allow_anon = kwargs.get("allow_anonymous", False) # The below adds a plugin to the amqtt.broker.plugins without having to futz with setup.py distribution = pkg_resources.Distribution("amqtt.broker.plugins") @@ -169,7 +169,7 @@ class MQTTServer: pkg_resources.working_set.add(distribution) # Initialize bot server - self.default_config = { + config = { "listeners": { "default": {"type": "tcp"}, "tls1": { @@ -193,35 +193,31 @@ class MQTTServer: }, } - self.broker = amqtt.broker.Broker(config=self.default_config) + self._broker = amqtt.broker.Broker(config=config) except Exception as e: mqttserverlog.exception(f"{e}") - async def broker_coro(self): + @property + def broker(self) -> Broker: + return self._broker + + async def start(self) -> None: mqttserverlog.info(f"Starting MQTT Server at {self._host}:{self._port}") try: await self.broker.start() - - except amqtt.broker.BrokerException as e: - mqttserverlog.exception(e) - # asyncio.create_task(bumper.shutdown()) - pass - except Exception as e: mqttserverlog.exception(f"{e}") - # asyncio.create_task(bumper.shutdown()) - pass + raise e class BumperMQTTServer_Plugin: - def __init__(self, context): + def __init__(self, context: BrokerContext) -> None: self.context = context try: self.auth_config = self.context.config["auth"] - self._users = dict() - self._read_password_file() + self._users = self._read_password_file() except KeyError: self.context.logger.warning( @@ -230,15 +226,13 @@ class BumperMQTTServer_Plugin: except Exception as e: mqttserverlog.exception(f"{e}") - async def authenticate(self, *args, **kwargs): + async def authenticate(self, session: Session, **kwargs: Dict[str, Any]) -> bool: authenticated = False + username = session.username + password = session.password + client_id = session.client_id try: - session = kwargs.get("session", None) - username = session.username - password = session.password - client_id = session.client_id - if "@" in client_id: didsplit = str(client_id).split("@") if not ( # if ecouser or bumper aren't in details it is a bot @@ -320,8 +314,9 @@ class BumperMQTTServer_Plugin: return authenticated - def _read_password_file(self): + def _read_password_file(self) -> Dict[str, str]: password_file = self.auth_config.get("password-file", None) + users: Dict[str, str] = {} if password_file: try: with open(password_file) as f: @@ -333,20 +328,22 @@ class BumperMQTTServer_Plugin: if not line.startswith("#"): # Allow comments in files (username, pwd_hash) = line.split(sep=":", maxsplit=3) if username: - self._users[username] = pwd_hash + users[username] = pwd_hash self.context.logger.debug( f"user: {username} - hash: {pwd_hash}" ) self.context.logger.debug( - f"{(len(self._users))} user(s) read from file {password_file}" + f"{(len(users))} user(s) read from file {password_file}" ) except FileNotFoundError: self.context.logger.warning(f"Password file {password_file} not found") - async def on_broker_client_connected(self, client_id): + return users + + async def on_broker_client_connected(self, client_id: str) -> None: self._set_client_connected(client_id, True) - def _set_client_connected(self, client_id, connected: bool): + def _set_client_connected(self, client_id: str, connected: bool) -> None: didsplit = str(client_id).split("@") bot = bumper.bot_get(didsplit[0]) @@ -359,7 +356,9 @@ class BumperMQTTServer_Plugin: if client: bumper.client_set_mqtt(client["resource"], connected) - async def on_broker_message_received(self, client_id, message): + async def on_broker_message_received( + self, message: IncomingApplicationMessage, **kwargs: Dict[str, Any] + ) -> None: topic = message.topic topic_split = str(topic).split("/") data_decoded = str(message.data.decode("utf-8")) @@ -392,5 +391,5 @@ class BumperMQTTServer_Plugin: f"Received Message - Topic: {topic} - Message: {data_decoded}" ) - async def on_broker_client_disconnected(self, client_id): + async def on_broker_client_disconnected(self, client_id: str) -> None: self._set_client_connected(client_id, False) diff --git a/requirements-test.txt b/requirements-test.txt index 16defd9..1a3012f 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -9,6 +9,4 @@ pytest-env==0.6.2 pytest-timeout==2.1.0 testfixtures==6.18.3 types-cachetools==4.2.9 - -#pbr = "*" -#autoflake = "*" +types-setuptools==57.4.8 diff --git a/tests/conftest.py b/tests/conftest.py index 0e3bd0a..bcf9903 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,7 +7,7 @@ from tests import HOST, MQTT_PORT @pytest.fixture async def mqtt_server(): mqtt_server = bumper.MQTTServer(HOST, MQTT_PORT, password_file="tests/passwd") - await mqtt_server.broker_coro() + await mqtt_server.start() bumper.mqtt_server = mqtt_server yield diff --git a/tests/test_confserver.py b/tests/test_confserver.py index 58409dc..8b83211 100644 --- a/tests/test_confserver.py +++ b/tests/test_confserver.py @@ -100,12 +100,12 @@ async def test_base(conf_server_client): # Start Helperbot mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) bumper.mqtt_helperbot = mqtt_helperbot - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() resp = await conf_server_client.get("/") assert resp.status == 200 - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() bumper.xmpp_server.disconnect() @@ -124,7 +124,7 @@ async def test_restartService(conf_server_client): # Start Helperbot mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) bumper.mqtt_helperbot = mqtt_helperbot - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() resp = await conf_server_client.get("/restart_Helperbot") assert resp.status == 200 @@ -135,7 +135,7 @@ async def test_restartService(conf_server_client): resp = await conf_server_client.get("/restart_XMPPServer") assert resp.status == 200 - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() xmpp_server.disconnect() diff --git a/tests/test_mqttserver.py b/tests/test_mqttserver.py index 42b13cd..22700e2 100644 --- a/tests/test_mqttserver.py +++ b/tests/test_mqttserver.py @@ -4,6 +4,7 @@ import time import amqtt import pytest +from amqtt.client import MQTTClient from amqtt.mqtt.constants import QOS_0 from testfixtures import LogCapture @@ -17,13 +18,13 @@ async def test_helperbot_message(): # Test broadcast message mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() assert ( - mqtt_helperbot.Client._connected_state._value == True + mqtt_helperbot.client._connected_state._value == True ) # 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(), QOS_0) + await mqtt_helperbot.client.publish(msg_topic_name, msg_payload.encode(), QOS_0) await asyncio.sleep(0.1) @@ -35,17 +36,17 @@ async def test_helperbot_message(): ) ) # Check broadcast message was logged l.clear() - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() # Send command to bot mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() assert ( - mqtt_helperbot.Client._connected_state._value == True + mqtt_helperbot.client._connected_state._value == True ) # 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(), QOS_0) + await mqtt_helperbot.client.publish(msg_topic_name, msg_payload.encode(), QOS_0) await asyncio.sleep(0.1) @@ -57,17 +58,17 @@ async def test_helperbot_message(): ) ) # Check send command message was logged l.clear() - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() # Received response to command mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() assert ( - mqtt_helperbot.Client._connected_state._value == True + mqtt_helperbot.client._connected_state._value == True ) # 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(), QOS_0) + await mqtt_helperbot.client.publish(msg_topic_name, msg_payload.encode(), QOS_0) await asyncio.sleep(0.1) @@ -79,17 +80,17 @@ async def test_helperbot_message(): ) ) # Check received response message was logged l.clear() - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() # Received unknown message mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() assert ( - mqtt_helperbot.Client._connected_state._value == True + mqtt_helperbot.client._connected_state._value == True ) # 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(), QOS_0) + await mqtt_helperbot.client.publish(msg_topic_name, msg_payload.encode(), QOS_0) await asyncio.sleep(0.1) @@ -101,17 +102,17 @@ async def test_helperbot_message(): ) ) # Check received message was logged l.clear() - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() # Received error message mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() assert ( - mqtt_helperbot.Client._connected_state._value == True + mqtt_helperbot.client._connected_state._value == True ) # 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(), QOS_0) + await mqtt_helperbot.client.publish(msg_topic_name, msg_payload.encode(), QOS_0) await asyncio.sleep(0.1) @@ -123,7 +124,7 @@ async def test_helperbot_message(): ) ) # Check received message was logged l.clear() - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() @pytest.mark.usefixtures("mqtt_server") @@ -132,9 +133,9 @@ async def test_helperbot_expire_message(): # Test broadcast message mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT, timeout) bumper.mqtt_helperbot = mqtt_helperbot - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() assert ( - mqtt_helperbot.Client._connected_state._value == True + mqtt_helperbot.client._connected_state._value == True ) # Check helperbot is connected expire_msg_payload = '{"ret":"ok","ver":"0.13.5"}' @@ -154,7 +155,7 @@ async def test_helperbot_expire_message(): await asyncio.sleep(0.1) msg_payload = "" msg_topic_name = "iot/atr/DustCaseST/bot_serial/ls1ok3/wC3g/x" - await mqtt_helperbot.Client.publish( + await mqtt_helperbot.client.publish( msg_topic_name, msg_payload.encode(), QOS_0 ) # Send another message to force get_msg @@ -162,7 +163,7 @@ async def test_helperbot_expire_message(): assert mqtt_helperbot.commands.get(request_id, None) == None - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() @pytest.mark.usefixtures("mqtt_server") @@ -170,9 +171,9 @@ async def test_helperbot_sendcommand(): timeout = 0.1 mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT, timeout) bumper.mqtt_helperbot = mqtt_helperbot - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() assert ( - mqtt_helperbot.Client._connected_state._value == True + mqtt_helperbot.client._connected_state._value == True ) # Check helperbot is connected cmdjson = { @@ -203,7 +204,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(), 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 == { @@ -235,7 +236,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(), 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 == { @@ -270,7 +271,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(), QOS_0) + await mqtt_helperbot.client.publish(msg_topic_name, msg_payload.encode(), QOS_0) commandresult = await mqtt_helperbot.send_command(cmdjson, "testj") @@ -300,7 +301,7 @@ async def test_helperbot_sendcommand(): "ret": "ok", } - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() async def test_mqttserver(): @@ -313,54 +314,49 @@ async def test_mqttserver(): HOST, MQTT_PORT, password_file="tests/passwd", allow_anonymous=True ) - await mqtt_server.broker_coro() + await mqtt_server.start() try: # Test helperbot connect mqtt_helperbot = bumper.MQTTHelperBot(HOST, MQTT_PORT) - await mqtt_helperbot.start_helper_bot() + await mqtt_helperbot.start() assert ( - mqtt_helperbot.Client._connected_state._value == True + mqtt_helperbot.client._connected_state._value == True ) # Check helperbot is connected - await mqtt_helperbot.Client.disconnect() + await mqtt_helperbot.client.disconnect() # Test client connect bumper.user_add("user_123") # Add user to db bumper.client_add("user_123", "ecouser.net", "resource_123") # Add client to db - 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 = amqtt.client.MQTTClient( - client_id=test_client.client_id, config={"check_hostname": False} + + client = MQTTClient( + client_id="user_123@ecouser.net/resource_123", + config={"check_hostname": False}, ) - await test_client.Client.connect( + await client.connect( f"mqtts://{HOST}:{MQTT_PORT}/", cafile=bumper.ca_cert, ) - assert ( - test_client.Client._connected_state._value == True - ) # Check client is connected - await test_client.Client.disconnect() - assert ( - test_client.Client._connected_state._value == False - ) # Check client is disconnected + assert client._connected_state._value == True # Check client is connected + await client.disconnect() + assert client._connected_state._value == False # Check client is disconnected # Test fake_bot connect - fake_bot = bumper.MQTTHelperBot(HOST, MQTT_PORT) - fake_bot.client_id = "bot_serial@ls1ok3/wC3g" - await fake_bot.start_helper_bot() - assert ( - fake_bot.Client._connected_state._value == True - ) # Check fake_bot is connected - await fake_bot.Client.disconnect() + client = MQTTClient( + client_id="bot_serial@ls1ok3/wC3g", config={"check_hostname": False} + ) + + await client.connect( + f"mqtts://{HOST}:{MQTT_PORT}/", + cafile=bumper.ca_cert, + ) + assert client._connected_state._value == True # Check fake_bot is connected + await client.disconnect() # Test file auth client connect - test_client = bumper.MQTTHelperBot(HOST, MQTT_PORT) - test_client.client_id = "test-file-auth" - # await test_client.start_helper_bot() - test_client.Client = amqtt.client.MQTTClient( - client_id=test_client.client_id, + client = MQTTClient( + client_id="test-file-auth", config={ "check_hostname": False, "auto_reconnect": False, @@ -369,24 +365,20 @@ async def test_mqttserver(): ) # good user/pass - await test_client.Client.connect( + await client.connect( f"mqtts://test-client:abc123!@{HOST}:{MQTT_PORT}/", cafile=bumper.ca_cert, cleansession=True, ) - assert ( - test_client.Client._connected_state._value == True - ) # Check client is connected - await test_client.Client.disconnect() - assert ( - test_client.Client._connected_state._value == False - ) # Check client is disconnected + assert client._connected_state._value == True # Check client is connected + await client.disconnect() + assert client._connected_state._value == False # Check client is disconnected # bad password with LogCapture() as l: - await test_client.Client.connect( + await client.connect( f"mqtts://test-client:notvalid!@{HOST}:{MQTT_PORT}/", cafile=bumper.ca_cert, cleansession=True, @@ -401,7 +393,7 @@ async def test_mqttserver(): order_matters=False, ) # no username in file - await test_client.Client.connect( + await client.connect( f"mqtts://test-client-noexist:notvalid!@{HOST}:{MQTT_PORT}/", cafile=bumper.ca_cert, cleansession=True, @@ -425,7 +417,7 @@ async def test_nofileauth_mqttserver(): mqtt_server = bumper.MQTTServer( HOST, MQTT_PORT, password_file="tests/passwd-notfound" ) - await mqtt_server.broker_coro() + await mqtt_server.start() await mqtt_server.broker.shutdown() l.check_present(