sv/tsun-gen3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test dispatcher exceptions

Browse Source
This commit is contained in:
Stefan Allius
2025-05-18 23:53:12 +02:00
parent 51aec15148
commit 55665c95f8
2 changed files with 23 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/src/mqtt.py
View File

@@ -148,7 +148,8 @@ class Mqtt(metaclass=Singleton):
f' {status}') f' {status}')
if status == 'online': if status == 'online':
self.ha_restarts += 1 self.ha_restarts += 1
await self.__cb_mqtt_is_up() if self.__cb_mqtt_is_up:
await self.__cb_mqtt_is_up()
async def _out_coeff(self, message): async def _out_coeff(self, message):
payload = message.payload.decode("UTF-8") payload = message.payload.decode("UTF-8")

21
app/tests/test_mqtt.py
View File

@@ -255,6 +255,10 @@ async def test_msg_dispatch(config_mqtt_conn, spy_modbus_cmd):
spy = spy_modbus_cmd spy = spy_modbus_cmd
try: try:
m = Mqtt(None) m = Mqtt(None)
msg = aiomqtt.Message(topic= 'homeassistant/status', payload= b'online', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
assert m.ha_restarts == 1
msg = aiomqtt.Message(topic= 'tsun/inv_1/rated_load', payload= b'2', qos= 0, retain = False, mid= 0, properties= None) msg = aiomqtt.Message(topic= 'tsun/inv_1/rated_load', payload= b'2', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg) await m.dispatch_msg(msg)
spy.assert_awaited_once_with(Modbus.WRITE_SINGLE_REG, 0x2008, 2, logging.INFO) spy.assert_awaited_once_with(Modbus.WRITE_SINGLE_REG, 0x2008, 2, logging.INFO)
@@ -279,6 +283,23 @@ async def test_msg_dispatch(config_mqtt_conn, spy_modbus_cmd):
await m.dispatch_msg(msg) await m.dispatch_msg(msg)
spy.assert_awaited_once_with(Modbus.READ_INPUTS, 0x3000, 10, logging.INFO) spy.assert_awaited_once_with(Modbus.READ_INPUTS, 0x3000, 10, logging.INFO)
# test dispatching with empty mapping table
m.topic_defs.clear()
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_1/modbus_read_inputs', payload= b'0x3000, 10', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_not_called()
# test dispatching with incomplete mapping table - invalid fnc defined
m.topic_defs.append(
{'prefix': 'entity_prefix', 'topic': '/+/dcu_power',
'full_topic': 'entity_prefix/+/dcu_power', 'fnc': 'invalid'}
)
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_1/modbus_read_inputs', payload= b'0x3000, 10', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_not_called()
finally: finally:
await m.close() await m.close()