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

fix solarman unit tests

Browse Source 
- fake Mqtt class
This commit is contained in:
Stefan Allius
2024-06-09 11:02:43 +02:00
parent 1deab4be6a
commit ff8adb5632
2 changed files with 55 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
app/src/gen3plus/solarman_v5.py
View File

@@ -375,7 +375,7 @@ class SolarmanV5(Message):
node_id = self.node_id
key = 'at_resp'
logger.info(f'{key}: {data_json}')
await self.publish_mqtt(f'{self.entity_prfx}{node_id}{key}', data_json) # noqa: E501
await self.mqtt.publish(f'{self.entity_prfx}{node_id}{key}', data_json) # noqa: E501
return
self.forward_at_cmd_resp = False
@@ -502,8 +502,9 @@ class SolarmanV5(Message):
self.__forward_msg()
async def publish_mqtt(self, key, data):
await self.mqtt.publish(key, data) # pragma: no cover
def publish_mqtt(self, key, data):
asyncio.ensure_future(
self.mqtt.publish(key, data))
def get_cmd_rsp_log_lvl(self) -> int:
ftype = self._recv_buffer[self.header_len]
@@ -527,8 +528,7 @@ class SolarmanV5(Message):
node_id = self.node_id
key = 'at_resp'
logger.info(f'{key}: {data_json}')
asyncio.ensure_future(
self.publish_mqtt(f'{self.entity_prfx}{node_id}{key}', data_json)) # noqa: E501
self.publish_mqtt(f'{self.entity_prfx}{node_id}{key}', data_json) # noqa: E501
return
elif ftype == self.MB_RTU_CMD:
valid = data[1]

61
app/tests/test_solarman.py
View File

@@ -24,12 +24,24 @@ class Writer():
def write(self, pdu: bytearray):
self.sent_pdu = pdu
class Mqtt():
def __init__(self):
self.key = ''
self.data = ''
async def publish(self, key, data):
self.key = key
self.data = data
class MemoryStream(SolarmanV5):
def __init__(self, msg, chunks = (0,), server_side: bool = True):
super().__init__(server_side)
if server_side:
self.mb.timeout = 1 # overwrite for faster testing
self.writer = Writer()
self.mqtt = Mqtt()
self.__msg = msg
self.__msg_len = len(msg)
self.__chunks = chunks
@@ -43,6 +55,8 @@ class MemoryStream(SolarmanV5):
self.test_exception_async_write = False
self.entity_prfx = ''
self.at_acl = {'mqtt': {'allow': ['AT+'], 'block': ['AT+WEBU']}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE', 'AT+TIME'], 'block': ['AT+WEBU']}}
self.key = ''
self.data = ''
def _timestamp(self):
return timestamp
@@ -54,6 +68,10 @@ class MemoryStream(SolarmanV5):
self.__msg += msg
self.__msg_len += len(msg)
def publish_mqtt(self, key, data):
self.key = key
self.data = data
def _read(self) -> int:
copied_bytes = 0
try:
@@ -478,9 +496,10 @@ def AtCommandIndMsgBlock(): # 0x4510
@pytest.fixture
def AtCommandRspMsg(): # 0x1510
msg = b'\xa5\x0a\x00\x10\x15\x03\x03' +get_sn() +b'\x01\x01'
msg = b'\xa5\x11\x00\x10\x15\x03\x03' +get_sn() +b'\x01\x01'
msg += total()
msg += hb()
msg += b'\x00\x00\x00\x00+ok'
msg += correct_checksum(msg)
msg += b'\x15'
return msg
@@ -1243,48 +1262,64 @@ async def test_msg_build_modbus_req(ConfigTsunInv1, DeviceIndMsg, DeviceRspMsg,
m.close()
@pytest.mark.asyncio
async def test_AT_cmd(ConfigTsunAllowAll, DeviceIndMsg, DeviceRspMsg, InverterIndMsg, InverterRspMsg, AtCommandIndMsg):
async def test_AT_cmd(ConfigTsunAllowAll, DeviceIndMsg, DeviceRspMsg, InverterIndMsg, InverterRspMsg, AtCommandIndMsg, AtCommandRspMsg):
ConfigTsunAllowAll
m = MemoryStream(DeviceIndMsg, (0,), True)
m.append_msg(InverterIndMsg)
m.read()
m.append_msg(AtCommandRspMsg)
m.read() # read device ind
assert m.control == 0x4110
assert str(m.seq) == '01:01'
assert m._recv_buffer==InverterIndMsg # unhandled next message
assert m._recv_buffer==InverterIndMsg + AtCommandRspMsg # unhandled next message
assert m._send_buffer==DeviceRspMsg
assert m._forward_buffer==DeviceIndMsg
m._send_buffer = bytearray(0) # clear send buffer for next test
m._forward_buffer = bytearray(0) # clear send buffer for next test
await m.send_at_cmd('AT+TIME=214028,1,60,120')
assert m._recv_buffer==InverterIndMsg # unhandled next message
assert m._recv_buffer==InverterIndMsg + AtCommandRspMsg # unhandled next message
assert m._send_buffer==b''
assert m._forward_buffer==b''
assert str(m.seq) == '01:01'
assert m.mqtt.key == ''
assert m.mqtt.data == ""
m.read()
m.read() # read inverter ind
assert m.control == 0x4210
assert str(m.seq) == '02:02'
assert m._recv_buffer==b''
assert m._recv_buffer==AtCommandRspMsg # unhandled next message
assert m._send_buffer==InverterRspMsg
assert m._forward_buffer==InverterIndMsg
m._send_buffer = bytearray(0) # clear send buffer for next test
m._forward_buffer = bytearray(0) # clear send buffer for next test
await m.send_at_cmd('AT+TIME=214028,1,60,120')
assert m._recv_buffer==b''
assert m._recv_buffer==AtCommandRspMsg # unhandled next message
assert m._send_buffer==AtCommandIndMsg
assert m._forward_buffer==b''
assert str(m.seq) == '02:03'
assert m.mqtt.key == ''
assert m.mqtt.data == ""
m._send_buffer = bytearray(0) # clear send buffer for next test
m.read() # read at resp
assert m.control == 0x1510
assert str(m.seq) == '03:03'
assert m._recv_buffer==b''
assert m._send_buffer==b''
assert m._forward_buffer==b''
assert m.key == 'at_resp'
assert m.data == "+ok"
m.test_exception_async_write = True
await m.send_at_cmd('AT+TIME=214028,1,60,120')
assert m._recv_buffer==b''
assert m._send_buffer==b''
assert m._forward_buffer==b''
assert str(m.seq) == '02:04'
assert str(m.seq) == '03:04'
assert m.forward_at_cmd_resp == False
assert m.mqtt.key == ''
assert m.mqtt.data == ""
m.close()
@pytest.mark.asyncio
@@ -1306,6 +1341,8 @@ async def test_AT_cmd_blocked(ConfigTsunAllowAll, DeviceIndMsg, DeviceRspMsg, In
assert m._send_buffer==b''
assert m._forward_buffer==b''
assert str(m.seq) == '01:01'
assert m.mqtt.key == ''
assert m.mqtt.data == ""
m.read()
assert m.control == 0x4210
@@ -1322,6 +1359,8 @@ async def test_AT_cmd_blocked(ConfigTsunAllowAll, DeviceIndMsg, DeviceRspMsg, In
assert m._forward_buffer==b''
assert str(m.seq) == '02:02'
assert m.forward_at_cmd_resp == False
assert m.mqtt.key == 'at_resp'
assert m.mqtt.data == "'AT+WEBU' is forbidden"
m.close()
def test_AT_cmd_ind(ConfigTsunInv1, AtCommandIndMsg):
@@ -1386,7 +1425,7 @@ def test_msg_at_command_rsp1(ConfigTsunInv1, AtCommandRspMsg):
assert m.control == 0x1510
assert str(m.seq) == '03:03'
assert m.header_len==11
assert m.data_len==10
assert m.data_len==17
assert m._forward_buffer==AtCommandRspMsg
assert m._send_buffer==b''
assert m.db.stat['proxy']['Unknown_Ctrl'] == 0
@@ -1405,7 +1444,7 @@ def test_msg_at_command_rsp2(ConfigTsunInv1, AtCommandRspMsg):
assert m.control == 0x1510
assert str(m.seq) == '03:03'
assert m.header_len==11
assert m.data_len==10
assert m.data_len==17
assert m._forward_buffer==b''
assert m._send_buffer==b''
assert m.db.stat['proxy']['Unknown_Ctrl'] == 0