resolve mypy issue on mqttserver.py
This commit is contained in:
Robert Resch
parent
479c0f1cba
commit
226f7556ce
7 changed files with 165 additions and 176 deletions
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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 = "<ctl ts='1547822804960' td='DustCaseST' st='0'/>"
|
||||
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 = "<ctl ts='1560904925396' td='errors' old='' new='110'/>"
|
||||
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 = "<ctl ts='1547822804960' td='DustCaseST' st='0'/>"
|
||||
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 = "<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_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(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue