move common info tests form test_infos_g3 to test_infos
This commit is contained in:
Stefan Allius
186
app/tests/test_infos.py
Normal file
186
app/tests/test_infos.py
Normal file
@@ -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
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user