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

Merge pull request #64 from s-allius/test-config

Browse Source 
Improve config parsing and add unit tests
This commit is contained in:
Stefan Allius
2024-04-28 20:58:30 +02:00
committed by GitHub
parent b1e577d357 e8902f7923
commit e603bb9baa
3 changed files with 188 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
app/src/config.py
View File

@@ -76,51 +76,74 @@ class Config():
)
@classmethod
def read(cls) -> None:
def class_init(cls): # pragma: no cover
try:
# make the default config transparaent by copying it
# in the config.example file
logging.debug('Copy Defaul Config to config.example.toml')
shutil.copy2("default_config.toml",
"config/config.example.toml")
except Exception:
pass
cls.read()
@classmethod
def _read_config_file(cls) -> dict: # pragma: no cover
usr_config = {}
try:
with open("config/config.toml", "rb") as f:
usr_config = tomllib.load(f)
except Exception as error:
err = f'Config.read: {error}'
logging.error(err)
logging.info(
'\n To create the missing config.toml file, '
'you can rename the template config.example.toml\n'
' and customize it for your scenario.\n')
return usr_config
@classmethod
def read(cls, path='') -> None | str:
'''Read config file, merge it with the default config
and sanitize the result'''
err = None
config = {}
logger = logging.getLogger('data')
try:
# make the default config transparaent by copying it
# in the config.example file
shutil.copy2("default_config.toml", "config/config.example.toml")
# read example config file as default configuration
with open("default_config.toml", "rb") as f:
cls.def_config = {}
with open(f"{path}default_config.toml", "rb") as f:
def_config = tomllib.load(f)
cls.def_config = cls.conf_schema.validate(def_config)
# overwrite the default values, with values from
# the config.toml file
usr_config = cls._read_config_file()
# merge the default and the user config
config = def_config.copy()
for key in ['tsun', 'solarman', 'mqtt', 'ha', 'inverters']:
if key in usr_config:
config[key] |= usr_config[key]
try:
with open("config/config.toml", "rb") as f:
usr_config = tomllib.load(f)
cls.config = cls.conf_schema.validate(config)
except Exception as error:
logging.error(f'Config.read: {error}')
logging.info(
'\n To create the missing config.toml file, '
'you can rename the template config.example.toml\n'
' and customize it for your scenario.\n')
usr_config = def_config
err = f'Config.read: {error}'
logging.error(err)
config['tsun'] = def_config['tsun'] | usr_config['tsun']
config['solarman'] = def_config['solarman'] | \
usr_config['solarman']
config['mqtt'] = def_config['mqtt'] | usr_config['mqtt']
config['ha'] = def_config['ha'] | usr_config['ha']
config['inverters'] = def_config['inverters'] | \
usr_config['inverters']
cls.config = cls.conf_schema.validate(config)
cls.def_config = cls.conf_schema.validate(def_config)
# logging.debug(f'Readed config: "{cls.config}" ')
except Exception as error:
logger.error(f'Config.read: {error}')
err = f'Config.read: {error}'
logger.error(err)
cls.config = {}
return err
@classmethod
def get(cls, member: str = None):
'''Get a named attribute from the proxy config. If member ==

2
app/src/server.py
View File

@@ -77,7 +77,7 @@ if __name__ == "__main__":
# logging.getLogger('mqtt').setLevel(log_level)
# read config file
Config.read()
Config.class_init()
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

138
app/tests/test_config.py Normal file
View File

@@ -0,0 +1,138 @@
# test_with_pytest.py
import tomllib
from schema import SchemaMissingKeyError
from app.src.config import Config
class TstConfig(Config):
@classmethod
def set(cls, cnf):
cls.config = cnf
@classmethod
def _read_config_file(cls) -> dict:
return cls.config
def test_empty_config():
cnf = {}
try:
Config.conf_schema.validate(cnf)
assert False
except SchemaMissingKeyError:
assert True
def test_default_config():
with open("app/config/default_config.toml", "rb") as f:
cnf = tomllib.load(f)
try:
validated = Config.conf_schema.validate(cnf)
assert True
except:
assert False
assert validated == {'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': '', 'monitor_sn': 0, 'suggested_area': ''}, 'Y170000000000001': {'monitor_sn': 2000000000, 'node_id': '', 'suggested_area': ''}}}
def test_full_config():
cnf = {'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': '', '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': {'node_id': '', 'suggested_area': '', 'pv1': {'type': 'type1', 'manufacturer': 'man1'}, 'pv2': {'type': 'type2', 'manufacturer': 'man2'}, 'pv3': {'type': 'type3', 'manufacturer': 'man3'}},
'Y170000000000001': {'monitor_sn': 2000000000, 'node_id': '', 'suggested_area': ''}}}
try:
validated = Config.conf_schema.validate(cnf)
assert True
except:
assert False
assert validated == {'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': '', 'monitor_sn': 0, 'pv1': {'manufacturer': 'man1','type': 'type1'},'pv2': {'manufacturer': 'man2','type': 'type2'},'pv3': {'manufacturer': 'man3','type': 'type3'}, 'suggested_area': ''}, 'Y170000000000001': {'monitor_sn': 2000000000, 'node_id': '', 'suggested_area': ''}}}
def test_mininum_config():
cnf = {'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': '', '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': {}}
}
try:
validated = Config.conf_schema.validate(cnf)
assert True
except:
assert False
assert validated == {'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': '', 'monitor_sn': 0, 'suggested_area': ''}}}
def test_read_empty():
cnf = {}
TstConfig.set(cnf)
err = TstConfig.read('app/config/')
assert err == None
cnf = TstConfig.get()
assert cnf == {'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': '', 'monitor_sn': 0, 'node_id': ''}, 'Y170000000000001': {'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': ''}}}
defcnf = TstConfig.def_config.get('solarman')
assert defcnf == {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}
assert True == TstConfig.is_default('solarman')
def test_no_file():
cnf = {}
TstConfig.set(cnf)
err = TstConfig.read('')
assert err == "Config.read: [Errno 2] No such file or directory: 'default_config.toml'"
cnf = TstConfig.get()
assert cnf == {}
defcnf = TstConfig.def_config.get('solarman')
assert defcnf == None
def test_read_cnf1():
cnf = {'solarman' : {'enabled': False}}
TstConfig.set(cnf)
err = TstConfig.read('app/config/')
assert err == None
cnf = TstConfig.get()
assert cnf == {'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': '', 'monitor_sn': 0, 'node_id': ''}, 'Y170000000000001': {'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': ''}}}
cnf = TstConfig.get('solarman')
assert cnf == {'enabled': False, 'host': 'iot.talent-monitoring.com', 'port': 10000}
defcnf = TstConfig.def_config.get('solarman')
assert defcnf == {'enabled': True, 'host': 'iot.talent-monitoring.com', 'port': 10000}
assert False == TstConfig.is_default('solarman')
def test_read_cnf2():
cnf = {'solarman' : {'enabled': 'FALSE'}}
TstConfig.set(cnf)
err = TstConfig.read('app/config/')
assert err == None
cnf = TstConfig.get()
assert cnf == {'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': '', 'monitor_sn': 0, 'node_id': ''}, 'Y170000000000001': {'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': ''}}}
assert True == TstConfig.is_default('solarman')
def test_read_cnf3():
cnf = {'solarman' : {'port': 'FALSE'}}
TstConfig.set(cnf)
err = TstConfig.read('app/config/')
assert err == 'Config.read: Key \'solarman\' error:\nKey \'port\' error:\nint(\'FALSE\') raised ValueError("invalid literal for int() with base 10: \'FALSE\'")'
cnf = TstConfig.get()
assert cnf == {'solarman': {'port': 'FALSE'}}
def test_read_cnf4():
cnf = {'solarman' : {'port': 5000}}
TstConfig.set(cnf)
err = TstConfig.read('app/config/')
assert err == None
cnf = TstConfig.get()
assert cnf == {'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': '', 'monitor_sn': 0, 'node_id': ''}, 'Y170000000000001': {'monitor_sn': 2000000000, 'suggested_area': '', 'node_id': ''}}}
assert False == TstConfig.is_default('solarman')
def test_read_cnf5():
cnf = {'solarman' : {'port': 1023}}
TstConfig.set(cnf)
err = TstConfig.read('app/config/')
assert err != None
def test_read_cnf6():
cnf = {'solarman' : {'port': 65536}}
TstConfig.set(cnf)
err = TstConfig.read('app/config/')
assert err != None