From 1ed281e7c4c44f621f340b6a193dd5ecf44e1800 Mon Sep 17 00:00:00 2001 From: Pedro Date: Wed, 15 Nov 2023 16:49:16 +0100 Subject: [PATCH] Convert and validate server side config data --- modem/config.ini.example | 3 +- modem/config.py | 180 +++++++++++++++------------------------ tests/test_config.py | 9 +- tests/test_server.py | 4 +- 4 files changed, 73 insertions(+), 123 deletions(-) diff --git a/modem/config.ini.example b/modem/config.ini.example index 23fdeb38..b6c27676 100644 --- a/modem/config.ini.example +++ b/modem/config.ini.example @@ -1,5 +1,5 @@ [NETWORK] -modemport = 3000 +modemport = 5000 [STATION] mycall = DJ2LS-5 @@ -8,7 +8,6 @@ ssid_list = [] enable_explorer = False enable_stats = False - [AUDIO] input_device = 4f55 output_device = 4f55 diff --git a/modem/config.py b/modem/config.py index 5da7ddbc..54e9d422 100644 --- a/modem/config.py +++ b/modem/config.py @@ -8,12 +8,57 @@ class CONFIG: """ + config_types = { + 'NETWORK': { + 'modemport': int, + }, + 'STATION': { + 'mycall': str, + 'mygrid': str, + 'ssid_list': list, + 'enable_explorer': bool, + 'enable_stats': bool, + }, + 'AUDIO': { + 'input_device': str, + 'output_device': str, + 'rx_audio_level': int, + 'tx_audio_level': int, + 'enable_auto_tune': bool, + }, + 'RADIO': { + 'radiocontrol': str, + 'rigctld_ip': str, + 'rigctld_port': int, + 'radioport': str, + }, + 'TCI': { + 'tci_ip': str, + 'tci_port': int, + }, + 'MESH': { + 'enable_protocol': bool, + }, + 'MODEM': { + 'enable_fft': bool, + 'tuning_range_fmax': int, + 'tuning_range_fmin': int, + 'enable_fsk': bool, + 'enable_low_bandwidth_mode': bool, + 'respond_to_cq': bool, + 'rx_buffer_size': int, + 'enable_scatter': bool, + 'tx_delay': int, + }, + } + def __init__(self, configfile: str): + # set up logger self.log = structlog.get_logger("CONFIG") # init configparser - self.config = configparser.ConfigParser(inline_comment_prefixes="#", allow_no_value=True) + self.parser = configparser.ConfigParser(inline_comment_prefixes="#", allow_no_value=True) try: self.config_name = configfile @@ -30,148 +75,58 @@ class CONFIG: check if config file exists """ try: - return bool(self.config.read(self.config_name, None)) + return bool(self.parser.read(self.config_name, None)) except Exception as configerror: self.log.error("[CFG] logfile init error", e=configerror) return False # Validates config data - def validate_network_settings(self, data): - if 'modemport' in data: - if not isinstance(data['modemport'], int): - raise ValueError("'modemport' in 'NETWORK' must be an integer.") - - def validate_station_settings(self, data): - for setting in ['mycall', 'mygrid']: - if setting in data and not data[setting]: - raise ValueError(f"'{setting}' in 'STATION' cannot be empty.") - if 'ssid_list' in data and not isinstance(data['ssid_list'], list): - raise ValueError("'ssid_list' in 'STATION' needs to be a list.") - - def validate_audio_settings(self, data): - for setting in ['input_device', 'output_device']: - if setting in data and not isinstance(data[setting], str): - raise ValueError(f"'{setting}' in 'AUDIO' must be a string.") - for setting in ['rx_audio_level', 'tx_audio_level']: - if setting in data and not isinstance(data[setting], int): - raise ValueError(f"'{setting}' in 'AUDIO' must be an integer.") - - def validate_radio_settings(self, data): - if 'radioport' in data and not (data['radioport'] is None or isinstance(data['radioport'], int)): - raise ValueError("'radioport' in 'RADIO' must be None or an integer.") - - def validate_tci_settings(self, data): - if 'tci_ip' in data and not isinstance(data['tci_ip'], str): - raise ValueError("'tci_ip' in 'TCI' must be a string.") - if 'tci_port' in data and not isinstance(data['tci_port'], int): - raise ValueError("'tci_port' in 'TCI' must be an integer.") - - def validate_modem_settings(self, data): - for setting in ['enable_fft', 'enable_fsk', 'enable_low_bandwidth_mode', 'respond_to_cq', 'enable_scatter']: - if setting in data and not isinstance(data[setting], bool): - raise ValueError(f"'{setting}' in 'MODEM' must be a boolean.") - for setting in ['tuning_range_fmax', 'tuning_range_fmin', 'rx_buffer_size', 'tx_delay']: - if setting in data and not isinstance(data[setting], int): - raise ValueError(f"'{setting}' in 'MODEM' must be an integer.") - - def validate_mesh_settings(self, data): - if 'enable_protocol' in data and not isinstance(data['enable_protocol'], bool): - raise ValueError("'enable_protocol' in 'MESH' must be a boolean.") - def validate(self, data): - for section, settings in data.items(): - if section == 'NETWORK': - self.validate_network_settings(settings) - elif section == 'STATION': - self.validate_station_settings(settings) - elif section == 'AUDIO': - self.validate_audio_settings(settings) - elif section == 'RADIO': - self.validate_radio_settings(settings) - elif section == 'TCI': - self.validate_tci_settings(settings) - elif section == 'MESH': - self.validate_mesh_settings(settings) - elif section == 'MODEM': - self.validate_modem_settings(settings) - else: - self.log.warning("wrong config", section=section) - - # converts values of settings from String to Value. - # For example 'False' (type String) will be converted to False (type Bool) - # This is also needed, because configparser reads configs as string - # configparser has a type conversion, but we would have to do this for every - # item. This approach is more flexible - def convert_types(self, config): - for setting in config: - value = config[setting] - - if isinstance(value, dict): - # If the value is a dictionary, apply the function recursively - config[setting] = self.convert_types(value) - - elif isinstance(value, list): - # If the value is a list, iterate through the list - new_list = [] - for item in value: - # Apply the function to each dictionary item in the list - if isinstance(item, dict): - new_list.append(self.convert_types(item)) - else: - new_list.append(item) - config[setting] = new_list - - elif isinstance(value, str): - # Attempt to convert string values - if value.lstrip('-').isdigit(): - config[setting] = int(value) - else: - try: - # Try converting to a float - float_value = float(value) - # If it's actually an integer (like -50.0), convert it to an integer - config[setting] = int(float_value) if float_value.is_integer() else float_value - except ValueError: - # Convert to boolean if applicable - if value.lower() in ['true', 'false']: - config[setting] = value.lower() == 'true' - - return config + for section in data: + for setting in data[section]: + if not isinstance(data[section][setting], self.config_types[section][setting]): + raise ValueError(f"{section}.{setting} must be {self.config_types[section][setting]}") # Handle special setting data type conversion # is_writing means data from a dict being writen to the config file # if False, it means the opposite direction - # TODO check if we can include this in function "convert_types" def handle_setting(self, section, setting, value, is_writing = False): - if (section == 'STATION' and setting == 'ssid_list'): + + if self.config_types[section][setting] == list: if (is_writing): return json.dumps(value) else: return json.loads(value) + + elif self.config_types[section][setting] == bool and not is_writing: + return self.parser.getboolean(section, setting) + + elif self.config_types[section][setting] == int and not is_writing: + return self.parser.getint(section, setting) + else: return value # Sets and writes config data from a dict containing data settings def write(self, data): - # convert datatypes - data = self.convert_types(data) + # Validate config data before writing self.validate(data) for section in data: # init section if it doesn't exist yet - if not section.upper() in self.config.keys(): - self.config[section] = {} + if not section.upper() in self.parser.keys(): + self.parser[section] = {} for setting in data[section]: new_value = self.handle_setting( section, setting, data[section][setting], True) - self.config[section][setting] = str(new_value) + self.parser[section][setting] = str(new_value) # Write config data to file try: with open(self.config_name, 'w') as configfile: - self.config.write(configfile) + self.parser.write(configfile) return self.read() except Exception as conferror: self.log.error("[CFG] reading logfile", e=conferror) @@ -186,10 +141,9 @@ class CONFIG: return False # at first just copy the config as read from file - result = {s:dict(self.config.items(s)) for s in self.config.sections()} - result = self.convert_types(result) + result = {s:dict(self.parser.items(s)) for s in self.parser.sections()} - # handle the special settings (like 'ssid_list') + # handle the special settings for section in result: for setting in result[section]: result[section][setting] = self.handle_setting( diff --git a/tests/test_config.py b/tests/test_config.py index ae7ae85c..8ddc1c7e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -18,7 +18,6 @@ class TestConfigMethods(unittest.TestCase): data = self.config.read() self.assertIsInstance(data, dict) - self.assertIn('NETWORK', data.keys()) self.assertIn('STATION', data.keys()) self.assertIn('AUDIO', data.keys()) self.assertIn('RADIO', data.keys()) @@ -41,15 +40,13 @@ class TestConfigMethods(unittest.TestCase): self.assertEqual(last_conf['STATION']['mycall'], oldcall) def test_validate_data(self): - data = {'NETWORK': {'modemport': "abc"}} + data = {'STATION': {'ssid_list': "abc"}} with self.assertRaises(ValueError): self.config.validate(data) - data = {'NETWORK': {'modemport': "3000"}} + data = {'STATION': {'ssid_list': [1, 2, 3]}} self.assertIsNone(self.config.validate(data)) - - - + if __name__ == '__main__': unittest.main() diff --git a/tests/test_server.py b/tests/test_server.py index f6696c4b..21578da5 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -47,7 +47,7 @@ class TestIntegration(unittest.TestCase): self.assertIn('RADIO', config) def test_config_post(self): - config = {'NETWORK': {'modemport' : '3050'}} + config = {'NETWORK': {'modemport' : 3050}} r = requests.post(self.url + '/config', headers={'Content-type': 'application/json'}, data = json.dumps(config)) @@ -56,7 +56,7 @@ class TestIntegration(unittest.TestCase): r = requests.get(self.url + '/config') self.assertEqual(r.status_code, 200) config = r.json() - self.assertEqual(config['NETWORK']['modemport'], '3050') + self.assertEqual(config['NETWORK']['modemport'], 3050) if __name__ == '__main__': unittest.main()