Index: /branches/ipp-259_genericise_backups/tools/backups/utils/config_parse_helper.py =================================================================== --- /branches/ipp-259_genericise_backups/tools/backups/utils/config_parse_helper.py (revision 40915) +++ /branches/ipp-259_genericise_backups/tools/backups/utils/config_parse_helper.py (revision 40915) @@ -0,0 +1,96 @@ +from enum import Enum +from configparser import ConfigParser +from backups.utils.errors import ConfigError, ValidationError + +# The as configs should be parsed should be pretty uniform. There's a limited +# amount that can be done with the actual loading. + +# Each backup file should be expecting particular lines from a config. So, this +# means a schema can be made for what each backup expects. + + +class SpecialSchemaProcessing(Enum): + """These enums define how a config line should be processed if it is a list + """ + commma_separated = 1 + space_separated = 2 + + +class ConfigSchemaLine(): + """Details the name of key, whether the option is required, + and type expected""" + + def __init__(self, key: str, required: bool, expected_type, special_processing: SpecialSchemaProcessing=None): + self.key = key + self.required = required + self.expected_type = expected_type + if expected_type is list: + if type(special_processing) is not SpecialSchemaProcessing: + raise ValidationError("ValidationError: special_processing must be defined if implementing a list in config") + self.special_processing = special_processing + + +class ConfigSchema(): + """Defines the schema to be used""" + + def __init__(self, section_name, options: [ConfigSchemaLine]): + self.section_name = section_name + self._process_options(options) + + def _process_options(self, options): + self.options = {} + if options is None: + return + if type(options) is not list: + raise ValidationError("ValidationError: ConfigSchema options must be '[ConfigSchemaLine]' or 'None'") + if len(options) == 0: + return + for o in options: + self.add_line(o) + + def add_line(self, line: ConfigSchemaLine): + self.options[line.key] = line + + +def _split_string_to_list(items: str, separator=',') -> [str]: + split_items = items.split(separator) + processed_items = [ i.strip(' ').strip('\n') for i in split_items] + return processed_items + + +def parse_config(config: ConfigParser(), schema: ConfigSchema) -> dict: + # attempt to parse everything as intended. + parsed_items = {} + + section = schema.section_name + for csl in schema.options: + + if config.has_option(section, csl.key): + + if csl.expected_type is bool: + parsed_items[csl.key] = config.getboolean(section, csl.key) + elif csl.expected is int: + parsed_items[csl.key] = config.getint(section, csl.key) + elif csl.expected_type is float: + parsed_items[csl.key] = config.getfloat(section, csl.key) + elif csl.expected_type is str: + parsed_items[csl.key] = config[section][csl.key] + elif csl.expected_type is list: + if csl.special_processing == SpecialSchemaProcessing.commma_separated: + items = config[section][csl.key] + parsed_items[csl.key] = _split_string_to_list(items, ' ') + # parse comma_separated stuff + elif csl.special_processing == SpecialSchemaProcessing.space_separated: + items = config[section][csl.key] + parsed_items[csl.key] = _split_string_to_list(items, ',') + # space separated processing + else: + raise ConfigError(f"ConfigError: Don't know how to process special schema {csl.special_processing}") + else: + raise ConfigError(f"ConfigError: Type '{csl.expected_type}' cannot be processed - need to modify {__name__}") + + else: + if csl.required: + raise ConfigError(f"ConfigError: Key '{csl.key}' is required but was not found in config") + + return parsed_items Index: /branches/ipp-259_genericise_backups/tools/backups/utils/config_parse_helper_test.py =================================================================== --- /branches/ipp-259_genericise_backups/tools/backups/utils/config_parse_helper_test.py (revision 40915) +++ /branches/ipp-259_genericise_backups/tools/backups/utils/config_parse_helper_test.py (revision 40915) @@ -0,0 +1,168 @@ +import pytest + +import backups.utils.config_parse_helper as helper +from backups.utils.errors import ConfigError, ValidationError + + +class TestSpecialSchemaProcessingEnum(): + + def test_all_enums(self): + cs = helper.SpecialSchemaProcessing.commma_separated + assert cs.name == "commma_separated" + assert cs.value == 1 + + ss = helper.SpecialSchemaProcessing.space_separated + assert ss.name == "space_separated" + assert ss.value == 2 + + assert len(list(helper.SpecialSchemaProcessing)) == 2 + + +class TestConfigSchemaLine(): + + def test_initialising_simple_schema_lines(self): + csl = helper.ConfigSchemaLine('key1', True, bool) + assert csl.key == 'key1' + assert csl.required is True + assert csl.expected_type is bool + assert csl.special_processing is None + + def test_initialising_list_in_schema(self): + csl = helper.ConfigSchemaLine('key1', True, list, helper.SpecialSchemaProcessing.commma_separated) + assert csl.key == 'key1' + assert csl.required is True + assert csl.expected_type is list + assert csl.special_processing is helper.SpecialSchemaProcessing.commma_separated + + def test_init_list_without_special_processing_fails(self): + not_a_valid_enum = 3 + with pytest.raises(ValidationError): + helper.ConfigSchemaLine('key1', True, list, not_a_valid_enum) + + +class TestConfigSchema(): + + def test_initalising_schema(self): + csl1 = helper.ConfigSchemaLine(key='is_cool', required=False, expected_type=bool) + csl2 = helper.ConfigSchemaLine('fav_word', True, str) + csl3 = helper.ConfigSchemaLine('some_float', False, float) + + schema = helper.ConfigSchema( 'section_name', [csl1, csl2, csl3] ) + assert schema.options == { 'is_cool' : csl1 + , 'fav_word' : csl2 + , 'some_float' : csl3 + } + + def test_empty_schema_ok(self): + helper.ConfigSchema("ok_with_none", None) + helper.ConfigSchema("ok_with_empty", []) + + def test_schema_fails_if_options_are_not_a_list(self): + with pytest.raises(ValidationError): + helper.ConfigSchema("fails_if_not_a_list", True) + + csl = helper.ConfigSchemaLine('key1', True, bool) + with pytest.raises(ValidationError): + helper.ConfigSchema("even_fails_on_a_single_csl", csl) + + def test_adding_to_schema(self): + config = helper.ConfigSchema("adding", []) + assert config.options == {} + + csl1 = helper.ConfigSchemaLine('key1', True, bool) + csl2 = helper.ConfigSchemaLine('key2', False, int) + csl3 = helper.ConfigSchemaLine('key3', True, str) + config.add_line(csl1) + assert config.options == { 'key1': csl1 } + + config.add_line(csl2) + assert config.options == { 'key1': csl1 + , 'key2': csl2 + } + + config.add_line(csl3) + assert config.options == { 'key1': csl1 + , 'key2': csl2 + , 'key3': csl3 + } + + def test_adding_a_duplicate_key_to_schema_replaces_previous(self): + config = helper.ConfigSchema("adding", []) + assert config.options == {} + + csl_ori = helper.ConfigSchemaLine('key', True, bool) + csl_alt = helper.ConfigSchemaLine('key', False, int) + + config.add_line(csl_ori) + assert config.options == { 'key': csl_ori } + + config.add_line(csl_alt) + assert config.options == { 'key': csl_alt } + + +class TestStringSplitting(): + + def test_split_comma_separated_list(self): + before_parse = "/hello/sir/, /glorius/filename,/some/actual/file.file" + expected = [ "/hello/sir/" + , "/glorius/filename" + , "/some/actual/file.file" + ] + result = helper._split_string_to_list(before_parse, ',') + assert result == expected + + def test_removal_of_newlines_too(self): + before_parse = "\n/hello/sir/, /glorius/filename\n,\n/some/actual/file.file" + expected = [ "/hello/sir/" + , "/glorius/filename" + , "/some/actual/file.file" + ] + result = helper._split_string_to_list(before_parse, ',') + assert result == expected +# def _split_string_to_list(items: str, separator=',') -> [str]: + + def test_space_separated_split(self): + before_parse = "--verbose --progress -a -f -z" + expected = [ "--verbose" + , "--progress" + , "-a" + , "-f" + , "-z" + ] + result = helper._split_string_to_list(before_parse, ' ') + assert result == expected + + +# class TestParsingConfig(): + +# def test_parsing_fails_with_missing_required_key(self): +# assert False is True + +# def test_parsing_schema_with_ints(self): +# assert False is True + +# def test_parsing_schema_with_floats(self): +# assert False is True + +# def test_parsing_schema_with_bools(self): +# assert False is True + +# def test_parsing_schema_with_strs(self): +# assert False is True + +# def test_parsing_schema_with_lists_comma_separated(self): +# assert False is True + +# def test_parsing_schema_with_lists_space_separated(self): +# assert False is True + +# def test_parsing_schema_fails_with_invalid_key(self): +# assert False is True + +# def test_parsing_schema_fails_with_unknown_type(self): +# assert False is True + +# def test_parsing_schema_fails_with_unknown_special_enum(self): +# assert False is True + +# def parse_config(config: ConfigParser(), schema: ConfigSchema) -> dict: