sv/tsun-gen3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SonarCloud setup (#168)

Browse Source 
* Code Cleanup (#158)

* print coverage report

* create sonar-project property file

* install all py dependencies in one step

* code cleanup

* reduce cognitive complexity

* do not build on *.yml changes

* optimise versionstring handling (#159)

- Reading the version string from the image updates
  it even if the image is re-pulled without re-deployment

* fix linter warning

* exclude *.pyi filese

* ignore some rules for tests

* cleanup (#160)

* Sonar qube 3 (#163)

fix SonarQube warnings in modbus.py

* Sonar qube 3 (#164)


* fix SonarQube warnings

* Sonar qube 3 (#165)

* cleanup

* Add support for TSUN Titan inverter
Fixes #161


* fix SonarQube warnings

* fix error

* rename field "config"

* SonarQube reads flake8 output

* don't stop on flake8 errors

* flake8 scan only app/src for SonarQube

* update flake8 run

* ignore flake8 C901

* cleanup

* fix linter warnings

* ignore changed *.yml files

* read sensor list solarman data packets

* catch 'No route to' error and log only in debug mode

* fix unit tests

* add sensor_list configuration

* adapt unit tests

* fix SonarQube warnings

* Sonar qube 3 (#166)

* add unittests for mqtt.py

* add mock

* move test requirements into a file

* fix unit tests

* fix formating

* initial version

* fix SonarQube warning
This commit is contained in:
Stefan Allius
2024-08-23 21:24:01 +02:00
committed by GitHub
parent db1169f61f
commit 58b42f7d7c
27 changed files with 899 additions and 500 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
app/tests/test_config.py
View File

@@ -7,11 +7,11 @@ class TstConfig(Config):
@classmethod
def set(cls, cnf):
cls.config = cnf
cls.act_config = cnf
@classmethod
def _read_config_file(cls) -> dict:
return cls.config
return cls.act_config
def test_empty_config():
@@ -30,7 +30,7 @@ def test_default_config():
validated = Config.conf_schema.validate(cnf)
except Exception:
assert False
assert validated == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'node_id': '', 'modbus_polling': False, 'monitor_sn': 0, 'suggested_area': ''}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'node_id': '', 'suggested_area': ''}}}
assert validated == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'node_id': '', 'modbus_polling': False, 'monitor_sn': 0, 'suggested_area': '', 'sensor_list': 688}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'node_id': '', 'suggested_area': '', 'sensor_list': 688}}}
def test_full_config():
cnf = {'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005},
@@ -40,13 +40,13 @@ def test_full_config():
'mqtt': {'host': 'mqtt', 'port': 1883, 'user': '', 'passwd': ''},
'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'},
'inverters': {'allow_all': True,
'R170000000000001': {'modbus_polling': True, 'node_id': '', 'suggested_area': '', 'pv1': {'type': 'type1', 'manufacturer': 'man1'}, 'pv2': {'type': 'type2', 'manufacturer': 'man2'}, 'pv3': {'type': 'type3', 'manufacturer': 'man3'}},
'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'node_id': '', 'suggested_area': ''}}}
'R170000000000001': {'modbus_polling': True, 'node_id': '', 'sensor_list': 0, 'suggested_area': '', 'pv1': {'type': 'type1', 'manufacturer': 'man1'}, 'pv2': {'type': 'type2', 'manufacturer': 'man2'}, 'pv3': {'type': 'type3', 'manufacturer': 'man3'}},
'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'node_id': '', 'sensor_list': 0x1511, 'suggested_area': ''}}}
try:
validated = Config.conf_schema.validate(cnf)
except Exception:
assert False
assert validated == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'node_id': '', 'modbus_polling': True, 'monitor_sn': 0, 'pv1': {'manufacturer': 'man1','type': 'type1'},'pv2': {'manufacturer': 'man2','type': 'type2'},'pv3': {'manufacturer': 'man3','type': 'type3'}, 'suggested_area': ''}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'node_id': '', 'suggested_area': ''}}}
assert validated == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'node_id': '', 'modbus_polling': True, 'monitor_sn': 0, 'pv1': {'manufacturer': 'man1','type': 'type1'},'pv2': {'manufacturer': 'man2','type': 'type2'},'pv3': {'manufacturer': 'man3','type': 'type3'}, 'suggested_area': '', 'sensor_list': 0}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'node_id': '', 'suggested_area': '', 'sensor_list': 5393}}}
def test_mininum_config():
cnf = {'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005},
@@ -63,7 +63,7 @@ def test_mininum_config():
validated = Config.conf_schema.validate(cnf)
except Exception:
assert False
assert validated == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'node_id': '', 'modbus_polling': True, 'monitor_sn': 0, 'suggested_area': ''}}}
assert validated == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'node_id': '', 'modbus_polling': True, 'monitor_sn': 0, 'suggested_area': '', 'sensor_list': 688}}}
def test_read_empty():
cnf = {}
@@ -71,7 +71,7 @@ def test_read_empty():
err = TstConfig.read('app/config/')
assert err == None
cnf = TstConfig.get()
assert cnf == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'suggested_area': '', 'modbus_polling': False, 'monitor_sn': 0, 'node_id': ''}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': ''}}}
assert cnf == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'suggested_area': '', 'modbus_polling': False, 'monitor_sn': 0, 'node_id': '', 'sensor_list': 688}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': '', 'sensor_list': 688}}}
defcnf = TstConfig.def_config.get('solarman')
assert defcnf == {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}
@@ -93,7 +93,7 @@ def test_read_cnf1():
err = TstConfig.read('app/config/')
assert err == None
cnf = TstConfig.get()
assert cnf == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': False, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'suggested_area': '', 'modbus_polling': False, 'monitor_sn': 0, 'node_id': ''}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': ''}}}
assert cnf == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': False, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'suggested_area': '', 'modbus_polling': False, 'monitor_sn': 0, 'node_id': '', 'sensor_list': 688}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': '', 'sensor_list': 688}}}
cnf = TstConfig.get('solarman')
assert cnf == {'enabled': False, 'host': 'iot.talent-monitoring.com', 'port': 10000}
defcnf = TstConfig.def_config.get('solarman')
@@ -106,7 +106,7 @@ def test_read_cnf2():
err = TstConfig.read('app/config/')
assert err == None
cnf = TstConfig.get()
assert cnf == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'suggested_area': '', 'modbus_polling': False, 'monitor_sn': 0, 'node_id': ''}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': ''}}}
assert cnf == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'suggested_area': '', 'modbus_polling': False, 'monitor_sn': 0, 'node_id': '', 'sensor_list': 688}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': '', 'sensor_list': 688}}}
assert True == TstConfig.is_default('solarman')
def test_read_cnf3():
@@ -123,7 +123,7 @@ def test_read_cnf4():
err = TstConfig.read('app/config/')
assert err == None
cnf = TstConfig.get()
assert cnf == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 5000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'suggested_area': '', 'modbus_polling': False, 'monitor_sn': 0, 'node_id': ''}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': ''}}}
assert cnf == {'gen3plus': {'at_acl': {'mqtt': {'allow': ['AT+'], 'block': []}, 'tsun': {'allow': ['AT+Z', 'AT+UPURL', 'AT+SUPDATE'], 'block': []}}}, 'tsun': {'enabled': True, 'host': 'logger.talent-monitoring.com', 'port': 5005}, 'solarman': {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 5000}, 'mqtt': {'host': 'mqtt', 'port': 1883, 'user': None, 'passwd': None}, 'ha': {'auto_conf_prefix': 'homeassistant', 'discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun', 'proxy_node_id': 'proxy', 'proxy_unique_id': 'P170000000000001'}, 'inverters': {'allow_all': True, 'R170000000000001': {'suggested_area': '', 'modbus_polling': False, 'monitor_sn': 0, 'node_id': '', 'sensor_list': 688}, 'Y170000000000001': {'modbus_polling': True, 'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': '', 'sensor_list': 688}}}
assert False == TstConfig.is_default('solarman')
def test_read_cnf5():

84
app/tests/test_infos_g3.py
View File

@@ -4,7 +4,7 @@ from app.src.infos import Register, ClrAtMidnight
from app.src.gen3.infos_g3 import InfosG3
@pytest.fixture
def ContrDataSeq(): # Get Time Request message
def contr_data_seq(): # 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'
@@ -14,7 +14,7 @@ def ContrDataSeq(): # Get Time Request message
return msg
@pytest.fixture
def Contr2DataSeq(): # Get Time Request message
def contr2_data_seq(): # Get Time Request message
msg = b'\x00\x00\x00\x39\x00\x09\x2b\xa8\x54\x10\x52'
msg += b'\x53\x57\x5f\x34\x30\x30\x5f\x56\x31\x2e\x30\x30\x2e\x32\x30\x00'
msg += b'\x09\x27\xc0\x54\x06\x52\x61\x79\x6d\x6f\x6e\x00\x09\x2f\x90\x54'
@@ -94,19 +94,19 @@ def Contr2DataSeq(): # Get Time Request message
return msg
@pytest.fixture
def InvDataSeq(): # Data indication from the controller
def inv_data_seq(): # Data indication from the controller
msg = b'\x00\x00\x00\x06\x00\x00\x00\x0a\x54\x08\x4d\x69\x63\x72\x6f\x69\x6e\x76\x00\x00\x00\x14\x54\x04\x54\x53\x55\x4e\x00\x00\x00\x1E\x54\x07\x56\x35\x2e\x30\x2e\x31\x31\x00\x00\x00\x28'
msg += b'\x54\x10T170000000000001\x00\x00\x00\x32\x54\x0a\x54\x53\x4f\x4c\x2d\x4d\x53\x36\x30\x30\x00\x00\x00\x3c\x54\x05\x41\x2c\x42\x2c\x43'
return msg
@pytest.fixture
def InvalidDataSeq(): # Data indication from the controller
def invalid_data_seq(): # Data indication from the controller
msg = b'\x00\x00\x00\x06\x00\x00\x00\x0a\x54\x08\x4d\x69\x63\x72\x6f\x69\x6e\x76\x00\x00\x00\x14\x64\x04\x54\x53\x55\x4e\x00\x00\x00\x1E\x54\x07\x56\x35\x2e\x30\x2e\x31\x31\x00\x00\x00\x28'
msg += b'\x54\x10T170000000000001\x00\x00\x00\x32\x54\x0a\x54\x53\x4f\x4c\x2d\x4d\x53\x36\x30\x30\x00\x00\x00\x3c\x54\x05\x41\x2c\x42\x2c\x43'
return msg
@pytest.fixture
def InvDataSeq2(): # Data indication from the controller
def inv_data_seq2(): # Data indication from the controller
msg = b'\x00\x00\x00\xa3\x00\x00\x00\x64\x53\x00\x01\x00\x00\x00\xc8\x53\x00\x02\x00\x00\x01\x2c\x53\x00\x00\x00\x00\x01\x90\x49\x00\x00\x00\x00\x00\x00\x01\x91\x53\x00\x00'
msg += b'\x00\x00\x01\x92\x53\x00\x00\x00\x00\x01\x93\x53\x00\x00\x00\x00\x01\x94\x53\x00\x00\x00\x00\x01\x95\x53\x00\x00\x00\x00\x01\x96\x53\x00\x00\x00\x00\x01\x97\x53\x00'
msg += b'\x00\x00\x00\x01\x98\x53\x00\x00\x00\x00\x01\x99\x53\x00\x00\x00\x00\x01\x9a\x53\x00\x00\x00\x00\x01\x9b\x53\x00\x00\x00\x00\x01\x9c\x53\x00\x00\x00\x00\x01\x9d\x53'
@@ -141,7 +141,7 @@ def InvDataSeq2(): # Data indication from the controller
return msg
@pytest.fixture
def InvDataNew(): # Data indication from DSP V5.0.17
def inv_data_new(): # Data indication from DSP V5.0.17
msg = b'\x00\x00\x00\xa3\x00\x00\x00\x00\x53\x00\x00'
msg += b'\x00\x00\x00\x80\x53\x00\x00\x00\x00\x01\x04\x53\x00\x00\x00\x00'
msg += b'\x01\x90\x41\x00\x00\x01\x91\x53\x00\x00\x00\x00\x01\x90\x53\x00'
@@ -217,7 +217,7 @@ def InvDataNew(): # Data indication from DSP V5.0.17
return msg
@pytest.fixture
def InvDataSeq2_Zero(): # Data indication from the controller
def inv_data_seq2_zero(): # Data indication from the controller
msg = b'\x00\x00\x00\xa3\x00\x00\x00\x64\x53\x00\x01\x00\x00\x00\xc8\x53\x00\x02\x00\x00\x01\x2c\x53\x00\x00\x00\x00\x01\x90\x49\x00\x00\x00\x00\x00\x00\x01\x91\x53\x00\x00'
msg += b'\x00\x00\x01\x92\x53\x00\x00\x00\x00\x01\x93\x53\x00\x00\x00\x00\x01\x94\x53\x00\x00\x00\x00\x01\x95\x53\x00\x00\x00\x00\x01\x96\x53\x00\x00\x00\x00\x01\x97\x53\x00'
msg += b'\x00\x00\x00\x01\x98\x53\x00\x00\x00\x00\x01\x99\x53\x00\x00\x00\x00\x01\x9a\x53\x00\x00\x00\x00\x01\x9b\x53\x00\x00\x00\x00\x01\x9c\x53\x00\x00\x00\x00\x01\x9d\x53'
@@ -252,37 +252,37 @@ def InvDataSeq2_Zero(): # Data indication from the controller
return msg
def test_parse_control(ContrDataSeq):
def test_parse_control(contr_data_seq):
i = InfosG3()
for key, result in i.parse (ContrDataSeq):
pass
for key, result in i.parse (contr_data_seq):
pass # side effect in calling i.parse()
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"}, "controller": {"Collect_Interval": 1, "Signal_Strength": 100, "Power_On_Time": 29, "Communication_Type": 1, "Connect_Count": 1, "Data_Up_Interval": 300}})
def test_parse_control2(Contr2DataSeq):
def test_parse_control2(contr2_data_seq):
i = InfosG3()
for key, result in i.parse (Contr2DataSeq):
pass
for key, result in i.parse (contr2_data_seq):
pass # side effect in calling i.parse()
assert json.dumps(i.db) == json.dumps(
{"collector": {"Collector_Fw_Version": "RSW_400_V1.00.20", "Chip_Type": "Raymon", "Chip_Model": "RSW-1-10001", "Trace_URL": "t.raymoniot.com", "Logger_URL": "logger.talent-monitoring.com"}, "controller": {"Collect_Interval": 1, "Signal_Strength": 16, "Power_On_Time": 334, "Communication_Type": 1, "Connect_Count": 1, "Data_Up_Interval": 300}})
def test_parse_inverter(InvDataSeq):
def test_parse_inverter(inv_data_seq):
i = InfosG3()
for key, result in i.parse (InvDataSeq):
pass
for key, result in i.parse (inv_data_seq):
pass # side effect in calling i.parse()
assert json.dumps(i.db) == json.dumps(
{"inverter": {"Product_Name": "Microinv", "Manufacturer": "TSUN", "Version": "V5.0.11", "Serial_Number": "T170000000000001", "Equipment_Model": "TSOL-MS600"}})
def test_parse_cont_and_invert(ContrDataSeq, InvDataSeq):
def test_parse_cont_and_invert(contr_data_seq, inv_data_seq):
i = InfosG3()
for key, result in i.parse (ContrDataSeq):
pass
for key, result in i.parse (contr_data_seq):
pass # side effect in calling i.parse()
for key, result in i.parse (InvDataSeq):
pass
for key, result in i.parse (inv_data_seq):
pass # side effect in calling i.parse()
assert json.dumps(i.db) == json.dumps(
{
@@ -290,7 +290,7 @@ def test_parse_cont_and_invert(ContrDataSeq, InvDataSeq):
"inverter": {"Product_Name": "Microinv", "Manufacturer": "TSUN", "Version": "V5.0.11", "Serial_Number": "T170000000000001", "Equipment_Model": "TSOL-MS600"}})
def test_build_ha_conf1(ContrDataSeq):
def test_build_ha_conf1(contr_data_seq):
i = InfosG3()
i.static_init() # initialize counter
@@ -346,14 +346,14 @@ def test_build_ha_conf1(ContrDataSeq):
assert tests==5
def test_build_ha_conf2(ContrDataSeq, InvDataSeq, InvDataSeq2):
def test_build_ha_conf2(contr_data_seq, inv_data_seq, inv_data_seq2):
i = InfosG3()
for key, result in i.parse (ContrDataSeq):
pass
for key, result in i.parse (InvDataSeq):
pass
for key, result in i.parse (InvDataSeq2):
pass
for key, result in i.parse (contr_data_seq):
pass # side effect in calling i.parse()
for key, result in i.parse (inv_data_seq):
pass # side effect in calling i.parse()
for key, result in i.parse (inv_data_seq2):
pass # side effect in calling i.parse()
tests = 0
for d_json, comp, node_id, id in i.ha_confs(ha_prfx="tsun/", node_id="garagendach/", snr='123', sug_area = 'roof'):
@@ -384,10 +384,10 @@ def test_build_ha_conf2(ContrDataSeq, InvDataSeq, InvDataSeq2):
tests +=1
assert tests==5
def test_must_incr_total(InvDataSeq2, InvDataSeq2_Zero):
def test_must_incr_total(inv_data_seq2, inv_data_seq2_zero):
i = InfosG3()
tests = 0
for key, update in i.parse (InvDataSeq2):
for key, update in i.parse (inv_data_seq2):
if key == 'total' or key == 'inverter' or key == 'env':
assert update == True
tests +=1
@@ -396,7 +396,7 @@ def test_must_incr_total(InvDataSeq2, InvDataSeq2_Zero):
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})
tests = 0
for key, update in i.parse (InvDataSeq2):
for key, update in i.parse (inv_data_seq2):
if key == 'total':
assert update == False
tests +=1
@@ -411,7 +411,7 @@ def test_must_incr_total(InvDataSeq2, InvDataSeq2_Zero):
assert json.dumps(i.db['inverter']) == json.dumps({"Rated_Power": 600, "No_Inputs": 2})
tests = 0
for key, update in i.parse (InvDataSeq2_Zero):
for key, update in i.parse (inv_data_seq2_zero):
if key == 'total':
assert update == False
tests +=1
@@ -424,10 +424,10 @@ def test_must_incr_total(InvDataSeq2, InvDataSeq2_Zero):
assert json.dumps(i.db['input']) == json.dumps({"pv1": {"Voltage": 33.6, "Current": 1.91, "Power": 0.0, "Daily_Generation": 1.08, "Total_Generation": 9.74}, "pv2": {"Voltage": 33.5, "Current": 1.36, "Power": 0.0, "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": 0})
def test_must_incr_total2(InvDataSeq2, InvDataSeq2_Zero):
def test_must_incr_total2(inv_data_seq2, inv_data_seq2_zero):
i = InfosG3()
tests = 0
for key, update in i.parse (InvDataSeq2_Zero):
for key, update in i.parse (inv_data_seq2_zero):
if key == 'total':
assert update == False
tests +=1
@@ -441,7 +441,7 @@ def test_must_incr_total2(InvDataSeq2, InvDataSeq2_Zero):
assert json.dumps(i.db['env']) == json.dumps({"Inverter_Temp": 0})
tests = 0
for key, update in i.parse (InvDataSeq2_Zero):
for key, update in i.parse (inv_data_seq2_zero):
if key == 'total':
assert update == False
tests +=1
@@ -455,7 +455,7 @@ def test_must_incr_total2(InvDataSeq2, InvDataSeq2_Zero):
assert json.dumps(i.db['env']) == json.dumps({"Inverter_Temp": 0})
tests = 0
for key, update in i.parse (InvDataSeq2):
for key, update in i.parse (inv_data_seq2):
if key == 'total':
assert update == True
tests +=1
@@ -467,10 +467,10 @@ def test_must_incr_total2(InvDataSeq2, InvDataSeq2_Zero):
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}})
def test_new_data_types(InvDataNew):
def test_new_data_types(inv_data_new):
i = InfosG3()
tests = 0
for key, update in i.parse (InvDataNew):
for key, update in i.parse (inv_data_new):
if key == 'events':
tests +=1
elif key == 'inverter':
@@ -487,7 +487,7 @@ def test_new_data_types(InvDataNew):
assert json.dumps(i.db['input']) == json.dumps({"pv1": {}})
assert json.dumps(i.db['events']) == json.dumps({"401_": 0, "404_": 0, "405_": 0, "408_": 0, "409_No_Utility": 0, "406_": 0, "416_": 0})
def test_invalid_data_type(InvalidDataSeq):
def test_invalid_data_type(invalid_data_seq):
i = InfosG3()
i.static_init() # initialize counter
@@ -495,8 +495,8 @@ def test_invalid_data_type(InvalidDataSeq):
assert val == 0
for key, result in i.parse (InvalidDataSeq):
pass
for key, result in i.parse (invalid_data_seq):
pass # side effect in calling i.parse()
assert json.dumps(i.db) == json.dumps({"inverter": {"Product_Name": "Microinv"}})
val = i.dev_value(Register.INVALID_DATA_TYPE) # check invalid data type counter

4
app/tests/test_infos_g3p.py
View File

@@ -70,7 +70,7 @@ def test_parse_4110(device_data: bytes):
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"},
'controller': {"Data_Up_Interval": 300, "Collect_Interval": 1, "Heartbeat_Interval": 120, "Signal_Strength": 100, "IP_Address": "192.168.80.49", "Sensor_List": "02b0"},
'collector': {"Chip_Model": "LSW5BLE_17_02B0_1.05", "Collector_Fw_Version": "V1.1.00.0B"},
})
@@ -82,7 +82,7 @@ def test_parse_4210(inverter_data: bytes):
pass # side effect is calling generator i.parse()
assert json.dumps(i.db) == json.dumps({
"controller": {"Power_On_Time": 2051},
"controller": {"Sensor_List": "02b0", "Power_On_Time": 2051},
"inverter": {"Serial_Number": "Y17E00000000000E", "Version": "V4.0.10", "Rated_Power": 600, "Max_Designed_Power": 2000},
"env": {"Inverter_Status": 1, "Inverter_Temp": 14},
"grid": {"Voltage": 224.8, "Current": 0.73, "Frequency": 50.05, "Output_Power": 165.8},

6
app/tests/test_modbus.py
View File

@@ -366,8 +366,8 @@ async def test_timeout():
def test_recv_unknown_data():
'''Receive a response with an unknwon register'''
mb = ModbusTestHelper()
assert 0x9000 not in mb.map
mb.map[0x9000] = {'reg': Register.TEST_REG1, 'fmt': '!H', 'ratio': 1}
assert 0x9000 not in mb.mb_reg_mapping
mb.mb_reg_mapping[0x9000] = {'reg': Register.TEST_REG1, 'fmt': '!H', 'ratio': 1}
mb.build_msg(1,3,0x9000,2)
@@ -379,7 +379,7 @@ def test_recv_unknown_data():
assert 0 == call
assert not mb.req_pend
del mb.map[0x9000]
del mb.mb_reg_mapping[0x9000]
def test_close():
'''Check queue handling for build_msg() calls'''

250
app/tests/test_mqtt.py Normal file
View File

@@ -0,0 +1,250 @@
# test_with_pytest.py
import pytest
import asyncio
import aiomqtt
import logging
from mock import patch, Mock
from app.src.mqtt import Mqtt
from app.src.modbus import Modbus
from app.src.gen3plus.solarman_v5 import SolarmanV5
from app.src.config import Config
pytest_plugins = ('pytest_asyncio',)
@pytest.fixture(scope="module")
def test_port():
return 1883
@pytest.fixture(scope="module")
def test_hostname():
# if getenv("GITHUB_ACTIONS") == "true":
# return 'mqtt'
# else:
return 'test.mosquitto.org'
@pytest.fixture
def config_mqtt_conn(test_hostname, test_port):
Config.act_config = {'mqtt':{'host': test_hostname, 'port': test_port, 'user': '', 'passwd': ''},
'ha':{'auto_conf_prefix': 'homeassistant','discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun'}
}
@pytest.fixture
def config_no_conn(test_port):
Config.act_config = {'mqtt':{'host': "", 'port': test_port, 'user': '', 'passwd': ''},
'ha':{'auto_conf_prefix': 'homeassistant','discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun'}
}
@pytest.fixture
def spy_at_cmd():
conn = SolarmanV5(server_side=True, client_mode= False)
conn.node_id = 'inv_2/'
with patch.object(conn, 'send_at_cmd', wraps=conn.send_at_cmd) as wrapped_conn:
yield wrapped_conn
conn.close()
@pytest.fixture
def spy_modbus_cmd():
conn = SolarmanV5(server_side=True, client_mode= False)
conn.node_id = 'inv_1/'
with patch.object(conn, 'send_modbus_cmd', wraps=conn.send_modbus_cmd) as wrapped_conn:
yield wrapped_conn
conn.close()
@pytest.fixture
def spy_modbus_cmd_client():
conn = SolarmanV5(server_side=False, client_mode= False)
conn.node_id = 'inv_1/'
with patch.object(conn, 'send_modbus_cmd', wraps=conn.send_modbus_cmd) as wrapped_conn:
yield wrapped_conn
conn.close()
def test_native_client(test_hostname, test_port):
"""Sanity check: Make sure the paho-mqtt client can connect to the test
MQTT server.
"""
import paho.mqtt.client as mqtt
import threading
c = mqtt.Client()
c.loop_start()
try:
# Just make sure the client connects successfully
on_connect = threading.Event()
c.on_connect = Mock(side_effect=lambda *_: on_connect.set())
c.connect_async(test_hostname, test_port)
assert on_connect.wait(5)
finally:
c.loop_stop()
@pytest.mark.asyncio
async def test_mqtt_no_config(config_no_conn):
_ = config_no_conn
assert asyncio.get_running_loop()
on_connect = asyncio.Event()
async def cb():
on_connect.set()
try:
m = Mqtt(cb)
assert m.task
await asyncio.sleep(1)
assert not on_connect.is_set()
try:
await m.publish('homeassistant/status', 'online')
assert False
except Exception:
pass
except TimeoutError:
assert False
finally:
await m.close()
@pytest.mark.asyncio
async def test_mqtt_connection(config_mqtt_conn):
_ = config_mqtt_conn
assert asyncio.get_running_loop()
on_connect = asyncio.Event()
async def cb():
on_connect.set()
try:
m = Mqtt(cb)
assert m.task
assert await asyncio.wait_for(on_connect.wait(), 5)
# await asyncio.sleep(1)
assert 0 == m.ha_restarts
await m.publish('homeassistant/status', 'online')
except TimeoutError:
assert False
finally:
await m.close()
await m.publish('homeassistant/status', 'online')
@pytest.mark.asyncio
async def test_msg_dispatch(config_mqtt_conn, spy_modbus_cmd):
_ = config_mqtt_conn
spy = spy_modbus_cmd
try:
m = Mqtt(None)
msg = aiomqtt.Message(topic= 'tsun/inv_1/rated_load', payload= b'2', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_awaited_once_with(Modbus.WRITE_SINGLE_REG, 0x2008, 2, logging.INFO)
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_1/out_coeff', payload= b'100', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_awaited_once_with(Modbus.WRITE_SINGLE_REG, 0x202c, 1024, logging.INFO)
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_1/out_coeff', payload= b'50', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_awaited_once_with(Modbus.WRITE_SINGLE_REG, 0x202c, 512, logging.INFO)
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_1/modbus_read_regs', payload= b'0x3000, 10', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_awaited_once_with(Modbus.READ_REGS, 0x3000, 10, logging.INFO)
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_1/modbus_read_inputs', payload= b'0x3000, 10', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_awaited_once_with(Modbus.READ_INPUTS, 0x3000, 10, logging.INFO)
finally:
await m.close()
@pytest.mark.asyncio
async def test_msg_dispatch_err(config_mqtt_conn, spy_modbus_cmd):
_ = config_mqtt_conn
spy = spy_modbus_cmd
try:
m = Mqtt(None)
# test out of range param
msg = aiomqtt.Message(topic= 'tsun/inv_1/out_coeff', payload= b'-1', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_not_called()
# test unknown node_id
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_2/out_coeff', payload= b'2', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_not_called()
# test invalid fload param
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_1/out_coeff', payload= b'2, 3', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_not_called()
spy.reset_mock()
msg = aiomqtt.Message(topic= 'tsun/inv_1/modbus_read_regs', payload= b'0x3000, 10, 7', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_not_called()
finally:
await m.close()
@pytest.mark.asyncio
async def test_msg_ignore_client_conn(config_mqtt_conn, spy_modbus_cmd_client):
'''don't call function if connnection is not in server mode'''
_ = config_mqtt_conn
spy = spy_modbus_cmd_client
try:
m = Mqtt(None)
msg = aiomqtt.Message(topic= 'tsun/inv_1/rated_load', payload= b'2', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_not_called()
finally:
await m.close()
@pytest.mark.asyncio
async def test_ha_reconnect(config_mqtt_conn):
_ = config_mqtt_conn
on_connect = asyncio.Event()
async def cb():
on_connect.set()
try:
m = Mqtt(cb)
msg = aiomqtt.Message(topic= 'homeassistant/status', payload= b'offline', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
assert not on_connect.is_set()
msg = aiomqtt.Message(topic= 'homeassistant/status', payload= b'online', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
assert on_connect.is_set()
finally:
await m.close()
@pytest.mark.asyncio
async def test_ignore_unknown_func(config_mqtt_conn):
'''don't dispatch for unknwon function names'''
_ = config_mqtt_conn
try:
m = Mqtt(None)
msg = aiomqtt.Message(topic= 'tsun/inv_1/rated_load', payload= b'2', qos= 0, retain = False, mid= 0, properties= None)
for _ in m.each_inverter(msg, 'unkown_fnc'):
assert False
finally:
await m.close()
@pytest.mark.asyncio
async def test_at_cmd_dispatch(config_mqtt_conn, spy_at_cmd):
_ = config_mqtt_conn
spy = spy_at_cmd
try:
m = Mqtt(None)
msg = aiomqtt.Message(topic= 'tsun/inv_2/at_cmd', payload= b'AT+', qos= 0, retain = False, mid= 0, properties= None)
await m.dispatch_msg(msg)
spy.assert_awaited_once_with('AT+')
finally:
await m.close()

18
app/tests/test_singleton.py Normal file
View File

@@ -0,0 +1,18 @@
# test_with_pytest.py
import pytest
from app.src.singleton import Singleton
class Test(metaclass=Singleton):
def __init__(self):
pass # is a dummy test class
def test_singleton_metaclass():
a = Test()
assert 1 == len(Singleton._instances)
b = Test()
assert 1 == len(Singleton._instances)
assert a is b
del a
assert 1 == len(Singleton._instances)
del b
assert 0 == len(Singleton._instances)

104
app/tests/test_solarman.py
View File

@@ -3,11 +3,12 @@ import struct
import time
import asyncio
import logging
from math import isclose
from app.src.gen3plus.solarman_v5 import SolarmanV5
from app.src.config import Config
from app.src.infos import Infos, Register
from app.src.modbus import Modbus
from app.src.messages import State
from app.src.messages import State, Message
pytest_plugins = ('pytest_asyncio',)
@@ -41,7 +42,7 @@ class MemoryStream(SolarmanV5):
super().__init__(server_side, client_mode=False)
if server_side:
self.mb.timeout = 0.4 # overwrite for faster testing
self.mb_start_timeout = 0.5
self.mb_first_timeout = 0.5
self.mb_timeout = 0.5
self.writer = Writer()
self.mqtt = Mqtt()
@@ -632,15 +633,15 @@ def msg_unknown_cmd_rsp(): # 0x1510
@pytest.fixture
def config_tsun_allow_all():
Config.config = {'solarman':{'enabled': True}, 'inverters':{'allow_all':True}}
Config.act_config = {'solarman':{'enabled': True}, 'inverters':{'allow_all':True}}
@pytest.fixture
def config_no_tsun_inv1():
Config.config = {'solarman':{'enabled': False},'inverters':{'Y170000000000001':{'monitor_sn': 2070233889, 'node_id':'inv1', 'modbus_polling': True, 'suggested_area':'roof'}}}
Config.act_config = {'solarman':{'enabled': False},'inverters':{'Y170000000000001':{'monitor_sn': 2070233889, 'node_id':'inv1', 'modbus_polling': True, 'suggested_area':'roof', 'sensor_list': 688}}}
@pytest.fixture
def config_tsun_inv1():
Config.config = {'solarman':{'enabled': True},'inverters':{'Y170000000000001':{'monitor_sn': 2070233889, 'node_id':'inv1', 'modbus_polling': True, 'suggested_area':'roof'}}}
Config.act_config = {'solarman':{'enabled': True},'inverters':{'Y170000000000001':{'monitor_sn': 2070233889, 'node_id':'inv1', 'modbus_polling': True, 'suggested_area':'roof', 'sensor_list': 688}}}
def test_read_message(device_ind_msg):
m = MemoryStream(device_ind_msg, (0,))
@@ -772,7 +773,7 @@ def test_invalid_checksum(invalid_checksum, device_ind_msg):
m.close()
def test_read_message_twice(config_no_tsun_inv1, device_ind_msg, device_rsp_msg):
config_no_tsun_inv1
_ = config_no_tsun_inv1
m = MemoryStream(device_ind_msg, (0,))
m.append_msg(device_ind_msg)
m.read() # read complete msg, and dispatch msg
@@ -814,7 +815,7 @@ def test_read_message_in_chunks(device_ind_msg):
m.close()
def test_read_message_in_chunks2(config_tsun_inv1, device_ind_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(device_ind_msg, (4,10,0))
m.read() # read 4 bytes, header incomplere
assert not m.header_valid
@@ -839,10 +840,10 @@ def test_read_message_in_chunks2(config_tsun_inv1, device_ind_msg):
m.close()
def test_read_two_messages(config_tsun_allow_all, device_ind_msg, device_rsp_msg, inverter_ind_msg, inverter_rsp_msg):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(device_ind_msg, (0,))
m.append_msg(inverter_ind_msg)
assert 0 == m.sensor_list
m._init_new_client_conn()
m.read() # read complete msg, and dispatch msg
assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0
@@ -857,6 +858,8 @@ def test_read_two_messages(config_tsun_allow_all, device_ind_msg, device_rsp_msg
assert m.msg_recvd[1]['control']==0x4210
assert m.msg_recvd[1]['seq']=='02:02'
assert m.msg_recvd[1]['data_len']==0x199
assert '02b0' == m.db.get_db_value(Register.SENSOR_LIST, None)
assert 0x02b0 == m.sensor_list
assert m._forward_buffer==device_ind_msg+inverter_ind_msg
assert m._send_buffer==device_rsp_msg+inverter_rsp_msg
@@ -866,7 +869,7 @@ def test_read_two_messages(config_tsun_allow_all, device_ind_msg, device_rsp_msg
m.close()
def test_read_two_messages2(config_tsun_allow_all, inverter_ind_msg, inverter_ind_msg_81, inverter_rsp_msg, inverter_rsp_msg_81):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(inverter_ind_msg, (0,))
m.append_msg(inverter_ind_msg_81)
m.read() # read complete msg, and dispatch msg
@@ -892,7 +895,7 @@ def test_read_two_messages2(config_tsun_allow_all, inverter_ind_msg, inverter_in
m.close()
def test_unkown_message(config_tsun_inv1, unknown_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(unknown_msg, (0,))
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -910,7 +913,7 @@ def test_unkown_message(config_tsun_inv1, unknown_msg):
m.close()
def test_device_rsp(config_tsun_inv1, device_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(device_rsp_msg, (0,), False)
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -928,7 +931,7 @@ def test_device_rsp(config_tsun_inv1, device_rsp_msg):
m.close()
def test_inverter_rsp(config_tsun_inv1, inverter_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(inverter_rsp_msg, (0,), False)
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -946,7 +949,7 @@ def test_inverter_rsp(config_tsun_inv1, inverter_rsp_msg):
m.close()
def test_heartbeat_ind(config_tsun_inv1, heartbeat_ind_msg, heartbeat_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(heartbeat_ind_msg, (0,))
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -963,7 +966,7 @@ def test_heartbeat_ind(config_tsun_inv1, heartbeat_ind_msg, heartbeat_rsp_msg):
m.close()
def test_heartbeat_ind2(config_tsun_inv1, heartbeat_ind_msg, heartbeat_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(heartbeat_ind_msg, (0,))
m.no_forwarding = True
m.read() # read complete msg, and dispatch msg
@@ -981,7 +984,7 @@ def test_heartbeat_ind2(config_tsun_inv1, heartbeat_ind_msg, heartbeat_rsp_msg):
m.close()
def test_heartbeat_rsp(config_tsun_inv1, heartbeat_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(heartbeat_rsp_msg, (0,), False)
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -999,7 +1002,7 @@ def test_heartbeat_rsp(config_tsun_inv1, heartbeat_rsp_msg):
m.close()
def test_sync_start_ind(config_tsun_inv1, sync_start_ind_msg, sync_start_rsp_msg, sync_start_fwd_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(sync_start_ind_msg, (0,))
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -1022,7 +1025,7 @@ def test_sync_start_ind(config_tsun_inv1, sync_start_ind_msg, sync_start_rsp_msg
m.close()
def test_sync_start_rsp(config_tsun_inv1, sync_start_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(sync_start_rsp_msg, (0,), False)
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -1040,7 +1043,7 @@ def test_sync_start_rsp(config_tsun_inv1, sync_start_rsp_msg):
m.close()
def test_sync_end_ind(config_tsun_inv1, sync_end_ind_msg, sync_end_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(sync_end_ind_msg, (0,))
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -1057,7 +1060,7 @@ def test_sync_end_ind(config_tsun_inv1, sync_end_ind_msg, sync_end_rsp_msg):
m.close()
def test_sync_end_rsp(config_tsun_inv1, sync_end_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(sync_end_rsp_msg, (0,), False)
m.read() # read complete msg, and dispatch msg
assert not m.header_valid # must be invalid, since msg was handled and buffer flushed
@@ -1075,8 +1078,9 @@ def test_sync_end_rsp(config_tsun_inv1, sync_end_rsp_msg):
m.close()
def test_build_modell_600(config_tsun_allow_all, inverter_ind_msg):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(inverter_ind_msg, (0,))
assert 0 == m.sensor_list
assert 0 == m.db.get_db_value(Register.MAX_DESIGNED_POWER, 0)
assert None == m.db.get_db_value(Register.RATED_POWER, None)
assert None == m.db.get_db_value(Register.INVERTER_TEMP, None)
@@ -1084,6 +1088,8 @@ def test_build_modell_600(config_tsun_allow_all, inverter_ind_msg):
assert 2000 == m.db.get_db_value(Register.MAX_DESIGNED_POWER, 0)
assert 600 == m.db.get_db_value(Register.RATED_POWER, 0)
assert 'TSOL-MS2000(600)' == m.db.get_db_value(Register.EQUIPMENT_MODEL, 0)
assert '02b0' == m.db.get_db_value(Register.SENSOR_LIST, None)
assert 0 == m.sensor_list # must not been set by an inverter data ind
m._send_buffer = bytearray(0) # clear send buffer for next test
m._init_new_client_conn()
@@ -1091,7 +1097,7 @@ def test_build_modell_600(config_tsun_allow_all, inverter_ind_msg):
m.close()
def test_build_modell_1600(config_tsun_allow_all, inverter_ind_msg1600):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(inverter_ind_msg1600, (0,))
assert 0 == m.db.get_db_value(Register.MAX_DESIGNED_POWER, 0)
assert None == m.db.get_db_value(Register.RATED_POWER, None)
@@ -1103,7 +1109,7 @@ def test_build_modell_1600(config_tsun_allow_all, inverter_ind_msg1600):
m.close()
def test_build_modell_1800(config_tsun_allow_all, inverter_ind_msg1800):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(inverter_ind_msg1800, (0,))
assert 0 == m.db.get_db_value(Register.MAX_DESIGNED_POWER, 0)
assert None == m.db.get_db_value(Register.RATED_POWER, None)
@@ -1115,7 +1121,7 @@ def test_build_modell_1800(config_tsun_allow_all, inverter_ind_msg1800):
m.close()
def test_build_modell_2000(config_tsun_allow_all, inverter_ind_msg2000):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(inverter_ind_msg2000, (0,))
assert 0 == m.db.get_db_value(Register.MAX_DESIGNED_POWER, 0)
assert None == m.db.get_db_value(Register.RATED_POWER, None)
@@ -1127,7 +1133,7 @@ def test_build_modell_2000(config_tsun_allow_all, inverter_ind_msg2000):
m.close()
def test_build_modell_800(config_tsun_allow_all, inverter_ind_msg800):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(inverter_ind_msg800, (0,))
assert 0 == m.db.get_db_value(Register.MAX_DESIGNED_POWER, 0)
assert None == m.db.get_db_value(Register.RATED_POWER, None)
@@ -1139,7 +1145,7 @@ def test_build_modell_800(config_tsun_allow_all, inverter_ind_msg800):
m.close()
def test_build_logger_modell(config_tsun_allow_all, device_ind_msg):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(device_ind_msg, (0,))
assert 0 == m.db.get_db_value(Register.COLLECTOR_FW_VERSION, 0)
assert 'IGEN TECH' == m.db.get_db_value(Register.CHIP_TYPE, None)
@@ -1150,6 +1156,7 @@ def test_build_logger_modell(config_tsun_allow_all, device_ind_msg):
m.close()
def test_msg_iterator():
Message._registry.clear()
m1 = SolarmanV5(server_side=True, client_mode=False)
m2 = SolarmanV5(server_side=True, client_mode=False)
m3 = SolarmanV5(server_side=True, client_mode=False)
@@ -1189,7 +1196,7 @@ def test_proxy_counter():
@pytest.mark.asyncio
async def test_msg_build_modbus_req(config_tsun_inv1, device_ind_msg, device_rsp_msg, inverter_ind_msg, inverter_rsp_msg, msg_modbus_cmd):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(device_ind_msg, (0,), True)
m.read()
assert m.control == 0x4110
@@ -1235,7 +1242,7 @@ async def test_msg_build_modbus_req(config_tsun_inv1, device_ind_msg, device_rsp
@pytest.mark.asyncio
async def test_at_cmd(config_tsun_allow_all, device_ind_msg, device_rsp_msg, inverter_ind_msg, inverter_rsp_msg, at_command_ind_msg, at_command_rsp_msg):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(device_ind_msg, (0,), True)
m.read() # read device ind
assert m.control == 0x4110
@@ -1292,7 +1299,7 @@ async def test_at_cmd(config_tsun_allow_all, device_ind_msg, device_rsp_msg, inv
@pytest.mark.asyncio
async def test_at_cmd_blocked(config_tsun_allow_all, device_ind_msg, device_rsp_msg, inverter_ind_msg, inverter_rsp_msg, at_command_ind_msg):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(device_ind_msg, (0,), True)
m.read()
assert m.control == 0x4110
@@ -1330,7 +1337,7 @@ async def test_at_cmd_blocked(config_tsun_allow_all, device_ind_msg, device_rsp_
m.close()
def test_at_cmd_ind(config_tsun_inv1, at_command_ind_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(at_command_ind_msg, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['AT_Command'] = 0
@@ -1354,7 +1361,7 @@ def test_at_cmd_ind(config_tsun_inv1, at_command_ind_msg):
m.close()
def test_at_cmd_ind_block(config_tsun_inv1, at_command_ind_msg_block):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(at_command_ind_msg_block, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['AT_Command'] = 0
@@ -1378,7 +1385,7 @@ def test_at_cmd_ind_block(config_tsun_inv1, at_command_ind_msg_block):
m.close()
def test_msg_at_command_rsp1(config_tsun_inv1, at_command_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(at_command_rsp_msg)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['Modbus_Command'] = 0
@@ -1397,7 +1404,7 @@ def test_msg_at_command_rsp1(config_tsun_inv1, at_command_rsp_msg):
m.close()
def test_msg_at_command_rsp2(config_tsun_inv1, at_command_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(at_command_rsp_msg)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['Modbus_Command'] = 0
@@ -1416,9 +1423,10 @@ def test_msg_at_command_rsp2(config_tsun_inv1, at_command_rsp_msg):
m.close()
def test_msg_modbus_req(config_tsun_inv1, msg_modbus_cmd, msg_modbus_cmd_fwd):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(b'')
m.snr = get_sn_int()
m.sensor_list = 0x2b0
m.state = State.up
c = m.createClientStream(msg_modbus_cmd)
@@ -1443,7 +1451,7 @@ def test_msg_modbus_req(config_tsun_inv1, msg_modbus_cmd, msg_modbus_cmd_fwd):
m.close()
def test_msg_modbus_req2(config_tsun_inv1, msg_modbus_cmd_crc_err):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(b'')
m.snr = get_sn_int()
m.state = State.up
@@ -1470,7 +1478,7 @@ def test_msg_modbus_req2(config_tsun_inv1, msg_modbus_cmd_crc_err):
m.close()
def test_msg_unknown_cmd_req(config_tsun_inv1, msg_unknown_cmd):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_unknown_cmd, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['AT_Command'] = 0
@@ -1493,7 +1501,7 @@ def test_msg_unknown_cmd_req(config_tsun_inv1, msg_unknown_cmd):
def test_msg_modbus_rsp1(config_tsun_inv1, msg_modbus_rsp):
'''Modbus response without a valid Modbus request must be dropped'''
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_modbus_rsp)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['Modbus_Command'] = 0
@@ -1512,7 +1520,7 @@ def test_msg_modbus_rsp1(config_tsun_inv1, msg_modbus_rsp):
def test_msg_modbus_rsp2(config_tsun_inv1, msg_modbus_rsp):
'''Modbus response with a valid Modbus request must be forwarded'''
config_tsun_inv1 # setup config structure
_ = config_tsun_inv1 # setup config structure
m = MemoryStream(msg_modbus_rsp)
m.mb.rsp_handler = m._SolarmanV5__forward_msg
@@ -1550,7 +1558,7 @@ def test_msg_modbus_rsp2(config_tsun_inv1, msg_modbus_rsp):
def test_msg_modbus_rsp3(config_tsun_inv1, msg_modbus_rsp):
'''Modbus response with a valid Modbus request must be forwarded'''
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_modbus_rsp)
m.mb.rsp_handler = m._SolarmanV5__forward_msg
@@ -1586,7 +1594,7 @@ def test_msg_modbus_rsp3(config_tsun_inv1, msg_modbus_rsp):
m.close()
def test_msg_unknown_rsp(config_tsun_inv1, msg_unknown_cmd_rsp):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_unknown_cmd_rsp)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['Modbus_Command'] = 0
@@ -1604,7 +1612,7 @@ def test_msg_unknown_rsp(config_tsun_inv1, msg_unknown_cmd_rsp):
m.close()
def test_msg_modbus_invalid(config_tsun_inv1, msg_modbus_invalid):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_modbus_invalid, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['Modbus_Command'] = 0
@@ -1618,7 +1626,7 @@ def test_msg_modbus_invalid(config_tsun_inv1, msg_modbus_invalid):
m.close()
def test_msg_modbus_fragment(config_tsun_inv1, msg_modbus_rsp):
config_tsun_inv1
_ = config_tsun_inv1
# receive more bytes than expected (7 bytes from the next msg)
m = MemoryStream(msg_modbus_rsp+b'\x00\x00\x00\x45\x10\x52\x31', (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -1644,7 +1652,7 @@ def test_msg_modbus_fragment(config_tsun_inv1, msg_modbus_rsp):
@pytest.mark.asyncio
async def test_modbus_polling(config_tsun_inv1, heartbeat_ind_msg, heartbeat_rsp_msg):
config_tsun_inv1
_ = config_tsun_inv1
assert asyncio.get_running_loop()
m = MemoryStream(heartbeat_ind_msg, (0,))
assert asyncio.get_running_loop() == m.mb_timer.loop
@@ -1665,7 +1673,7 @@ async def test_modbus_polling(config_tsun_inv1, heartbeat_ind_msg, heartbeat_rsp
m._send_buffer = bytearray(0) # clear send buffer for next test
assert m.state == State.up
assert m.mb_timeout == 0.5
assert isclose(m.mb_timeout, 0.5)
assert next(m.mb_timer.exp_count) == 0
await asyncio.sleep(0.5)
@@ -1685,24 +1693,24 @@ async def test_modbus_polling(config_tsun_inv1, heartbeat_ind_msg, heartbeat_rsp
@pytest.mark.asyncio
async def test_start_client_mode(config_tsun_inv1):
config_tsun_inv1
_ = config_tsun_inv1
assert asyncio.get_running_loop()
m = MemoryStream(b'')
assert m.state == State.init
assert m.no_forwarding == False
assert m.mb_timer.tim == None
assert asyncio.get_running_loop() == m.mb_timer.loop
await m.send_start_cmd(get_sn_int(), '192.168.1.1', m.mb_start_timeout)
await m.send_start_cmd(get_sn_int(), '192.168.1.1', m.mb_first_timeout)
assert m.writer.sent_pdu==bytearray(b'\xa5\x17\x00\x10E\x01\x00!Ce{\x02\xb0\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x030\x00\x000J\xde\xf1\x15')
assert m.db.get_db_value(Register.IP_ADDRESS) == '192.168.1.1'
assert m.db.get_db_value(Register.POLLING_INTERVAL) == 0.5
assert isclose(m.db.get_db_value(Register.POLLING_INTERVAL), 0.5)
assert m.db.get_db_value(Register.HEARTBEAT_INTERVAL) == 120
assert m.state == State.up
assert m.no_forwarding == True
assert m._send_buffer==b''
assert m.mb_timeout == 0.5
assert isclose(m.mb_timeout, 0.5)
assert next(m.mb_timer.exp_count) == 0
await asyncio.sleep(0.5)

101
app/tests/test_talent.py
View File

@@ -1,5 +1,6 @@
# test_with_pytest.py
import pytest, logging, asyncio
from math import isclose
from app.src.gen3.talent import Talent, Control
from app.src.config import Config
from app.src.infos import Infos, Register
@@ -27,7 +28,7 @@ class MemoryStream(Talent):
super().__init__(server_side)
if server_side:
self.mb.timeout = 0.4 # overwrite for faster testing
self.mb_start_timeout = 0.5
self.mb_first_timeout = 0.5
self.mb_timeout = 0.5
self.writer = Writer()
self.__msg = msg
@@ -97,12 +98,12 @@ class MemoryStream(Talent):
@pytest.fixture
def msg_contact_info(): # Contact Info message
Config.config = {'tsun':{'enabled': True}}
Config.act_config = {'tsun':{'enabled': True}}
return b'\x00\x00\x00\x2c\x10R170000000000001\x91\x00\x08solarhub\x0fsolarhub\x40123456'
@pytest.fixture
def msg_contact_info_long_id(): # Contact Info message with longer ID
Config.config = {'tsun':{'enabled': True}}
Config.act_config = {'tsun':{'enabled': True}}
return b'\x00\x00\x00\x2d\x11R1700000000000011\x91\x00\x08solarhub\x0fsolarhub\x40123456'
@pytest.fixture
@@ -352,19 +353,19 @@ def msg_unknown(): # Get Time Request message
@pytest.fixture
def config_tsun_allow_all():
Config.config = {'tsun':{'enabled': True}, 'inverters':{'allow_all':True}}
Config.act_config = {'tsun':{'enabled': True}, 'inverters':{'allow_all':True}}
@pytest.fixture
def config_no_tsun_inv1():
Config.config = {'tsun':{'enabled': False},'inverters':{'R170000000000001':{'node_id':'inv1', 'modbus_polling': True, 'suggested_area':'roof'}}}
Config.act_config = {'tsun':{'enabled': False},'inverters':{'R170000000000001':{'node_id':'inv1', 'modbus_polling': True, 'suggested_area':'roof'}}}
@pytest.fixture
def config_tsun_inv1():
Config.config = {'tsun':{'enabled': True},'inverters':{'R170000000000001':{'node_id':'inv1', 'modbus_polling': True, 'suggested_area':'roof'}}}
Config.act_config = {'tsun':{'enabled': True},'inverters':{'R170000000000001':{'node_id':'inv1', 'modbus_polling': True, 'suggested_area':'roof'}}}
@pytest.fixture
def config_no_modbus_poll():
Config.config = {'tsun':{'enabled': True},'inverters':{'R170000000000001':{'node_id':'inv1', 'modbus_polling': False, 'suggested_area':'roof'}}}
Config.act_config = {'tsun':{'enabled': True},'inverters':{'R170000000000001':{'node_id':'inv1', 'modbus_polling': False, 'suggested_area':'roof'}}}
@pytest.fixture
def msg_ota_req(): # Over the air update request from tsun cloud
@@ -540,7 +541,7 @@ def test_read_message(msg_contact_info):
m.close()
def test_read_message_twice(config_no_tsun_inv1, msg_inverter_ind):
config_no_tsun_inv1
_ = config_no_tsun_inv1
m = MemoryStream(msg_inverter_ind, (0,))
m.append_msg(msg_inverter_ind)
m.read() # read complete msg, and dispatch msg
@@ -621,7 +622,7 @@ def test_read_message_in_chunks2(msg_contact_info):
m.close()
def test_read_two_messages(config_tsun_allow_all, msg2_contact_info,msg_contact_rsp,msg_contact_rsp2):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(msg2_contact_info, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -651,7 +652,7 @@ def test_read_two_messages(config_tsun_allow_all, msg2_contact_info,msg_contact_
m.close()
def test_msg_contact_resp(config_tsun_inv1, msg_contact_rsp):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_contact_rsp, (0,), False)
m.await_conn_resp_cnt = 1
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -671,7 +672,7 @@ def test_msg_contact_resp(config_tsun_inv1, msg_contact_rsp):
m.close()
def test_msg_contact_resp_2(config_tsun_inv1, msg_contact_rsp):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_contact_rsp, (0,), False)
m.await_conn_resp_cnt = 0
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -691,7 +692,7 @@ def test_msg_contact_resp_2(config_tsun_inv1, msg_contact_rsp):
m.close()
def test_msg_contact_resp_3(config_tsun_inv1, msg_contact_rsp):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_contact_rsp, (0,), True)
m.await_conn_resp_cnt = 0
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -711,7 +712,7 @@ def test_msg_contact_resp_3(config_tsun_inv1, msg_contact_rsp):
m.close()
def test_msg_contact_invalid(config_tsun_inv1, msg_contact_invalid):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_contact_invalid, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -729,7 +730,7 @@ def test_msg_contact_invalid(config_tsun_inv1, msg_contact_invalid):
m.close()
def test_msg_get_time(config_tsun_inv1, msg_get_time):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_get_time, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -748,7 +749,7 @@ def test_msg_get_time(config_tsun_inv1, msg_get_time):
m.close()
def test_msg_get_time_autark(config_no_tsun_inv1, msg_get_time):
config_no_tsun_inv1
_ = config_no_tsun_inv1
m = MemoryStream(msg_get_time, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -767,7 +768,7 @@ def test_msg_get_time_autark(config_no_tsun_inv1, msg_get_time):
m.close()
def test_msg_time_resp(config_tsun_inv1, msg_time_rsp):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_time_rsp, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -786,7 +787,7 @@ def test_msg_time_resp(config_tsun_inv1, msg_time_rsp):
m.close()
def test_msg_time_resp_autark(config_no_tsun_inv1, msg_time_rsp):
config_no_tsun_inv1
_ = config_no_tsun_inv1
m = MemoryStream(msg_time_rsp, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -805,7 +806,7 @@ def test_msg_time_resp_autark(config_no_tsun_inv1, msg_time_rsp):
m.close()
def test_msg_time_inv_resp(config_tsun_inv1, msg_time_rsp_inv):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_time_rsp_inv, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -824,7 +825,7 @@ def test_msg_time_inv_resp(config_tsun_inv1, msg_time_rsp_inv):
m.close()
def test_msg_time_invalid(config_tsun_inv1, msg_time_invalid):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_time_invalid, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -843,7 +844,7 @@ def test_msg_time_invalid(config_tsun_inv1, msg_time_invalid):
m.close()
def test_msg_time_invalid_autark(config_no_tsun_inv1, msg_time_invalid):
config_no_tsun_inv1
_ = config_no_tsun_inv1
m = MemoryStream(msg_time_invalid, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -862,7 +863,7 @@ def test_msg_time_invalid_autark(config_no_tsun_inv1, msg_time_invalid):
m.close()
def test_msg_cntrl_ind(config_tsun_inv1, msg_controller_ind, msg_controller_ind_ts_offs, msg_controller_ack):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_controller_ind, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -885,7 +886,7 @@ def test_msg_cntrl_ind(config_tsun_inv1, msg_controller_ind, msg_controller_ind_
m.close()
def test_msg_cntrl_ack(config_tsun_inv1, msg_controller_ack):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_controller_ack, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -903,7 +904,7 @@ def test_msg_cntrl_ack(config_tsun_inv1, msg_controller_ack):
m.close()
def test_msg_cntrl_invalid(config_tsun_inv1, msg_controller_invalid):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_controller_invalid, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -926,7 +927,7 @@ def test_msg_cntrl_invalid(config_tsun_inv1, msg_controller_invalid):
m.close()
def test_msg_inv_ind(config_tsun_inv1, msg_inverter_ind, msg_inverter_ind_ts_offs, msg_inverter_ack):
config_tsun_inv1
_ = config_tsun_inv1
tracer.setLevel(logging.DEBUG)
m = MemoryStream(msg_inverter_ind, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -950,7 +951,7 @@ def test_msg_inv_ind(config_tsun_inv1, msg_inverter_ind, msg_inverter_ind_ts_off
m.close()
def test_msg_inv_ind1(config_tsun_inv1, msg_inverter_ind2, msg_inverter_ind_ts_offs, msg_inverter_ack):
config_tsun_inv1
_ = config_tsun_inv1
tracer.setLevel(logging.DEBUG)
m = MemoryStream(msg_inverter_ind2, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -974,7 +975,7 @@ def test_msg_inv_ind1(config_tsun_inv1, msg_inverter_ind2, msg_inverter_ind_ts_o
m.close()
def test_msg_inv_ind2(config_tsun_inv1, msg_inverter_ind_new, msg_inverter_ind_ts_offs, msg_inverter_ack):
config_tsun_inv1
_ = config_tsun_inv1
tracer.setLevel(logging.DEBUG)
m = MemoryStream(msg_inverter_ind_new, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -1002,7 +1003,7 @@ def test_msg_inv_ind2(config_tsun_inv1, msg_inverter_ind_new, msg_inverter_ind_t
def test_msg_inv_ind3(config_tsun_inv1, msg_inverter_ind_0w, msg_inverter_ack):
'''test that after close the invert_status will be resetted if the grid power is <2W'''
config_tsun_inv1
_ = config_tsun_inv1
tracer.setLevel(logging.DEBUG)
m = MemoryStream(msg_inverter_ind_0w, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -1023,13 +1024,13 @@ def test_msg_inv_ind3(config_tsun_inv1, msg_inverter_ind_0w, msg_inverter_ack):
assert m._forward_buffer==msg_inverter_ind_0w
assert m._send_buffer==msg_inverter_ack
assert m.db.get_db_value(Register.INVERTER_STATUS) == None
assert m.db.db['grid']['Output_Power'] == 0.5
assert isclose(m.db.db['grid']['Output_Power'], 0.5)
m.close()
assert m.db.get_db_value(Register.INVERTER_STATUS) == 0
def test_msg_inv_ack(config_tsun_inv1, msg_inverter_ack):
config_tsun_inv1
_ = config_tsun_inv1
tracer.setLevel(logging.ERROR)
m = MemoryStream(msg_inverter_ack, (0,), False)
@@ -1049,7 +1050,7 @@ def test_msg_inv_ack(config_tsun_inv1, msg_inverter_ack):
m.close()
def test_msg_inv_invalid(config_tsun_inv1, msg_inverter_invalid):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_inverter_invalid, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.read() # read complete msg, and dispatch msg
@@ -1072,7 +1073,7 @@ def test_msg_inv_invalid(config_tsun_inv1, msg_inverter_invalid):
m.close()
def test_msg_ota_req(config_tsun_inv1, msg_ota_req):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_ota_req, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['OTA_Start_Msg'] = 0
@@ -1097,7 +1098,7 @@ def test_msg_ota_req(config_tsun_inv1, msg_ota_req):
m.close()
def test_msg_ota_ack(config_tsun_inv1, msg_ota_ack):
config_tsun_inv1
_ = config_tsun_inv1
tracer.setLevel(logging.ERROR)
m = MemoryStream(msg_ota_ack, (0,), False)
@@ -1124,7 +1125,7 @@ def test_msg_ota_ack(config_tsun_inv1, msg_ota_ack):
m.close()
def test_msg_ota_invalid(config_tsun_inv1, msg_ota_invalid):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_ota_invalid, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['OTA_Start_Msg'] = 0
@@ -1206,15 +1207,15 @@ def test_timestamp_cnv():
m = MemoryStream(b'')
ts = 1722645998453 # Saturday, 3. August 2024 00:46:38.453 (GMT+2:00)
utc =1722638798.453 # GMT: Friday, 2. August 2024 22:46:38.453
assert utc == m._utcfromts(ts)
assert isclose(utc, m._utcfromts(ts))
ts = 1691246944000 # Saturday, 5. August 2023 14:49:04 (GMT+2:00)
utc =1691239744.0 # GMT: Saturday, 5. August 2023 12:49:04
assert utc == m._utcfromts(ts)
assert isclose(utc, m._utcfromts(ts))
ts = 1704152544000 # Monday, 1. January 2024 23:42:24 (GMT+1:00)
utc =1704148944.0 # GMT: Monday, 1. January 2024 22:42:24
assert utc == m._utcfromts(ts)
assert isclose(utc, m._utcfromts(ts))
m.close()
@@ -1267,7 +1268,7 @@ def test_proxy_counter():
m.close()
def test_msg_modbus_req(config_tsun_inv1, msg_modbus_cmd):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(b'')
m.id_str = b"R170000000000001"
m.state = State.up
@@ -1298,7 +1299,7 @@ def test_msg_modbus_req(config_tsun_inv1, msg_modbus_cmd):
m.close()
def test_msg_modbus_req2(config_tsun_inv1, msg_modbus_cmd):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(b'')
m.id_str = b"R170000000000001"
@@ -1328,7 +1329,7 @@ def test_msg_modbus_req2(config_tsun_inv1, msg_modbus_cmd):
m.close()
def test_msg_modbus_req3(config_tsun_inv1, msg_modbus_cmd_crc_err):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(b'')
m.id_str = b"R170000000000001"
c = m.createClientStream(msg_modbus_cmd_crc_err)
@@ -1357,7 +1358,7 @@ def test_msg_modbus_req3(config_tsun_inv1, msg_modbus_cmd_crc_err):
def test_msg_modbus_rsp1(config_tsun_inv1, msg_modbus_rsp):
'''Modbus response without a valid Modbus request must be dropped'''
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_modbus_rsp)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['Modbus_Command'] = 0
@@ -1378,7 +1379,7 @@ def test_msg_modbus_rsp1(config_tsun_inv1, msg_modbus_rsp):
def test_msg_modbus_cloud_rsp(config_tsun_inv1, msg_modbus_rsp):
'''Modbus response from TSUN without a valid Modbus request must be dropped'''
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_modbus_rsp, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['Unknown_Msg'] = 0
@@ -1401,7 +1402,7 @@ def test_msg_modbus_cloud_rsp(config_tsun_inv1, msg_modbus_rsp):
def test_msg_modbus_rsp2(config_tsun_inv1, msg_modbus_rsp20):
'''Modbus response with a valid Modbus request must be forwarded'''
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_modbus_rsp20)
m.append_msg(msg_modbus_rsp20)
@@ -1431,7 +1432,7 @@ def test_msg_modbus_rsp2(config_tsun_inv1, msg_modbus_rsp20):
def test_msg_modbus_rsp3(config_tsun_inv1, msg_modbus_rsp21):
'''Modbus response with a valid Modbus request must be forwarded'''
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_modbus_rsp21)
m.append_msg(msg_modbus_rsp21)
@@ -1460,7 +1461,7 @@ def test_msg_modbus_rsp3(config_tsun_inv1, msg_modbus_rsp21):
m.close()
def test_msg_modbus_invalid(config_tsun_inv1, msg_modbus_inv):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(msg_modbus_inv, (0,), False)
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.db.stat['proxy']['Modbus_Command'] = 0
@@ -1480,7 +1481,7 @@ def test_msg_modbus_invalid(config_tsun_inv1, msg_modbus_inv):
m.close()
def test_msg_modbus_fragment(config_tsun_inv1, msg_modbus_rsp20):
config_tsun_inv1
_ = config_tsun_inv1
# receive more bytes than expected (7 bytes from the next msg)
m = MemoryStream(msg_modbus_rsp20+b'\x00\x00\x00\x45\x10\x52\x31', (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
@@ -1512,7 +1513,7 @@ def test_msg_modbus_fragment(config_tsun_inv1, msg_modbus_rsp20):
@pytest.mark.asyncio
async def test_msg_build_modbus_req(config_tsun_inv1, msg_modbus_cmd):
config_tsun_inv1
_ = config_tsun_inv1
m = MemoryStream(b'', (0,), True)
m.id_str = b"R170000000000001"
await m.send_modbus_cmd(Modbus.WRITE_SINGLE_REG, 0x2008, 0, logging.DEBUG)
@@ -1538,7 +1539,7 @@ async def test_msg_build_modbus_req(config_tsun_inv1, msg_modbus_cmd):
m.close()
def test_modbus_no_polling(config_no_modbus_poll, msg_get_time):
config_no_modbus_poll
_ = config_no_modbus_poll
m = MemoryStream(msg_get_time, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
m.modbus_polling = False
@@ -1559,7 +1560,7 @@ def test_modbus_no_polling(config_no_modbus_poll, msg_get_time):
@pytest.mark.asyncio
async def test_modbus_polling(config_tsun_inv1, msg_inverter_ind):
config_tsun_inv1
_ = config_tsun_inv1
assert asyncio.get_running_loop()
m = MemoryStream(msg_inverter_ind, (0,))
@@ -1581,7 +1582,7 @@ async def test_modbus_polling(config_tsun_inv1, msg_inverter_ind):
assert m.db.stat['proxy']['Unknown_Ctrl'] == 0
m._send_buffer = bytearray(0) # clear send buffer for next test
assert m.mb_timeout == 0.5
assert isclose(m.mb_timeout, 0.5)
assert next(m.mb_timer.exp_count) == 0
await asyncio.sleep(0.5)
@@ -1599,7 +1600,7 @@ async def test_modbus_polling(config_tsun_inv1, msg_inverter_ind):
m.close()
def test_broken_recv_buf(config_tsun_allow_all, broken_recv_buf):
config_tsun_allow_all
_ = config_tsun_allow_all
m = MemoryStream(broken_recv_buf, (0,))
m.db.stat['proxy']['Unknown_Ctrl'] = 0
assert m.db.stat['proxy']['Invalid_Data_Type'] == 0