diff --git a/CHANGELOG.md b/CHANGELOG.md index ed901bc..8412be2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +- add Modbus fifo and timeout handler - build version string in the same format as TSUN for GEN3 invterts - add graceful shutdown - parse Modbus values and store them in the database diff --git a/app/proxy.svg b/app/proxy.svg index cef1e69..9ee1aba 100644 --- a/app/proxy.svg +++ b/app/proxy.svg @@ -65,14 +65,15 @@ A3 - -Modbus - - -build_msg() -recv_req() -recv_resp() -check_crc() + +Modbus + +err +retry_cnt + +build_msg() +recv_req() +recv_resp() @@ -166,9 +167,9 @@ A6->A3 - - -1 + + +1 has diff --git a/app/proxy.yuml b/app/proxy.yuml index 7514a93..e2ce12b 100644 --- a/app/proxy.yuml +++ b/app/proxy.yuml @@ -4,7 +4,7 @@ [note: You can stick notes on diagrams too!{bg:cornsilk}] [Singleton]^[Mqtt|ha_restarts;__client;__cb_MqttIsUp|publish();close()] -[Modbus||build_msg();recv_req();recv_resp();check_crc()] +[Modbus|err;retry_cnt|build_msg();recv_req();recv_resp()] [IterRegistry||__iter__]^[Message|server_side:bool;header_valid:bool;header_len:unsigned;data_len:unsigned;unique_id;node_id;sug_area;_recv_buffer:bytearray;_send_buffer:bytearray;_forward_buffer:bytearray;db:Infos;new_data:list|_read():void;close():void;inc_counter():void;dec_counter():void] [Message]^[Talent|await_conn_resp_cnt;id_str;contact_name;contact_mail;db:InfosG3;mb:Modbus;switch|msg_contact_info();msg_ota_update();msg_get_time();msg_collector_data();msg_inverter_data();msg_unknown();;close()] [Message]^[SolarmanV5|control;serial;snr;db:InfosG3P;mb:Modbus;switch|msg_unknown();;close()] diff --git a/app/src/gen3/talent.py b/app/src/gen3/talent.py index 976d468..def26fc 100644 --- a/app/src/gen3/talent.py +++ b/app/src/gen3/talent.py @@ -36,13 +36,12 @@ class Control: class Talent(Message): def __init__(self, server_side: bool, id_str=b''): - super().__init__(server_side) + super().__init__(server_side, self.send_modbus_cb, 11) self.await_conn_resp_cnt = 0 self.id_str = id_str self.contact_name = b'' self.contact_mail = b'' self.db = InfosG3() - self.forward_modbus_resp = False self.switch = { 0x00: self.msg_contact_info, 0x13: self.msg_ota_update, @@ -65,6 +64,7 @@ class Talent(Message): # deallocated by the garbage collector ==> we get a memory leak self.switch.clear() self.state = self.STATE_CLOSED + super().close() def __set_serial_no(self, serial_no: str): @@ -122,20 +122,25 @@ class Talent(Message): f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}') return + def send_modbus_cb(self, modbus_pdu: bytearray, state: str): + if self.state != self.STATE_UP: + return + + self.__build_header(0x70, 0x77) + self._send_buffer += b'\x00\x01\xa3\x28' # fixme + self._send_buffer += struct.pack('!B', len(modbus_pdu)) + self._send_buffer += modbus_pdu + self.__finish_send_msg() + + hex_dump_memory(logging.INFO, f'Send Modbus {state}:{self.addr}:', + self._send_buffer, len(self._send_buffer)) + self.writer.write(self._send_buffer) + self._send_buffer = bytearray(0) # self._send_buffer[sent:] + async def send_modbus_cmd(self, func, addr, val) -> None: if self.state != self.STATE_UP: return - self.forward_modbus_resp = False - self.__build_header(0x70, 0x77) - self._send_buffer += b'\x00\x01\xa3\x28' # fixme - modbus_msg = self.mb.build_msg(Modbus.INV_ADDR, func, addr, val) - self._send_buffer += struct.pack('!B', len(modbus_msg)) - self._send_buffer += modbus_msg - self.__finish_send_msg() - try: - await self.async_write('Send Modbus Command:') - except Exception: - self._send_buffer = bytearray(0) + self.mb.build_msg(Modbus.INV_ADDR, func, addr, val) def _init_new_client_conn(self) -> bool: contact_name = self.contact_name @@ -391,11 +396,11 @@ class Talent(Message): self.header_len+self.data_len] if self.ctrl.is_req(): - if not self.remoteStream.mb.recv_req(data[hdr_len:]): - self.inc_counter('Invalid_Msg_Format') - else: + if self.remoteStream.mb.recv_req(data[hdr_len:], + self.msg_forward): self.inc_counter('Modbus_Command') - self.remoteStream.forward_modbus_resp = True + else: + self.inc_counter('Invalid_Msg_Format') elif self.ctrl.is_ind(): # logger.debug(f'Modbus Ind MsgLen: {modbus_len}') self.modbus_elms = 0 @@ -405,12 +410,12 @@ class Talent(Message): if update: self.new_data[key] = True self.modbus_elms += 1 # count for unit tests - - if not self.forward_modbus_resp: - return else: logger.warning('Unknown Ctrl') self.inc_counter('Unknown_Ctrl') + self.forward(self._recv_buffer, self.header_len+self.data_len) + + def msg_forward(self): self.forward(self._recv_buffer, self.header_len+self.data_len) def msg_unknown(self): diff --git a/app/src/gen3plus/solarman_v5.py b/app/src/gen3plus/solarman_v5.py index 91d2bd0..a53769a 100644 --- a/app/src/gen3plus/solarman_v5.py +++ b/app/src/gen3plus/solarman_v5.py @@ -52,7 +52,7 @@ class SolarmanV5(Message): MB_RTU_CMD = 2 def __init__(self, server_side: bool): - super().__init__(server_side) + super().__init__(server_side, self.send_modbus_cb, 5) self.header_len = 11 # overwrite construcor in class Message self.control = 0 @@ -60,7 +60,6 @@ class SolarmanV5(Message): self.snr = 0 self.db = InfosG3P() self.time_ofs = 0 - self.forward_modbus_resp = False self.forward_at_cmd_resp = False self.switch = { @@ -104,6 +103,7 @@ class SolarmanV5(Message): # deallocated by the garbage collector ==> we get a memory leak self.switch.clear() self.state = self.STATE_CLOSED + super().close() def __set_serial_no(self, snr: int): serial_no = str(snr) @@ -301,20 +301,23 @@ class SolarmanV5(Message): self._heartbeat()) self.__finish_send_msg() - async def send_modbus_cmd(self, func, addr, val) -> None: + def send_modbus_cb(self, pdu: bytearray, state: str): if self.state != self.STATE_UP: return - self.forward_modbus_resp = False self.__build_header(0x4510) self._send_buffer += struct.pack(' None: + if self.state != self.STATE_UP: + return + self.mb.build_msg(Modbus.INV_ADDR, func, addr, val) async def send_at_cmd(self, AT_cmd: str) -> None: if self.state != self.STATE_UP: @@ -429,14 +432,14 @@ class SolarmanV5(Message): self.inc_counter('AT_Command') self.forward_at_cmd_resp = True elif ftype == self.MB_RTU_CMD: - if not self.remoteStream.mb.recv_req(data[15:]): - self.inc_counter('Invalid_Msg_Format') - else: + if self.remoteStream.mb.recv_req(data[15:], + self.__forward_msg()): self.inc_counter('Modbus_Command') - self.remoteStream.forward_modbus_resp = True + else: + self.inc_counter('Invalid_Msg_Format') + return self.__forward_msg() - # self.__send_ack_rsp(0x1510, ftype) def msg_command_rsp(self): data = self._recv_buffer[self.header_len: @@ -464,9 +467,7 @@ class SolarmanV5(Message): if inv_update: self.__build_model_name() - - if not self.forward_modbus_resp: - return + return self.__forward_msg() def msg_hbeat_ind(self): diff --git a/app/src/infos.py b/app/src/infos.py index dadacd7..eaf7062 100644 --- a/app/src/infos.py +++ b/app/src/infos.py @@ -224,22 +224,22 @@ class Infos: # 0xffffff03: {'name':['proxy', 'Voltage'], 'level': logging.DEBUG, 'unit': 'V', 'ha':{'dev':'proxy', 'dev_cla': 'voltage', 'stat_cla': 'measurement', 'id':'proxy_volt_', 'fmt':'| float','name': 'Grid Voltage'}}, # noqa: E501 # events - Register.EVENT_401: {'name': ['events', '401_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_402: {'name': ['events', '402_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_403: {'name': ['events', '403_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_404: {'name': ['events', '404_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_405: {'name': ['events', '405_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_406: {'name': ['events', '406_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_407: {'name': ['events', '407_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_408: {'name': ['events', '408_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_409: {'name': ['events', '409_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_410: {'name': ['events', '410_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_411: {'name': ['events', '411_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_412: {'name': ['events', '412_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_413: {'name': ['events', '413_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_414: {'name': ['events', '414_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_415: {'name': ['events', '415_GridFreqOverRating'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 - Register.EVENT_416: {'name': ['events', '416_'], 'level': logging.DEBUG, 'unit': ''}, # noqa: E501 + Register.EVENT_401: {'name': ['events', '401_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_402: {'name': ['events', '402_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_403: {'name': ['events', '403_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_404: {'name': ['events', '404_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_405: {'name': ['events', '405_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_406: {'name': ['events', '406_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_407: {'name': ['events', '407_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_408: {'name': ['events', '408_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_409: {'name': ['events', '409_No_Utility'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_410: {'name': ['events', '410_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_411: {'name': ['events', '411_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_412: {'name': ['events', '412_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_413: {'name': ['events', '413_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_414: {'name': ['events', '414_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_415: {'name': ['events', '415_GridFreqOverRating'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 + Register.EVENT_416: {'name': ['events', '416_'], 'level': logging.WARNING, 'unit': ''}, # noqa: E501 # grid measures: Register.GRID_VOLTAGE: {'name': ['grid', 'Voltage'], 'level': logging.DEBUG, 'unit': 'V', 'ha': {'dev': 'inverter', 'dev_cla': 'voltage', 'stat_cla': 'measurement', 'id': 'out_volt_', 'fmt': '| float', 'name': 'Grid Voltage', 'ent_cat': 'diagnostic'}}, # noqa: E501 diff --git a/app/src/messages.py b/app/src/messages.py index 01e3429..4968609 100644 --- a/app/src/messages.py +++ b/app/src/messages.py @@ -56,12 +56,14 @@ class Message(metaclass=IterRegistry): STATE_UP = 2 STATE_CLOSED = 3 - def __init__(self, server_side: bool): + def __init__(self, server_side: bool, send_modbus_cb, mb_timeout): self._registry.append(weakref.ref(self)) self.server_side = server_side if server_side: - self.mb = Modbus() + self.mb = Modbus(send_modbus_cb, mb_timeout) + else: + self.mb = None self.header_valid = False self.header_len = 0 @@ -91,6 +93,9 @@ class Message(metaclass=IterRegistry): Our puplic methods ''' def close(self) -> None: + if self.mb: + del self.mb + self.mb = None pass # pragma: no cover def inc_counter(self, counter: str) -> None: diff --git a/app/src/modbus.py b/app/src/modbus.py index 3ef0f5f..9cfeacd 100644 --- a/app/src/modbus.py +++ b/app/src/modbus.py @@ -1,33 +1,41 @@ +'''MODBUS module for TSUN inverter support + +TSUN uses the MODBUS in the RTU transmission mode over serial line. +see: https://modbus.org/docs/Modbus_Application_Protocol_V1_1b3.pdf +see: https://modbus.org/docs/Modbus_over_serial_line_V1_02.pdf + +A Modbus PDU consists of: 'Function-Code' + 'Data' +A Modbus RTU message consists of: 'Addr' + 'Modbus-PDU' + 'CRC-16' +The inverter is a MODBUS server and the proxy the MODBUS client. + +The 16-bit CRC is known as CRC-16-ANSI(reverse) +see: https://en.wikipedia.org/wiki/Computation_of_cyclic_redundancy_checks +''' import struct import logging -from typing import Generator +import asyncio +from typing import Generator, Callable if __name__ == "app.src.modbus": from app.src.infos import Register else: # pragma: no cover from infos import Register -####### -# TSUN uses the Modbus in the RTU transmission mode. -# see: https://modbus.org/docs/Modbus_over_serial_line_V1_02.pdf -# -# A Modbus PDU consists of: 'Function-Code' + 'Data' -# A Modbus RTU message consists of: 'Addr' + 'Modbus-PDU' + 'CRC-16' -# -# The 16-bit CRC is known as CRC-16-ANSI(reverse) -# see: https://en.wikipedia.org/wiki/Computation_of_cyclic_redundancy_checks -####### CRC_POLY = 0xA001 # (LSBF/reverse) CRC_INIT = 0xFFFF class Modbus(): + '''Simple MODBUS implementation with TX queue and retransmit timer''' INV_ADDR = 1 + '''MODBUS server address of the TSUN inverter''' READ_REGS = 3 + '''MODBUS function code: Read Holding Register''' READ_INPUTS = 4 + '''MODBUS function code: Read Input Register''' WRITE_SINGLE_REG = 6 - '''Modbus function codes''' + '''Modbus function code: Write Single Register''' __crc_tab = [] map = { @@ -65,97 +73,221 @@ class Modbus(): 0x3029: {'reg': Register.PV4_TOTAL_GENERATION, 'fmt': '!L', 'ratio': 0.01}, # noqa: E501 } - def __init__(self): + def __init__(self, snd_handler: Callable[[str], None], timeout: int = 1): if not len(self.__crc_tab): self.__build_crc_tab(CRC_POLY) + self.que = asyncio.Queue(100) + self.snd_handler = snd_handler + '''Send handler to transmit a MODBUS RTU request''' + self.rsp_handler = None + '''Response handler to forward the response''' + self.timeout = timeout + '''MODBUS response timeout in seconds''' + self.max_retries = 3 + '''Max retransmit for MODBUS requests''' + self.retry_cnt = 0 + self.last_req = b'' + self.counter = {} + '''Dictenary with statistic counter''' + self.counter['timeouts'] = 0 + self.counter['retries'] = {} + for i in range(0, self.max_retries+1): + self.counter['retries'][f'{i}'] = 0 + self.last_addr = 0 self.last_fcode = 0 self.last_len = 0 self.last_reg = 0 self.err = 0 + self.loop = asyncio.get_event_loop() + self.req_pend = False + self.tim = None - def build_msg(self, addr, func, reg, val): + def __del__(self): + logging.info(f'Modbus __del__:\n {self.counter}') + + def build_msg(self, addr: int, func: int, reg: int, val: int) -> None: + """Build MODBUS RTU request frame and add it to the tx queue + + Keyword arguments: + addr: RTU server address (inverter) + func: MODBUS function code + reg: 16-bit register number + val: 16 bit value + """ msg = struct.pack('>BBHH', addr, func, reg, val) msg += struct.pack(' bool: + def recv_req(self, buf: bytearray, + rsp_handler: Callable[[None], None] = None) -> bool: + """Add the received Modbus RTU request to the tx queue + + Keyword arguments: + buf: Modbus RTU pdu incl ADDR byte and trailing CRC + rsp_handler: Callback, if the received pdu is valid + + Returns: + True: PDU was added to the queue + False: PDU was ignored, due to an error + """ # logging.info(f'recv_req: first byte modbus:{buf[0]} len:{len(buf)}') - if not self.check_crc(buf): + if not self.__check_crc(buf): self.err = 1 logging.error('Modbus recv: CRC error') return False - if buf[0] != self.INV_ADDR: - self.err = 2 - logging.info(f'Modbus recv: Wrong addr{buf[0]}') - return False - res = struct.unpack_from('>BHH', buf, 1) - self.last_fcode = res[0] - self.last_reg = res[1] - self.last_len = res[2] - self.err = 0 + self.que.put_nowait({'req': buf, + 'rsp_hdl': rsp_handler}) + if self.que.qsize() == 1: + self.__send_next_from_que() + return True def recv_resp(self, info_db, buf: bytearray, node_id: str) -> \ - Generator[tuple[str, bool], None, None]: + Generator[tuple[str, bool, int | float | str], None, None]: + """Generator which check and parse a received MODBUS response. + + Keyword arguments: + info_db: database for info lockups + buf: received Modbus RTU response frame + node_id: string for logging which identifies the slave + + Returns on error and set Self.err to: + 1: CRC error + 2: Wrong server address + 3: Unexpected function code + 4: Unexpected data length + 5: No MODBUS request pending + """ # logging.info(f'recv_resp: first byte modbus:{buf[0]} len:{len(buf)}') - if not self.check_crc(buf): + if not self.req_pend: + self.err = 5 + return + if not self.__check_crc(buf): logging.error('Modbus resp: CRC error') self.err = 1 return - if buf[0] != self.INV_ADDR: + if buf[0] != self.last_addr: logging.info(f'Modbus resp: Wrong addr {buf[0]}') self.err = 2 return - if buf[1] != self.last_fcode: - logging.info(f'Modbus: Wrong fcode {buf[1]} != {self.last_fcode}') + fcode = buf[1] + if fcode != self.last_fcode: + logging.info(f'Modbus: Wrong fcode {fcode} != {self.last_fcode}') self.err = 3 return - elmlen = buf[2] >> 1 - if elmlen != self.last_len: - logging.info(f'Modbus: len error {elmlen} != {self.last_len}') - self.err = 4 + if self.last_addr == self.INV_ADDR and \ + (fcode == 3 or fcode == 4): + elmlen = buf[2] >> 1 + if elmlen != self.last_len: + logging.info(f'Modbus: len error {elmlen} != {self.last_len}') + self.err = 4 + return + first_reg = self.last_reg # save last_reg before sending next pdu + self.__stop_timer() # stop timer and send next pdu + + for i in range(0, elmlen): + addr = first_reg+i + if addr in self.map: + row = self.map[addr] + info_id = row['reg'] + fmt = row['fmt'] + val = struct.unpack_from(fmt, buf, 3+2*i) + result = val[0] + + if 'eval' in row: + result = eval(row['eval']) + if 'ratio' in row: + result = round(result * row['ratio'], 2) + + keys, level, unit, must_incr = info_db._key_obj(info_id) + + if keys: + name, update = info_db.update_db(keys, must_incr, + result) + yield keys[0], update, result + if update: + info_db.tracer.log(level, + f'[\'{node_id}\']MODBUS: {name}' + f' : {result}{unit}') + else: + self.__stop_timer() + + self.counter['retries'][f'{self.retry_cnt}'] += 1 + if self.rsp_handler: + self.rsp_handler() + self.__send_next_from_que() + + ''' + MODBUS response timer + ''' + def __start_timer(self) -> None: + '''Start response timer and set `req_pend` to True''' + self.req_pend = True + self.tim = self.loop.call_later(self.timeout, self.__timeout_cb) + # logging.debug(f'Modbus start timer {self}') + + def __stop_timer(self) -> None: + '''Stop response timer and set `req_pend` to False''' + self.req_pend = False + # logging.debug(f'Modbus stop timer {self}') + if self.tim: + self.tim.cancel() + + def __timeout_cb(self) -> None: + '''Rsponse timeout handler retransmit pdu or send next pdu''' + self.req_pend = False + + if self.retry_cnt < self.max_retries: + logging.debug(f'Modbus retrans {self}') + self.retry_cnt += 1 + self.__start_timer() + self.snd_handler(self.last_req, state='Retrans') + else: + logging.info(f'Modbus timeout {self}') + self.counter['timeouts'] += 1 + self.__send_next_from_que() + + def __send_next_from_que(self) -> None: + '''Get next MODBUS pdu from queue and transmit it''' + if self.req_pend: return - self.err = 0 + try: + item = self.que.get_nowait() + req = item['req'] + self.last_req = req + self.rsp_handler = item['rsp_hdl'] + self.last_addr = req[0] + self.last_fcode = req[1] - for i in range(0, elmlen): - addr = self.last_reg+i - if addr in self.map: - row = self.map[addr] - info_id = row['reg'] - fmt = row['fmt'] - val = struct.unpack_from(fmt, buf, 3+2*i) - result = val[0] + res = struct.unpack_from('>HH', req, 2) + self.last_reg = res[0] + self.last_len = res[1] + self.retry_cnt = 0 + self.__start_timer() + self.snd_handler(self.last_req, state='Command') + except asyncio.QueueEmpty: + pass - if 'eval' in row: - result = eval(row['eval']) - if 'ratio' in row: - result = round(result * row['ratio'], 2) - - keys, level, unit, must_incr = info_db._key_obj(info_id) - - if keys: - name, update = info_db.update_db(keys, must_incr, result) - yield keys[0], update, result - if update: - info_db.tracer.log(level, - f'[\'{node_id}\']MODBUS: {name}' - f' : {result}{unit}') - - def check_crc(self, msg) -> bool: + ''' + Helper function for CRC-16 handling + ''' + def __check_crc(self, msg: bytearray) -> bool: + '''Check CRC-16 and returns True if valid''' return 0 == self.__calc_crc(msg) - def __calc_crc(self, buffer: bytes) -> int: + def __calc_crc(self, buffer: bytearray) -> int: + '''Build CRC-16 for buffer and returns it''' crc = CRC_INIT for cur in buffer: crc = (crc >> 8) ^ self.__crc_tab[(crc ^ cur) & 0xFF] return crc - def __build_crc_tab(self, poly) -> None: + def __build_crc_tab(self, poly: int) -> None: + '''Build CRC-16 helper table, must be called exactly one time''' for index in range(256): data = index << 1 crc = 0 diff --git a/app/tests/test_modbus.py b/app/tests/test_modbus.py index ed6a68a..051401f 100644 --- a/app/tests/test_modbus.py +++ b/app/tests/test_modbus.py @@ -1,110 +1,177 @@ # test_with_pytest.py -# import pytest, logging +import pytest +import asyncio from app.src.modbus import Modbus -from app.src.infos import Infos +from app.src.infos import Infos, Register + +pytest_plugins = ('pytest_asyncio',) +pytestmark = pytest.mark.asyncio(scope="module") class TestHelper(Modbus): def __init__(self): - super().__init__() + super().__init__(self.send_cb) self.db = Infos() + self.pdu = None + self.send_calls = 0 + self.recv_responses = 0 + def send_cb(self, pdu: bytearray, state: str): + self.pdu = pdu + self.send_calls += 1 + def resp_handler(self): + self.recv_responses += 1 def test_modbus_crc(): - mb = Modbus() + '''Check CRC-16 calculation''' + mb = Modbus(None) assert 0x0b02 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x04') assert 0 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x04\x02\x0b') - assert mb.check_crc(b'\x01\x06\x20\x08\x00\x04\x02\x0b') + assert mb._Modbus__check_crc(b'\x01\x06\x20\x08\x00\x04\x02\x0b') assert 0xc803 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x00') assert 0 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x00\x03\xc8') - assert mb.check_crc(b'\x01\x06\x20\x08\x00\x00\x03\xc8') + assert mb._Modbus__check_crc(b'\x01\x06\x20\x08\x00\x00\x03\xc8') assert 0x5c75 == mb._Modbus__calc_crc(b'\x01\x03\x08\x01\x2c\x00\x2c\x02\x2c\x2c\x46') + def test_build_modbus_pdu(): - mb = Modbus() - pdu = mb.build_msg(1,6,0x2000,0x12) - assert pdu == b'\x01\x06\x20\x00\x00\x12\x02\x07' - assert mb.check_crc(pdu) - -def test_recv_req_crc(): - mb = Modbus() - res = mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x08') - assert not res - assert mb.last_fcode == 0 - assert mb.last_reg == 0 - assert mb.last_len == 0 - assert mb.err == 1 - -def test_recv_req_addr(): - mb = Modbus() - res = mb.recv_req(b'\x02\x06\x20\x00\x00\x12\x02\x34') - assert not res - assert mb.last_fcode == 0 - assert mb.last_reg == 0 - assert mb.last_len == 0 - assert mb.err == 2 + '''Check building and sending a MODBUS RTU''' + mb = TestHelper() + mb.build_msg(1,6,0x2000,0x12) + assert mb.pdu == b'\x01\x06\x20\x00\x00\x12\x02\x07' + assert mb._Modbus__check_crc(mb.pdu) + assert mb.last_addr == 1 + assert mb.last_fcode == 6 + assert mb.last_reg == 0x2000 + assert mb.last_len == 18 + assert mb.err == 0 def test_recv_req(): - mb = Modbus() - res = mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x07') - assert res + '''Receive a valid request, which must transmitted''' + mb = TestHelper() + assert mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x07') assert mb.last_fcode == 6 assert mb.last_reg == 0x2000 assert mb.last_len == 0x12 assert mb.err == 0 -def test_recv_recv_crc(): +def test_recv_req_crc_err(): + '''Receive a request with invalid CRC, which must be dropped''' mb = TestHelper() + assert not mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x08') + assert mb.pdu == None + assert mb.last_fcode == 0 + assert mb.last_reg == 0 + assert mb.last_len == 0 + assert mb.err == 1 + +def test_recv_resp_crc_err(): + '''Receive a response with invalid CRC, which must be dropped''' + mb = TestHelper() + # simulate a transmitted request + mb.req_pend = True + mb.last_addr = 1 mb.last_fcode = 3 mb.last_reg == 0x300e mb.last_len == 2 - + # check matching response, but with CRC error call = 0 for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf3', 'test'): call += 1 assert mb.err == 1 assert 0 == call + assert mb.req_pend == True + # cleanup queue + mb._Modbus__stop_timer() + assert not mb.req_pend -def test_recv_recv_addr(): +def test_recv_resp_invalid_addr(): + '''Receive a response with wrong server addr, which must be dropped''' mb = TestHelper() + mb.req_pend = True + # simulate a transmitted request + mb.last_addr = 1 mb.last_fcode = 3 mb.last_reg == 0x300e mb.last_len == 2 + # check not matching response, with wrong server addr call = 0 for key, update in mb.recv_resp(mb.db, b'\x02\x03\x04\x01\x2c\x00\x46\x88\xf4', 'test'): call += 1 assert mb.err == 2 assert 0 == call + assert mb.req_pend == True + assert mb.que.qsize() == 0 + + # cleanup queue + mb._Modbus__stop_timer() + assert not mb.req_pend def test_recv_recv_fcode(): + '''Receive a response with wrong function code, which must be dropped''' mb = TestHelper() - mb.last_fcode = 4 - mb.last_reg == 0x300e - mb.last_len == 2 - + mb.build_msg(1,4,0x300e,2) + assert mb.que.qsize() == 0 + assert mb.req_pend + + # check not matching response, with wrong function code call = 0 for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'): call += 1 + assert mb.err == 3 assert 0 == call + assert mb.req_pend == True + assert mb.que.qsize() == 0 -def test_recv_recv_len(): + # cleanup queue + mb._Modbus__stop_timer() + assert not mb.req_pend + +def test_recv_resp_len(): + '''Receive a response with wrong data length, which must be dropped''' mb = TestHelper() - mb.last_fcode = 3 - mb.last_reg == 0x300e - mb.last_len == 2 + mb.build_msg(1,3,0x300e,3) + assert mb.que.qsize() == 0 + assert mb.req_pend + assert mb.last_len == 3 + # check not matching response, with wrong data length call = 0 for key, update, _ in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'): call += 1 + assert mb.err == 4 assert 0 == call + assert mb.req_pend == True + assert mb.que.qsize() == 0 -def test_build_recv(): + # cleanup queue + mb._Modbus__stop_timer() + assert not mb.req_pend + +def test_recv_unexpect_resp(): + '''Receive a response when we havb't sent a request''' mb = TestHelper() - pdu = mb.build_msg(1,3,0x3007,6) - assert mb.check_crc(pdu) - assert mb.err == 0 + assert not mb.req_pend + + # check unexpected response, which must be dropped + call = 0 + for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'): + call += 1 + + assert mb.err == 5 + assert 0 == call + assert mb.req_pend == False + assert mb.que.qsize() == 0 + +def test_parse_resp(): + '''Receive matching response and parse the values''' + mb = TestHelper() + mb.build_msg(1,3,0x3007,6) + assert mb.que.qsize() == 0 + assert mb.req_pend + call = 0 exp_result = ['V0.0.212', 4.4, 0.7, 0.7, 30] for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'): @@ -120,15 +187,49 @@ def test_build_recv(): call += 1 assert 0 == mb.err assert 5 == call - + assert mb.que.qsize() == 0 + assert not mb.req_pend + +def test_queue(): + mb = TestHelper() + mb.build_msg(1,3,0x3022,4) + assert mb.que.qsize() == 0 + assert mb.req_pend + + assert mb.send_calls == 1 + assert mb.pdu == b'\x01\x030"\x00\x04\xeb\x03' + mb.pdu = None + assert mb.send_calls == 1 + assert mb.pdu == None + + assert mb.que.qsize() == 0 + + # cleanup queue + mb._Modbus__stop_timer() + assert not mb.req_pend + +def test_queue2(): + '''Check queue handling for build_msg() calls''' + mb = TestHelper() + mb.build_msg(1,3,0x3007,6) + mb.build_msg(1,6,0x2008,4) + assert mb.que.qsize() == 1 + assert mb.req_pend + mb.build_msg(1,3,0x3007,6) + assert mb.que.qsize() == 2 + assert mb.req_pend + + assert mb.send_calls == 1 + assert mb.pdu == b'\x01\x030\x07\x00\x06{\t' call = 0 + exp_result = ['V0.0.212', 4.4, 0.7, 0.7, 30] for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'): if key == 'grid': - assert update == False + assert update == True elif key == 'inverter': - assert update == False + assert update == True elif key == 'env': - assert update == False + assert update == True else: assert False assert exp_result[call] == val @@ -136,19 +237,144 @@ def test_build_recv(): assert 0 == mb.err assert 5 == call -def test_build_long(): - mb = TestHelper() - pdu = mb.build_msg(1,3,0x3022,4) - assert mb.check_crc(pdu) - assert mb.err == 0 + assert mb.que.qsize() == 1 + assert mb.send_calls == 2 + assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b' + + for key, update, val in mb.recv_resp(mb.db, b'\x01\x06\x20\x08\x00\x04\x02\x0b', 'test'): + pass + + assert mb.que.qsize() == 0 + assert mb.send_calls == 3 + assert mb.pdu == b'\x01\x030\x07\x00\x06{\t' call = 0 - exp_result = [3.0, 28841.4, 113.34] - for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x08\x01\x2c\x00\x2c\x02\x2c\x2c\x46\x75\x5c', 'test'): - if key == 'input': - assert update == True - assert exp_result[call] == val - else: - assert False + for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'): call += 1 assert 0 == mb.err - assert 3 == call + assert 5 == call + + assert mb.que.qsize() == 0 + assert not mb.req_pend + +def test_queue3(): + '''Check queue handling for recv_req() calls''' + mb = TestHelper() + assert mb.recv_req(b'\x01\x03\x30\x07\x00\x06{\t', mb.resp_handler) + assert mb.recv_req(b'\x01\x06\x20\x08\x00\x04\x02\x0b', mb.resp_handler) + assert mb.que.qsize() == 1 + assert mb.req_pend + assert mb.recv_req(b'\x01\x03\x30\x07\x00\x06{\t') + assert mb.que.qsize() == 2 + assert mb.req_pend + + assert mb.send_calls == 1 + assert mb.pdu == b'\x01\x030\x07\x00\x06{\t' + assert mb.recv_responses == 0 + + call = 0 + exp_result = ['V0.0.212', 4.4, 0.7, 0.7, 30] + for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'): + if key == 'grid': + assert update == True + elif key == 'inverter': + assert update == True + elif key == 'env': + assert update == True + else: + assert False + assert exp_result[call] == val + call += 1 + assert 0 == mb.err + assert 5 == call + assert mb.recv_responses == 1 + + assert mb.que.qsize() == 1 + assert mb.send_calls == 2 + assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b' + + for key, update, val in mb.recv_resp(mb.db, b'\x01\x06\x20\x08\x00\x04\x02\x0b', 'test'): + pass + assert 0 == mb.err + assert mb.recv_responses == 2 + + assert mb.que.qsize() == 0 + assert mb.send_calls == 3 + assert mb.pdu == b'\x01\x030\x07\x00\x06{\t' + call = 0 + for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'): + call += 1 + assert 0 == mb.err + assert mb.recv_responses == 2 + assert 5 == call + + + assert mb.que.qsize() == 0 + assert not mb.req_pend + +@pytest.mark.asyncio +async def test_timeout(): + '''Test MODBUS response timeout and RTU retransmitting''' + assert asyncio.get_running_loop() + mb = TestHelper() + mb.max_retries = 2 + mb.timeout = 0.1 # 100ms timeout for fast testing, expect a time resolution of at least 10ms + assert asyncio.get_running_loop() == mb.loop + mb.build_msg(1,3,0x3007,6) + mb.build_msg(1,6,0x2008,4) + + assert mb.que.qsize() == 1 + assert mb.req_pend + assert mb.retry_cnt == 0 + assert mb.send_calls == 1 + assert mb.pdu == b'\x01\x030\x07\x00\x06{\t' + + mb.pdu = None + await asyncio.sleep(0.11) # wait for first timeout and retransmittion + assert mb.que.qsize() == 1 + assert mb.req_pend + assert mb.retry_cnt == 1 + assert mb.send_calls == 2 + assert mb.pdu == b'\x01\x030\x07\x00\x06{\t' + + mb.pdu = None + await asyncio.sleep(0.11) # wait for second timeout and retransmittion + assert mb.que.qsize() == 1 + assert mb.req_pend + assert mb.retry_cnt == 2 + assert mb.send_calls == 3 + assert mb.pdu == b'\x01\x030\x07\x00\x06{\t' + + mb.pdu = None + await asyncio.sleep(0.11) # wait for third timeout and next pdu + assert mb.que.qsize() == 0 + assert mb.req_pend + assert mb.retry_cnt == 0 + assert mb.send_calls == 4 + assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b' + + mb.max_retries = 0 # next pdu without retranmsission + await asyncio.sleep(0.11) # wait for fourth timout + assert mb.que.qsize() == 0 + assert not mb.req_pend + assert mb.retry_cnt == 0 + assert mb.send_calls == 4 + + # assert mb.counter == {} + +def test_recv_unknown_data(): + '''Receive a response with an unknwon register''' + mb = TestHelper() + assert 0x9000 not in mb.map + mb.map[0x9000] = {'reg': Register.TEST_REG1, 'fmt': '!H', 'ratio': 1} + + mb.build_msg(1,3,0x9000,2) + + # check matching response, but with CRC error + call = 0 + for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'): + call += 1 + assert mb.err == 0 + assert 0 == call + assert not mb.req_pend + + del mb.map[0x9000] diff --git a/app/tests/test_solarman.py b/app/tests/test_solarman.py index 46df675..e339412 100644 --- a/app/tests/test_solarman.py +++ b/app/tests/test_solarman.py @@ -16,9 +16,19 @@ Infos.static_init() timestamp = int(time.time()) # 1712861197 heartbeat = 60 +class Writer(): + def __init__(self): + self.sent_pdu = b'' + + def write(self, pdu: bytearray): + self.sent_pdu = pdu + 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.__msg = msg self.__msg_len = len(msg) self.__chunks = chunks @@ -35,7 +45,6 @@ class MemoryStream(SolarmanV5): def _heartbeat(self) -> int: return heartbeat - def append_msg(self, msg): self.__msg += msg @@ -1194,7 +1203,8 @@ async def test_msg_build_modbus_req(ConfigTsunInv1, DeviceIndMsg, DeviceRspMsg, assert m._recv_buffer==InverterIndMsg # unhandled next message assert 0 == m.send_msg_ofs assert m._forward_buffer == b'' - assert m._send_buffer == b'' # modbus command must be ignore, cause connection is still not up + assert m.writer.sent_pdu == b'' # modbus command must be ignore, cause connection is still not up + assert m._send_buffer == b'' # modbus command must be ignore, cause connection is still not up m.read() assert m.control == 0x4210 @@ -1208,7 +1218,8 @@ async def test_msg_build_modbus_req(ConfigTsunInv1, DeviceIndMsg, DeviceRspMsg, await m.send_modbus_cmd(Modbus.WRITE_SINGLE_REG, 0x2008, 0) assert 0 == m.send_msg_ofs assert m._forward_buffer == b'' - assert m._send_buffer == MsgModbusCmd + assert m.writer.sent_pdu == MsgModbusCmd + assert m._send_buffer == b'' m._send_buffer = bytearray(0) # clear send buffer for next test m.test_exception_async_write = True @@ -1266,7 +1277,6 @@ async def test_AT_cmd(ConfigTsunAllowAll, DeviceIndMsg, DeviceRspMsg, InverterIn def test_at_command_ind(ConfigTsunInv1, AtCommandIndMsg): ConfigTsunInv1 m = MemoryStream(AtCommandIndMsg, (0,), False) - m.forward_modbus_resp = False m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['AT_Command'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 @@ -1285,7 +1295,6 @@ def test_at_command_ind(ConfigTsunInv1, AtCommandIndMsg): assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 assert m.db.stat['proxy']['AT_Command'] == 1 assert m.db.stat['proxy']['Modbus_Command'] == 0 - assert m.forward_modbus_resp == False m.close() def test_msg_at_command_rsp1(ConfigTsunInv1, AtCommandRspMsg): @@ -1331,7 +1340,6 @@ def test_msg_modbus_req(ConfigTsunInv1, MsgModbusCmd): m = MemoryStream(b'') c = m.createClientStream(MsgModbusCmd) - m.forward_modbus_resp = False m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['AT_Command'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 @@ -1349,7 +1357,6 @@ def test_msg_modbus_req(ConfigTsunInv1, MsgModbusCmd): assert m.db.stat['proxy']['AT_Command'] == 0 assert m.db.stat['proxy']['Modbus_Command'] == 1 assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 - assert m.forward_modbus_resp == True m.close() def test_msg_modbus_req2(ConfigTsunInv1, MsgModbusCmdCrcErr): @@ -1357,7 +1364,6 @@ def test_msg_modbus_req2(ConfigTsunInv1, MsgModbusCmdCrcErr): m = MemoryStream(b'') c = m.createClientStream(MsgModbusCmdCrcErr) - m.forward_modbus_resp = False m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['AT_Command'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 @@ -1375,13 +1381,11 @@ def test_msg_modbus_req2(ConfigTsunInv1, MsgModbusCmdCrcErr): assert m.db.stat['proxy']['AT_Command'] == 0 assert m.db.stat['proxy']['Modbus_Command'] == 0 assert m.db.stat['proxy']['Invalid_Msg_Format'] == 1 - assert m.forward_modbus_resp == True m.close() def test_msg_unknown_cmd_req(ConfigTsunInv1, MsgUnknownCmd): ConfigTsunInv1 m = MemoryStream(MsgUnknownCmd, (0,), False) - m.forward_modbus_resp = False m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['AT_Command'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 @@ -1399,15 +1403,14 @@ def test_msg_unknown_cmd_req(ConfigTsunInv1, MsgUnknownCmd): assert m.db.stat['proxy']['AT_Command'] == 0 assert m.db.stat['proxy']['Modbus_Command'] == 0 assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 - assert m.forward_modbus_resp == False m.close() def test_msg_modbus_rsp1(ConfigTsunInv1, MsgModbusRsp): + '''Modbus response without a valid Modbus request must be dropped''' ConfigTsunInv1 m = MemoryStream(MsgModbusRsp) m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 - m.forward_modbus_resp = False m.read() # read complete msg, and dispatch msg assert not m.header_valid # must be invalid, since msg was handled and buffer flushed assert m.msg_count == 1 @@ -1422,33 +1425,18 @@ def test_msg_modbus_rsp1(ConfigTsunInv1, MsgModbusRsp): m.close() def test_msg_modbus_rsp2(ConfigTsunInv1, MsgModbusRsp): - ConfigTsunInv1 - m = MemoryStream(MsgModbusRsp) - m.db.stat['proxy']['Unknown_Ctrl'] = 0 - m.db.stat['proxy']['Modbus_Command'] = 0 - m.forward_modbus_resp = True - m.read() # read complete msg, and dispatch msg - assert not m.header_valid # must be invalid, since msg was handled and buffer flushed - assert m.msg_count == 1 - assert m.control == 0x1510 - assert str(m.seq) == '03:03' - assert m.header_len==11 - assert m.data_len==59 - assert m._forward_buffer==MsgModbusRsp - assert m._send_buffer==b'' - assert m.db.stat['proxy']['Unknown_Ctrl'] == 0 - assert m.db.stat['proxy']['Modbus_Command'] == 0 - m.close() - -def test_msg_modbus_rsp3(ConfigTsunInv1, MsgModbusRsp): + '''Modbus response with a valid Modbus request must be forwarded''' ConfigTsunInv1 m = MemoryStream(MsgModbusRsp) m.append_msg(MsgModbusRsp) - m.forward_modbus_resp = True + m.mb.rsp_handler = m._SolarmanV5__forward_msg + m.mb.last_addr = 1 m.mb.last_fcode = 3 m.mb.last_len = 20 m.mb.last_reg = 0x3008 + m.mb.req_pend = True + m.mb.err = 0 # assert m.db.db == {'inverter': {'Manufacturer': 'TSUN', 'Equipment_Model': 'TSOL-MSxx00'}} m.new_data['inverter'] = False @@ -1465,7 +1453,7 @@ def test_msg_modbus_rsp3(ConfigTsunInv1, MsgModbusRsp): m.read() # read complete msg, and dispatch msg assert not m.header_valid # must be invalid, since msg was handled and buffer flushed - assert m.mb.err == 0 + assert m.mb.err == 5 assert m.msg_count == 2 assert m._forward_buffer==MsgModbusRsp assert m._send_buffer==b'' @@ -1480,7 +1468,6 @@ def test_msg_unknown_rsp(ConfigTsunInv1, MsgUnknownCmdRsp): m = MemoryStream(MsgUnknownCmdRsp) m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 - m.forward_modbus_resp = True m.read() # read complete msg, and dispatch msg assert not m.header_valid # must be invalid, since msg was handled and buffer flushed assert m.msg_count == 1 @@ -1514,10 +1501,14 @@ def test_msg_modbus_fragment(ConfigTsunInv1, MsgModbusRsp): m = MemoryStream(MsgModbusRsp+b'\x00\x00\x00\x45\x10\x52\x31', (0,)) m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 - m.forward_modbus_resp = True + m.mb.rsp_handler = m._SolarmanV5__forward_msg + m.mb.last_addr = 1 m.mb.last_fcode = 3 m.mb.last_len = 20 m.mb.last_reg = 0x3008 + m.mb.req_pend = True + m.mb.err = 0 + m.read() # read complete msg, and dispatch msg assert not m.header_valid # must be invalid, since msg was handled and buffer flushed assert m.msg_count == 1 diff --git a/app/tests/test_talent.py b/app/tests/test_talent.py index ed39111..3387d99 100644 --- a/app/tests/test_talent.py +++ b/app/tests/test_talent.py @@ -12,10 +12,21 @@ pytest_plugins = ('pytest_asyncio',) Infos.static_init() tracer = logging.getLogger('tracer') - + + +class Writer(): + def __init__(self): + self.sent_pdu = b'' + + def write(self, pdu: bytearray): + self.sent_pdu = pdu + class MemoryStream(Talent): 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.__msg = msg self.__msg_len = len(msg) self.__chunks = chunks @@ -803,6 +814,9 @@ def test_proxy_counter(): def test_msg_modbus_req(ConfigTsunInv1, MsgModbusCmd): ConfigTsunInv1 m = MemoryStream(b'') + m.id_str = b"R170000000000001" + m.state = m.STATE_UP + c = m.createClientStream(MsgModbusCmd) m.db.stat['proxy']['Unknown_Ctrl'] = 0 @@ -817,8 +831,12 @@ def test_msg_modbus_req(ConfigTsunInv1, MsgModbusCmd): assert c.msg_id==119 assert c.header_len==23 assert c.data_len==13 - assert c._forward_buffer==MsgModbusCmd + assert c._forward_buffer==b'' assert c._send_buffer==b'' + assert m.id_str == b"R170000000000001" + assert m._forward_buffer==b'' + assert m._send_buffer==b'' + assert m.writer.sent_pdu == MsgModbusCmd assert m.db.stat['proxy']['Unknown_Ctrl'] == 0 assert m.db.stat['proxy']['Modbus_Command'] == 1 assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 @@ -827,6 +845,7 @@ def test_msg_modbus_req(ConfigTsunInv1, MsgModbusCmd): def test_msg_modbus_req2(ConfigTsunInv1, MsgModbusCmdCrcErr): ConfigTsunInv1 m = MemoryStream(b'') + m.id_str = b"R170000000000001" c = m.createClientStream(MsgModbusCmdCrcErr) m.db.stat['proxy']['Unknown_Ctrl'] = 0 @@ -841,19 +860,22 @@ def test_msg_modbus_req2(ConfigTsunInv1, MsgModbusCmdCrcErr): assert c.msg_id==119 assert c.header_len==23 assert c.data_len==13 - assert c._forward_buffer==MsgModbusCmdCrcErr + assert c._forward_buffer==b'' assert c._send_buffer==b'' + assert m._forward_buffer==b'' + assert m._send_buffer==b'' + assert m.writer.sent_pdu ==b'' assert m.db.stat['proxy']['Unknown_Ctrl'] == 0 assert m.db.stat['proxy']['Modbus_Command'] == 0 assert m.db.stat['proxy']['Invalid_Msg_Format'] == 1 m.close() def test_msg_modbus_rsp1(ConfigTsunInv1, MsgModbusRsp): + '''Modbus response without a valid Modbus request must be dropped''' ConfigTsunInv1 m = MemoryStream(MsgModbusRsp) m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 - m.forward_modbus_resp = False m.read() # read complete msg, and dispatch msg assert not m.header_valid # must be invalid, since msg was handled and buffer flushed assert m.msg_count == 1 @@ -869,36 +891,20 @@ def test_msg_modbus_rsp1(ConfigTsunInv1, MsgModbusRsp): assert m.db.stat['proxy']['Modbus_Command'] == 0 m.close() -def test_msg_modbus_rsp2(ConfigTsunInv1, MsgModbusRsp): - ConfigTsunInv1 - m = MemoryStream(MsgModbusRsp) - m.db.stat['proxy']['Unknown_Ctrl'] = 0 - m.db.stat['proxy']['Modbus_Command'] = 0 - m.forward_modbus_resp = True - m.read() # read complete msg, and dispatch msg - assert not m.header_valid # must be invalid, since msg was handled and buffer flushed - assert m.msg_count == 1 - assert m.id_str == b"R170000000000001" - assert m.unique_id == 'R170000000000001' - assert int(m.ctrl)==145 - assert m.msg_id==119 - assert m.header_len==23 - assert m.data_len==13 - assert m._forward_buffer==MsgModbusRsp - assert m._send_buffer==b'' - assert m.db.stat['proxy']['Unknown_Ctrl'] == 0 - assert m.db.stat['proxy']['Modbus_Command'] == 0 - m.close() - -def test_msg_modbus_rsp3(ConfigTsunInv1, MsgModbusResp20): +def test_msg_modbus_rsp2(ConfigTsunInv1, MsgModbusResp20): + '''Modbus response with a valid Modbus request must be forwarded''' ConfigTsunInv1 m = MemoryStream(MsgModbusResp20) m.append_msg(MsgModbusResp20) - m.forward_modbus_resp = True + m.mb.rsp_handler = m.msg_forward + m.mb.last_addr = 1 m.mb.last_fcode = 3 m.mb.last_len = 20 m.mb.last_reg = 0x3008 + m.mb.req_pend = True + m.mb.err = 0 + assert m.db.db == {} m.new_data['inverter'] = False @@ -915,7 +921,7 @@ def test_msg_modbus_rsp3(ConfigTsunInv1, MsgModbusResp20): m.read() # read complete msg, and dispatch msg assert not m.header_valid # must be invalid, since msg was handled and buffer flushed - assert m.mb.err == 0 + assert m.mb.err == 5 assert m.msg_count == 2 assert m._forward_buffer==MsgModbusResp20 assert m._send_buffer==b'' @@ -951,10 +957,14 @@ def test_msg_modbus_fragment(ConfigTsunInv1, MsgModbusResp20): m = MemoryStream(MsgModbusResp20+b'\x00\x00\x00\x45\x10\x52\x31', (0,)) m.db.stat['proxy']['Unknown_Ctrl'] = 0 m.db.stat['proxy']['Modbus_Command'] = 0 - m.forward_modbus_resp = True + m.mb.rsp_handler = m.msg_forward + m.mb.last_addr = 1 m.mb.last_fcode = 3 m.mb.last_len = 20 m.mb.last_reg = 0x3008 + m.mb.req_pend = True + m.mb.err = 0 + m.read() # read complete msg, and dispatch msg assert not m.header_valid # must be invalid, since msg was handled and buffer flushed assert m.msg_count == 1 @@ -981,19 +991,22 @@ async def test_msg_build_modbus_req(ConfigTsunInv1, MsgModbusCmd): assert 0 == m.send_msg_ofs assert m._forward_buffer == b'' assert m._send_buffer == b'' + assert m.writer.sent_pdu == b'' m.state = m.STATE_UP await m.send_modbus_cmd(Modbus.WRITE_SINGLE_REG, 0x2008, 0) assert 0 == m.send_msg_ofs assert m._forward_buffer == b'' - assert m._send_buffer == MsgModbusCmd + assert m._send_buffer == b'' + assert m.writer.sent_pdu == MsgModbusCmd - m._send_buffer = bytearray(0) # clear send buffer for next test + m.writer.sent_pdu = bytearray(0) # clear send buffer for next test m.test_exception_async_write = True await m.send_modbus_cmd(Modbus.WRITE_SINGLE_REG, 0x2008, 0) assert 0 == m.send_msg_ofs assert m._forward_buffer == b'' assert m._send_buffer == b'' + assert m.writer.sent_pdu == b'' m.close() ''' def test_zombie_conn(ConfigTsunInv1, MsgInverterInd):