From a016e6c3e2ea58635b6c504143f76c5c96a053c0 Mon Sep 17 00:00:00 2001 From: Stefan Allius Date: Sun, 24 Sep 2023 22:47:43 +0200 Subject: [PATCH] add unit tests --- app/src/messages.py | 3 +- app/tests/test_infos.py | 55 +++++++++++++ app/tests/test_messages.py | 160 +++++++++++++++++++++++++++++++++++++ 3 files changed, 217 insertions(+), 1 deletion(-) create mode 100644 app/tests/test_infos.py create mode 100644 app/tests/test_messages.py diff --git a/app/src/messages.py b/app/src/messages.py index 4f8b57b..b831ef4 100644 --- a/app/src/messages.py +++ b/app/src/messages.py @@ -1,12 +1,13 @@ import struct, logging, time, datetime import weakref -from config import Config from datetime import datetime if __name__ == "app.src.messages": from app.src.infos import Infos + from app.src.config import Config else: from infos import Infos + from config import Config logger = logging.getLogger('msg') diff --git a/app/tests/test_infos.py b/app/tests/test_infos.py new file mode 100644 index 0000000..f7030cf --- /dev/null +++ b/app/tests/test_infos.py @@ -0,0 +1,55 @@ +# test_with_pytest.py +import pytest, json +from app.src.infos import Infos + +@pytest.fixture +def ContrDataSeq(): # Get Time Request message + msg = b'\x00\x00\x00\x15\x00\x09\x2b\xa8\x54\x10\x52\x53\x57\x5f\x34\x30\x30\x5f\x56\x31\x2e\x30\x30\x2e\x30\x36\x00\x09\x27\xc0\x54\x06\x52\x61\x79\x6d\x6f' + msg += b'\x6e\x00\x09\x2f\x90\x54\x0b\x52\x53\x57\x2d\x31\x2d\x31\x30\x30\x30\x31\x00\x09\x5a\x88\x54\x0f\x74\x2e\x72\x61\x79\x6d\x6f\x6e\x69\x6f\x74\x2e\x63\x6f\x6d\x00\x09\x5a\xec\x54' + msg += b'\x1c\x6c\x6f\x67\x67\x65\x72\x2e\x74\x61\x6c\x65\x6e\x74\x2d\x6d\x6f\x6e\x69\x74\x6f\x72\x69\x6e\x67\x2e\x63\x6f\x6d\x00\x0d\x00\x20\x49\x00\x00\x00\x01\x00\x0c\x35\x00\x49\x00' + msg += b'\x00\x00\x64\x00\x0c\x96\xa8\x49\x00\x00\x00\x1d\x00\x0c\x7f\x38\x49\x00\x00\x00\x01\x00\x0c\xfc\x38\x49\x00\x00\x00\x01\x00\x0c\xf8\x50\x49\x00\x00\x01\x2c\x00\x0c\x63\xe0\x49' + msg += b'\x00\x00\x00\x00\x00\x0c\x67\xc8\x49\x00\x00\x00\x00\x00\x0c\x50\x58\x49\x00\x00\x00\x01\x00\x09\x5e\x70\x49\x00\x00\x13\x8d\x00\x09\x5e\xd4\x49\x00\x00\x13\x8d\x00\x09\x5b\x50' + msg += b'\x49\x00\x00\x00\x02\x00\x0d\x04\x08\x49\x00\x00\x00\x00\x00\x07\xa1\x84\x49\x00\x00\x00\x01\x00\x0c\x50\x59\x49\x00\x00\x00\x4c\x00\x0d\x1f\x60\x49\x00\x00\x00\x00' + return msg + + +def test_parse_control(ContrDataSeq): + i = Infos() + for key, result in i.parse (ContrDataSeq): + pass + + assert json.dumps(i.db) == json.dumps( +{"collector": {"Collector_Fw_Version": "RSW_400_V1.00.06", "Chip_Type": "Raymon", "Chip_Model": "RSW-1-10001", "Trace_URL": "t.raymoniot.com", "Logger_URL": "logger.talent-monitoring.com", "Data_Up_Interval": 300}, "env": {"Signal_Strength": 100}, "total": {"Power_On_Time": 29}}) + +def test_build_ha_conf(): + i = Infos() + d_json, id = next (i.ha_confs(prfx="tsun/garagendach/", snr='123')) + assert id == 'out_power_123' + assert d_json == json.dumps({"name": "Actual Power", "stat_t": "tsun/garagendach/grid", "dev_cla": "power", "stat_cla": "measurement", "uniq_id": "out_power_123", "val_tpl": "{{value_json['Output_Power'] | float}}", "unit_of_meas": "W", "dev": {"name": "Microinverter", "mdl": "MS-600", "ids": ["inverter_123"], "mf": "TSUN", "sa": "", "sw": "0.01", "hw": "Hw0.01"}}) + +def test_build_ha_conf2(): + i = Infos() + tests = 0 + for d_json, id in i.ha_confs(prfx="tsun/garagendach/", snr='123'): + + if id == 'out_power_123': + assert d_json == json.dumps({"name": "Actual Power", "stat_t": "tsun/garagendach/grid", "dev_cla": "power", "stat_cla": "measurement", "uniq_id": "out_power_123", "val_tpl": "{{value_json['Output_Power'] | float}}", "unit_of_meas": "W", "dev": {"name": "Microinverter", "mdl": "MS-600", "ids": ["inverter_123"], "mf": "TSUN", "sa": "", "sw": "0.01", "hw": "Hw0.01"}}) + tests +=1 + + elif id == 'daily_gen_123': + assert d_json == json.dumps({"name": "Daily Generation", "stat_t": "tsun/garagendach/total", "dev_cla": "energy", "stat_cla": "total_increasing", "uniq_id": "daily_gen_123", "val_tpl": "{{value_json['Daily_Generation'] | float}}", "unit_of_meas": "kWh", "dev": {"name": "Microinverter", "mdl": "MS-600", "ids": ["inverter_123"], "mf": "TSUN", "sa": "", "sw": "0.01", "hw": "Hw0.01"}}) + tests +=1 + + elif id == 'power_pv1_123': + assert d_json == json.dumps({"name": "Power PV1", "stat_t": "tsun/garagendach/input", "dev_cla": "power", "stat_cla": "measurement", "uniq_id": "power_pv1_123", "val_tpl": "{{ (value_json['pv1']['Power'] | float)}}", "unit_of_meas": "W", "dev": {"name": "Microinverter", "mdl": "MS-600", "ids": ["inverter_123"], "mf": "TSUN", "sa": "", "sw": "0.01", "hw": "Hw0.01"}}) + tests +=1 + + elif id == 'total_gen_123': + assert d_json == json.dumps({"name": "Total Generation", "stat_t": "tsun/garagendach/total", "dev_cla": "energy", "stat_cla": "total", "uniq_id": "total_gen_123", "val_tpl": "{{value_json['Total_Generation'] | float}}", "unit_of_meas": "kWh", "icon": "mdi:solar-power", "dev": {"name": "Microinverter", "mdl": "MS-600", "ids": ["inverter_123"], "mf": "TSUN", "sa": "", "sw": "0.01", "hw": "Hw0.01"}}) + tests +=1 + assert tests==4 + +def test_build_ha_conf3(): + i = Infos() + for d_json, id in i.ha_confs(prfx="tsun/garagendach/", snr='123'): + pass diff --git a/app/tests/test_messages.py b/app/tests/test_messages.py new file mode 100644 index 0000000..65a98a4 --- /dev/null +++ b/app/tests/test_messages.py @@ -0,0 +1,160 @@ +# test_with_pytest.py +import pytest +from app.src.messages import Message, Control +from app.src.config import Config + + +class MemoryStream(Message): + def __init__(self, msg, chunks = (0,)): + super().__init__() + self.__msg = msg + self.__msg_len = len(msg) + self.__chunks = chunks + self.__offs = 0 + self.__chunk_idx = 0 + self.msg_count = 0 + self.server_side = False + self.addr = 'Test: SrvSide' + + def _read(self) -> int: + copied_bytes = 0 + try: + if (self.__offs < self.__msg_len): + len = self.__chunks[self.__chunk_idx] + self.__chunk_idx += 1 + if len!=0: + self._recv_buffer += self.__msg[self.__offs:len] + copied_bytes = len - self.__offs + self.__offs = len + else: + self._recv_buffer += self.__msg[self.__offs:] + copied_bytes = self.__msg_len - self.__offs + self.__offs = self.__msg_len + except: + pass + return copied_bytes + + + def _Message__flush_recv_msg(self) -> None: + super()._Message__flush_recv_msg() + self.msg_count += 1 + return + + def __del__ (self): + super().__del__() + + +@pytest.fixture +def MsgContactInfo(): # Contact Info message + Config.config = {'tsun':{'enabled': True}} + return b'\x00\x00\x00\x2c\x10R170000000000001\x91\x00\x08solarhub\x0fsolarhub\x40123456' + +@pytest.fixture +def MsgContactInfo_LongId(): # Contact Info message with longer ID + Config.config = {'tsun':{'enabled': True}} + return b'\x00\x00\x00\x2d\x11R1700000000000011\x91\x00\x08solarhub\x0fsolarhub\x40123456' + +@pytest.fixture +def Msg2ContactInfo(): # two Contact Info messages + Config.config = {'tsun':{'enabled': True}} + return b'\x00\x00\x00\x2c\x10R170000000000001\x91\x00\x08solarhub\x0fsolarhub\x40123456\x00\x00\x00\x2c\x10R170000000000002\x91\x00\x08solarhub\x0fsolarhub\x40123456' + + + + +def test_read_message(MsgContactInfo): + m = MemoryStream(MsgContactInfo, (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 + assert m.id_str == b"R170000000000001" + assert int(m.ctrl)==145 + assert m.msg_id==0 + assert m.header_len==23 + assert m.data_len==25 + + +def test_read_message_long_id(MsgContactInfo_LongId): + m = MemoryStream(MsgContactInfo_LongId, (23,24)) + m.read() # read 23 bytes, one is missing + assert not m.header_valid # must be invalid, since header not complete + assert m.msg_count == 0 + m.read() # read the missing byte + assert m.header_valid # must be valid, since header is complete but not the msg + assert m.msg_count == 0 + assert m.id_str == b"R1700000000000011" + assert int(m.ctrl)==145 + assert m.msg_id==0 + assert m.header_len==24 + assert m.data_len==25 + m.read() # try to read rest of message, but there is no chunk available + assert m.header_valid # must be valid, since header is complete but not the msg + assert m.msg_count == 0 + + +def test_read_message_in_chunks(MsgContactInfo): + m = MemoryStream(MsgContactInfo, (4,23,0)) + m.read() # read 4 bytes, header incomplere + assert not m.header_valid # must be invalid, since header not complete + assert m.msg_count == 0 + m.read() # read missing bytes for complete header + assert m.header_valid # must be valid, since header is complete but not the msg + assert m.msg_count == 0 + assert m.id_str == b"R170000000000001" + assert int(m.ctrl)==145 + assert m.msg_id==0 + assert m.header_len==23 + assert m.data_len==25 + m.read() # read rest of message + assert not m.header_valid # must be invalid, since msg was handled and buffer flushed + assert m.msg_count == 1 + +def test_read_message_in_chunks2(MsgContactInfo): + m = MemoryStream(MsgContactInfo, (4,10,0)) + m.read() # read 4 bytes, header incomplere + assert not m.header_valid + assert m.msg_count == 0 + m.read() # read 6 more bytes, header incomplere + assert not m.header_valid + assert m.msg_count == 0 + m.read() # read rest of message + assert not m.header_valid # must be invalid, since msg was handled and buffer flushed + assert m.header_len==23 + assert m.data_len==25 + assert m.id_str == b"R170000000000001" + assert int(m.ctrl)==145 + assert m.msg_id==0 + assert m.msg_count == 1 + while m.read(): # read rest of message + pass + assert m.msg_count == 1 + assert not m.header_valid # must be invalid, since msg was handled and buffer flushed + +def test_read_two_messages(Msg2ContactInfo): + m = MemoryStream(Msg2ContactInfo, (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 + assert m.id_str == b"R170000000000001" + assert int(m.ctrl)==145 + assert m.msg_id==0 + assert m.header_len==23 + assert m.data_len==25 + 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 == 2 + assert m.id_str == b"R170000000000002" + assert int(m.ctrl)==145 + assert m.msg_id==0 + assert m.header_len==23 + assert m.data_len==25 + +def test_ctrl_byte(): + c = Control(0x91) + assert c.is_ind() + assert not c.is_resp() + c = Control(0x99) + assert not c.is_ind() + assert c.is_resp() + +