make test code more clean (#148)

This commit is contained in:
Stefan Allius
2024-08-08 20:27:39 +02:00
committed by GitHub
parent a779c90965
commit 24ece4fece
3 changed files with 28 additions and 32 deletions

View File

@@ -1,6 +1,6 @@
# test_with_pytest.py
import pytest
import json
import json, math
import logging
from app.src.infos import Register, ClrAtMidnight
from app.src.infos import Infos
@@ -77,7 +77,7 @@ def test_table_definition():
for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'):
pass
pass # sideeffect is calling generator i.ha_proxy_confs()
val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter
assert val == 0
@@ -222,24 +222,24 @@ def test_get_value():
i.set_db_def_value(Register.PV2_VOLTAGE, 30.3)
assert 30 == i.get_db_value(Register.PV1_VOLTAGE, None)
assert 30.3 == i.get_db_value(Register.PV2_VOLTAGE, None)
assert math.isclose(30.3,i.get_db_value(Register.PV2_VOLTAGE, None), rel_tol=1e-09, abs_tol=1e-09)
def test_update_value():
i = Infos()
assert None == i.get_db_value(Register.PV1_VOLTAGE, None)
keys = i.info_defs[Register.PV1_VOLTAGE]['name']
name, update = i.update_db(keys, True, 30)
_, update = i.update_db(keys, True, 30)
assert update == True
assert 30 == i.get_db_value(Register.PV1_VOLTAGE, None)
keys = i.info_defs[Register.PV1_VOLTAGE]['name']
name, update = i.update_db(keys, True, 30)
_, update = i.update_db(keys, True, 30)
assert update == False
assert 30 == i.get_db_value(Register.PV1_VOLTAGE, None)
keys = i.info_defs[Register.PV1_VOLTAGE]['name']
name, update = i.update_db(keys, False, 29)
_, update = i.update_db(keys, False, 29)
assert update == True
assert 29 == i.get_db_value(Register.PV1_VOLTAGE, None)

View File

@@ -1,12 +1,12 @@
# test_with_pytest.py
import pytest, json
import pytest, json, math
from app.src.infos import Register
from app.src.gen3plus.infos_g3p import InfosG3P
from app.src.gen3plus.infos_g3p import RegisterMap
@pytest.fixture
def DeviceData(): # 0x4110 ftype: 0x02
def device_data(): # 0x4110 ftype: 0x02
msg = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\xba\xd2\x00\x00'
msg += b'\x19\x00\x00\x00\x00\x00\x00\x00\x05\x3c\x78\x01\x64\x01\x4c\x53'
msg += b'\x57\x35\x42\x4c\x45\x5f\x31\x37\x5f\x30\x32\x42\x30\x5f\x31\x2e'
@@ -24,7 +24,7 @@ def DeviceData(): # 0x4110 ftype: 0x02
return msg
@pytest.fixture
def InverterData(): # 0x4210 ftype: 0x01
def inverter_data(): # 0x4210 ftype: 0x01
msg = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\xb0\x02\xbc\xc8'
msg += b'\x24\x32\x6c\x1f\x00\x00\xa0\x47\xe4\x33\x01\x00\x03\x08\x00\x00'
msg += b'\x59\x31\x37\x45\x30\x30\x30\x30\x30\x30\x30\x30\x30\x30\x30\x45'
@@ -63,23 +63,23 @@ def test_default_db():
"collector": {"Chip_Type": "IGEN TECH"},
})
def test_parse_4110(DeviceData: bytes):
def test_parse_4110(device_data: bytes):
i = InfosG3P(client_mode=False)
i.db.clear()
for key, update in i.parse (DeviceData, 0x41, 2):
pass
for key, update in i.parse (device_data, 0x41, 2):
pass # side effect is calling generator i.parse()
assert json.dumps(i.db) == json.dumps({
'controller': {"Data_Up_Interval": 300, "Collect_Interval": 1, "Heartbeat_Interval": 120, "Signal_Strength": 100, "IP_Address": "192.168.80.49"},
'collector': {"Chip_Model": "LSW5BLE_17_02B0_1.05", "Collector_Fw_Version": "V1.1.00.0B"},
})
def test_parse_4210(InverterData: bytes):
def test_parse_4210(inverter_data: bytes):
i = InfosG3P(client_mode=False)
i.db.clear()
for key, update in i.parse (InverterData, 0x42, 1):
pass
for key, update in i.parse (inverter_data, 0x42, 1):
pass # side effect is calling generator i.parse()
assert json.dumps(i.db) == json.dumps({
"controller": {"Power_On_Time": 2051},
@@ -233,28 +233,27 @@ def test_build_ha_conf2():
assert tests==8
def test_exception_and_eval(InverterData: bytes):
def test_exception_and_eval(inverter_data: bytes):
# add eval to convert temperature from °F to °C
RegisterMap.map[0x420100d8]['eval'] = '(result-32)/1.8'
# map PV1_VOLTAGE to invalid register
RegisterMap.map[0x420100e0]['reg'] = Register.TEST_REG2
# set invalid maping entry for OUTPUT_POWER (string instead of dict type)
Backup = RegisterMap.map[0x420100de]
backup = RegisterMap.map[0x420100de]
RegisterMap.map[0x420100de] = 'invalid_entry'
i = InfosG3P(client_mode=False)
# i.db.clear()
for key, update in i.parse (InverterData, 0x42, 1):
pass
assert 12.2222 == round (i.get_db_value(Register.INVERTER_TEMP, 0),4)
for key, update in i.parse (inverter_data, 0x42, 1):
pass # side effect is calling generator i.parse()
assert math.isclose(12.2222, round (i.get_db_value(Register.INVERTER_TEMP, 0),4), rel_tol=1e-09, abs_tol=1e-09)
del RegisterMap.map[0x420100d8]['eval'] # remove eval
RegisterMap.map[0x420100e0]['reg'] = Register.PV1_VOLTAGE # reset mapping
RegisterMap.map[0x420100de] = Backup # reset mapping
RegisterMap.map[0x420100de] = backup # reset mapping
for key, update in i.parse (InverterData, 0x42, 1):
pass
for key, update in i.parse (inverter_data, 0x42, 1):
pass # side effect is calling generator i.parse()
assert 54 == i.get_db_value(Register.INVERTER_TEMP, 0)

View File

@@ -5,7 +5,6 @@ from app.src.modbus import Modbus
from app.src.infos import Infos, Register
pytest_plugins = ('pytest_asyncio',)
# pytestmark = pytest.mark.asyncio(scope="module")
class ModbusTestHelper(Modbus):
def __init__(self):
@@ -76,8 +75,8 @@ def test_recv_resp_crc_err():
mb.req_pend = True
mb.last_addr = 1
mb.last_fcode = 3
mb.last_reg == 0x300e
mb.last_len == 2
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'):
@@ -96,8 +95,8 @@ def test_recv_resp_invalid_addr():
# simulate a transmitted request
mb.last_addr = 1
mb.last_fcode = 3
mb.last_reg == 0x300e
mb.last_len == 2
mb.last_reg = 0x300e
mb.last_len = 2
# check not matching response, with wrong server addr
call = 0
@@ -298,7 +297,7 @@ def test_queue3():
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
pass # no code in loop is OK; calling the generator is the purpose
assert 0 == mb.err
assert mb.recv_responses == 2
@@ -364,8 +363,6 @@ async def test_timeout():
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 = ModbusTestHelper()