From 2ade04e6ccea508e9e5a0b787067c18d7982495d Mon Sep 17 00:00:00 2001 From: Stefan Allius Date: Sun, 14 Apr 2024 16:01:30 +0200 Subject: [PATCH] move common info tests form test_infos_g3 to test_infos --- app/tests/test_infos.py | 186 +++++++++++++++++++++++++++++++++++++ app/tests/test_infos_g3.py | 186 +------------------------------------ 2 files changed, 187 insertions(+), 185 deletions(-) create mode 100644 app/tests/test_infos.py diff --git a/app/tests/test_infos.py b/app/tests/test_infos.py new file mode 100644 index 0000000..aa0b3fd --- /dev/null +++ b/app/tests/test_infos.py @@ -0,0 +1,186 @@ +# test_with_pytest.py +import pytest, json +from app.src.infos import Register, ClrAtMidnight +from app.src.infos import Infos + +def test_statistic_counter(): + i = Infos() + val = i.dev_value("Test-String") + assert val == "Test-String" + + val = i.dev_value(0xffffffff) # invalid addr + assert val == None + + val = i.dev_value(Register.INVERTER_CNT) # valid addr but not initiliazed + assert val == None or val == 0 + + i.static_init() # initialize counter + assert json.dumps(i.stat) == json.dumps({"proxy": {"Inverter_Cnt": 0, "Unknown_SNR": 0, "Unknown_Msg": 0, "Invalid_Data_Type": 0, "Internal_Error": 0,"Unknown_Ctrl": 0, "OTA_Start_Msg": 0, "SW_Exception": 0, "Invalid_Msg_Format": 0, "AT_Command": 0}}) + + val = i.dev_value(Register.INVERTER_CNT) # valid and initiliazed addr + assert val == 0 + + i.inc_counter('Inverter_Cnt') + assert json.dumps(i.stat) == json.dumps({"proxy": {"Inverter_Cnt": 1, "Unknown_SNR": 0, "Unknown_Msg": 0, "Invalid_Data_Type": 0, "Internal_Error": 0,"Unknown_Ctrl": 0, "OTA_Start_Msg": 0, "SW_Exception": 0, "Invalid_Msg_Format": 0, "AT_Command": 0}}) + val = i.dev_value(Register.INVERTER_CNT) + assert val == 1 + + i.dec_counter('Inverter_Cnt') + val = i.dev_value(Register.INVERTER_CNT) + assert val == 0 + +def test_dep_rules(): + i = Infos() + i.static_init() # initialize counter + + res = i.ignore_this_device({}) + assert res == True + + res = i.ignore_this_device({'reg':0xffffffff}) + assert res == True + + i.inc_counter('Inverter_Cnt') # is 1 + val = i.dev_value(Register.INVERTER_CNT) + assert val == 1 + res = i.ignore_this_device({'reg': Register.INVERTER_CNT}) + assert res == True + res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'less_eq': 2}) + assert res == False + res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'gte': 2}) + assert res == True + + i.inc_counter('Inverter_Cnt') # is 2 + res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'less_eq': 2}) + assert res == False + res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'gte': 2}) + assert res == False + + i.inc_counter('Inverter_Cnt') # is 3 + res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'less_eq': 2}) + assert res == True + res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'gte': 2}) + assert res == False + +def test_table_definition(): + i = Infos() + i.static_init() # initialize counter + + val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter + assert val == 0 + + # for d_json, comp, node_id, id in i.ha_confs(ha_prfx="tsun/", node_id="garagendach/", snr='123', sug_area = 'roof'): + # pass + for reg in Register: + i.ha_conf(reg, ha_prfx="tsun/", node_id="garagendach/", snr='123', singleton=False, sug_area = 'roof') # noqa: E501 + + + for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'): + pass + + val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter + assert val == 0 + + # test missing 'fmt' value + i.info_defs[Register.TEST_REG1] = {'name':['proxy', 'Internal_Test1'], 'singleton': True, 'ha':{'dev':'proxy', 'dev_cla': None, 'stat_cla': None, 'id':'intern_test1_'}} + + tests = 0 + for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'): + if id == 'intern_test1_456': + tests +=1 + + assert tests == 1 + + val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter + assert val == 1 + + # test missing 'dev' value + i.info_defs[Register.TEST_REG1] = {'name':['proxy', 'Internal_Test2'], 'singleton': True, 'ha':{'dev_cla': None, 'stat_cla': None, 'id':'intern_test2_', 'fmt':'| int'}} + tests = 0 + for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'): + if id == 'intern_test2_456': + tests +=1 + + assert tests == 1 + + val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter + assert val == 2 + + + + # test invalid 'via' value + i.info_devs['test_dev'] = {'via':'xyz', 'name':'Module PV1'} + + i.info_defs[Register.TEST_REG1] = {'name':['proxy', 'Internal_Test2'], 'singleton': True, 'ha':{'dev':'test_dev', 'dev_cla': None, 'stat_cla': None, 'id':'intern_test2_', 'fmt':'| int'}} + tests = 0 + for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'): + if id == 'intern_test2_456': + tests +=1 + + assert tests == 1 + + val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter + assert val == 3 + +def test_clr_at_midnight(): + i = Infos() + i.static_init() # initialize counter + i.set_db_def_value(Register.NO_INPUTS, 2) + val = i.dev_value(Register.NO_INPUTS) # valid addr but not initiliazed + assert val == 2 + i.info_defs[Register.TEST_REG1] = { # add a entry with incomplete ha definition + 'name': ['test', 'grp', 'REG_1'], 'ha': {'dev_cla': None } + } + i.reg_clr_at_midnight('tsun/inv_1/') + # tsun/inv_2/input + assert json.dumps(ClrAtMidnight.db['tsun/inv_1/total']) == json.dumps({'Daily_Generation': 0}) + assert json.dumps(ClrAtMidnight.db['tsun/inv_1/input']) == json.dumps({"pv1": {"Daily_Generation": 0}, "pv2": {"Daily_Generation": 0}}) + + i.reg_clr_at_midnight('tsun/inv_1/') + assert json.dumps(ClrAtMidnight.db['tsun/inv_1/total']) == json.dumps({'Daily_Generation': 0}) + assert json.dumps(ClrAtMidnight.db['tsun/inv_1/input']) == json.dumps({"pv1": {"Daily_Generation": 0}, "pv2": {"Daily_Generation": 0}}) + + test = 0 + for key, data in ClrAtMidnight.elm(): + if key == 'tsun/inv_1/total': + assert json.dumps(data) == json.dumps({'Daily_Generation': 0}) + test += 1 + elif key == 'tsun/inv_1/input': + assert json.dumps(data) == json.dumps({"pv1": {"Daily_Generation": 0}, "pv2": {"Daily_Generation": 0}}) + test += 1 + assert test == 2 + assert json.dumps(ClrAtMidnight.db) == json.dumps({}) + + i.reg_clr_at_midnight('tsun/inv_1/') + +def test_pv_module_config(): + i = Infos() + # i.set_db_def_value(Register.NO_INPUTS, 2) + + dt = { + 'pv1':{'manufacturer':'TSUN1','type': 'Module 100W'}, + 'pv2':{'manufacturer':'TSUN2'}, + 'pv3':{'manufacturer':'TSUN3','type': 'Module 300W'}, + 'pv4':{'type': 'Module 400W'}, + 'pv5':{}, + } + i.set_pv_module_details(dt) + assert 'TSUN1' == i.dev_value(Register.PV1_MANUFACTURER) + assert 'TSUN2' == i.dev_value(Register.PV2_MANUFACTURER) + assert 'TSUN3' == i.dev_value(Register.PV3_MANUFACTURER) + assert None == i.dev_value(Register.PV4_MANUFACTURER) + assert None == i.dev_value(Register.PV5_MANUFACTURER) + assert 'Module 100W' == i.dev_value(Register.PV1_MODEL) + assert None == i.dev_value(Register.PV2_MODEL) + assert 'Module 300W' == i.dev_value(Register.PV3_MODEL) + assert 'Module 400W' == i.dev_value(Register.PV4_MODEL) + assert None == i.dev_value(Register.PV5_MODEL) + +def test_broken_info_defs(): + i = Infos() + val = i.get_db_value(Register.NO_INPUTS, 666) + assert val == 666 + i.info_defs[Register.TEST_REG1] = 'test' # add a string instead of a dict + val = i.get_db_value(Register.TEST_REG1, 666) + assert val == 666 + i.set_db_def_value(Register.TEST_REG1, 2) + del i.info_defs[Register.TEST_REG1] # delete the broken entry diff --git a/app/tests/test_infos_g3.py b/app/tests/test_infos_g3.py index 8bf419d..193536e 100644 --- a/app/tests/test_infos_g3.py +++ b/app/tests/test_infos_g3.py @@ -390,124 +390,7 @@ def test_must_incr_total2(InvDataSeq2, InvDataSeq2_Zero): assert tests==3 assert json.dumps(i.db['total']) == json.dumps({'Daily_Generation': 1.7, 'Total_Generation': 17.36}) assert json.dumps(i.db['input']) == json.dumps({"pv1": {"Voltage": 33.6, "Current": 1.91, "Power": 64.5, "Daily_Generation": 1.08, "Total_Generation": 9.74}, "pv2": {"Voltage": 33.5, "Current": 1.36, "Power": 45.7, "Daily_Generation": 0.62, "Total_Generation": 7.62}, "pv3": {"Voltage": 0.0, "Current": 0.0, "Power": 0.0}, "pv4": {"Voltage": 0.0, "Current": 0.0, "Power": 0.0}}) - assert json.dumps(i.db['env']) == json.dumps({"Inverter_Temp": 23}) - - -def test_statistic_counter(): - i = InfosG3() - val = i.dev_value("Test-String") - assert val == "Test-String" - - val = i.dev_value(0xffffffff) # invalid addr - assert val == None - - val = i.dev_value(Register.INVERTER_CNT) # valid addr but not initiliazed - assert val == None or val == 0 - - i.static_init() # initialize counter - assert json.dumps(i.stat) == json.dumps({"proxy": {"Inverter_Cnt": 0, "Unknown_SNR": 0, "Unknown_Msg": 0, "Invalid_Data_Type": 0, "Internal_Error": 0,"Unknown_Ctrl": 0, "OTA_Start_Msg": 0, "SW_Exception": 0, "Invalid_Msg_Format": 0, "AT_Command": 0}}) - - val = i.dev_value(Register.INVERTER_CNT) # valid and initiliazed addr - assert val == 0 - - i.inc_counter('Inverter_Cnt') - assert json.dumps(i.stat) == json.dumps({"proxy": {"Inverter_Cnt": 1, "Unknown_SNR": 0, "Unknown_Msg": 0, "Invalid_Data_Type": 0, "Internal_Error": 0,"Unknown_Ctrl": 0, "OTA_Start_Msg": 0, "SW_Exception": 0, "Invalid_Msg_Format": 0, "AT_Command": 0}}) - val = i.dev_value(Register.INVERTER_CNT) - assert val == 1 - - i.dec_counter('Inverter_Cnt') - val = i.dev_value(Register.INVERTER_CNT) - assert val == 0 - -def test_dep_rules(): - i = InfosG3() - i.static_init() # initialize counter - - res = i.ignore_this_device({}) - assert res == True - - res = i.ignore_this_device({'reg':0xffffffff}) - assert res == True - - i.inc_counter('Inverter_Cnt') # is 1 - val = i.dev_value(Register.INVERTER_CNT) - assert val == 1 - res = i.ignore_this_device({'reg': Register.INVERTER_CNT}) - assert res == True - res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'less_eq': 2}) - assert res == False - res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'gte': 2}) - assert res == True - - i.inc_counter('Inverter_Cnt') # is 2 - res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'less_eq': 2}) - assert res == False - res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'gte': 2}) - assert res == False - - i.inc_counter('Inverter_Cnt') # is 3 - res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'less_eq': 2}) - assert res == True - res = i.ignore_this_device({'reg': Register.INVERTER_CNT, 'gte': 2}) - assert res == False - -def test_table_definition(): - i = InfosG3() - i.static_init() # initialize counter - - val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter - assert val == 0 - - for d_json, comp, node_id, id in i.ha_confs(ha_prfx="tsun/", node_id="garagendach/", snr='123', sug_area = 'roof'): - pass - - for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'): - pass - - val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter - assert val == 0 - - # test missing 'fmt' value - i.info_defs[Register.TEST_REG1] = {'name':['proxy', 'Internal_Test1'], 'singleton': True, 'ha':{'dev':'proxy', 'dev_cla': None, 'stat_cla': None, 'id':'intern_test1_'}} - - tests = 0 - for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'): - if id == 'intern_test1_456': - tests +=1 - - assert tests == 1 - - val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter - assert val == 1 - - # test missing 'dev' value - i.info_defs[Register.TEST_REG1] = {'name':['proxy', 'Internal_Test2'], 'singleton': True, 'ha':{'dev_cla': None, 'stat_cla': None, 'id':'intern_test2_', 'fmt':'| int'}} - tests = 0 - for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'): - if id == 'intern_test2_456': - tests +=1 - - assert tests == 1 - - val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter - assert val == 2 - - - - # test invalid 'via' value - i.info_devs['test_dev'] = {'via':'xyz', 'name':'Module PV1'} - - i.info_defs[Register.TEST_REG1] = {'name':['proxy', 'Internal_Test2'], 'singleton': True, 'ha':{'dev':'test_dev', 'dev_cla': None, 'stat_cla': None, 'id':'intern_test2_', 'fmt':'| int'}} - tests = 0 - for d_json, comp, node_id, id in i.ha_proxy_confs(ha_prfx="tsun/", node_id = 'proxy/', snr = '456'): - if id == 'intern_test2_456': - tests +=1 - - assert tests == 1 - - val = i.dev_value(Register.INTERNAL_ERROR) # check internal error counter - assert val == 3 - + def test_invalid_data_type(InvalidDataSeq): i = InfosG3() @@ -523,70 +406,3 @@ def test_invalid_data_type(InvalidDataSeq): val = i.dev_value(Register.INVALID_DATA_TYPE) # check invalid data type counter assert val == 1 - -def test_clr_at_midnight(): - i = InfosG3() - i.static_init() # initialize counter - i.set_db_def_value(Register.NO_INPUTS, 2) - val = i.dev_value(Register.NO_INPUTS) # valid addr but not initiliazed - assert val == 2 - i.info_defs[Register.TEST_REG1] = { # add a entry with incomplete ha definition - 'name': ['test', 'grp', 'REG_1'], 'ha': {'dev_cla': None } - } - i.reg_clr_at_midnight('tsun/inv_1/') - # tsun/inv_2/input - assert json.dumps(ClrAtMidnight.db['tsun/inv_1/total']) == json.dumps({'Daily_Generation': 0}) - assert json.dumps(ClrAtMidnight.db['tsun/inv_1/input']) == json.dumps({"pv1": {"Daily_Generation": 0}, "pv2": {"Daily_Generation": 0}}) - - i.reg_clr_at_midnight('tsun/inv_1/') - assert json.dumps(ClrAtMidnight.db['tsun/inv_1/total']) == json.dumps({'Daily_Generation': 0}) - assert json.dumps(ClrAtMidnight.db['tsun/inv_1/input']) == json.dumps({"pv1": {"Daily_Generation": 0}, "pv2": {"Daily_Generation": 0}}) - - test = 0 - for key, data in ClrAtMidnight.elm(): - if key == 'tsun/inv_1/total': - assert json.dumps(data) == json.dumps({'Daily_Generation': 0}) - test += 1 - elif key == 'tsun/inv_1/input': - assert json.dumps(data) == json.dumps({"pv1": {"Daily_Generation": 0}, "pv2": {"Daily_Generation": 0}}) - test += 1 - assert test == 2 - assert json.dumps(ClrAtMidnight.db) == json.dumps({}) - - i.reg_clr_at_midnight('tsun/inv_1/') - - - - -def test_pv_module_config(): - i = InfosG3() - # i.set_db_def_value(Register.NO_INPUTS, 2) - - dt = { - 'pv1':{'manufacturer':'TSUN1','type': 'Module 100W'}, - 'pv2':{'manufacturer':'TSUN2'}, - 'pv3':{'manufacturer':'TSUN3','type': 'Module 300W'}, - 'pv4':{'type': 'Module 400W'}, - 'pv5':{}, - } - i.set_pv_module_details(dt) - assert 'TSUN1' == i.dev_value(Register.PV1_MANUFACTURER) - assert 'TSUN2' == i.dev_value(Register.PV2_MANUFACTURER) - assert 'TSUN3' == i.dev_value(Register.PV3_MANUFACTURER) - assert None == i.dev_value(Register.PV4_MANUFACTURER) - assert None == i.dev_value(Register.PV5_MANUFACTURER) - assert 'Module 100W' == i.dev_value(Register.PV1_MODEL) - assert None == i.dev_value(Register.PV2_MODEL) - assert 'Module 300W' == i.dev_value(Register.PV3_MODEL) - assert 'Module 400W' == i.dev_value(Register.PV4_MODEL) - assert None == i.dev_value(Register.PV5_MODEL) - -def test_broken_info_defs(): - i = InfosG3() - val = i.get_db_value(Register.NO_INPUTS, 666) - assert val == 666 - i.info_defs[Register.TEST_REG1] = 'test' # add a string instead of a dict - val = i.get_db_value(Register.TEST_REG1, 666) - assert val == 666 - i.set_db_def_value(Register.TEST_REG1, 2) -