diff --git a/.azure-pipelines/docker-sonic-vs/Dockerfile b/.azure-pipelines/docker-sonic-vs/Dockerfile index 4e0a50e7a406..2b3e63423201 100644 --- a/.azure-pipelines/docker-sonic-vs/Dockerfile +++ b/.azure-pipelines/docker-sonic-vs/Dockerfile @@ -4,4 +4,8 @@ ARG docker_container_name ADD ["wheels", "/wheels"] -RUN pip3 install --no-deps --force-reinstall /wheels/sonic_utilities-1.2-py3-none-any.whl +# Uninstalls only sonic-utilities and does not impact its dependencies +RUN pip3 uninstall -y sonic-utilities + +# Installs sonic-utilities, adds missing dependencies, upgrades out-dated depndencies +RUN pip3 install /wheels/sonic_utilities-1.2-py3-none-any.whl diff --git a/config/main.py b/config/main.py index 244a2570a6d3..6fad33f9c103 100644 --- a/config/main.py +++ b/config/main.py @@ -3,6 +3,7 @@ import click import ipaddress import json +import jsonpatch import netaddr import netifaces import os @@ -11,6 +12,7 @@ import sys import time +from generic_config_updater.generic_updater import GenericUpdater, ConfigFormat from socket import AF_INET, AF_INET6 from minigraph import parse_device_desc_xml from portconfig import get_child_ports @@ -826,7 +828,7 @@ def cache_arp_entries(): if filter_err: click.echo("Could not filter FDB entries prior to reloading") success = False - + # If we are able to successfully cache ARP table info, signal SWSS to restore from our cache # by creating /host/config-reload/needs-restore if success: @@ -987,6 +989,129 @@ def load(filename, yes): log.log_info("'load' executing...") clicommon.run_command(command, display_cmd=True) +@config.command('apply-patch') +@click.argument('patch-file-path', type=str, required=True) +@click.option('-f', '--format', type=click.Choice([e.name for e in ConfigFormat]), + default=ConfigFormat.CONFIGDB.name, + help='format of config of the patch is either ConfigDb(ABNF) or SonicYang') +@click.option('-d', '--dry-run', is_flag=True, default=False, help='test out the command without affecting config state') +@click.option('-v', '--verbose', is_flag=True, default=False, help='print additional details of what the operation is doing') +@click.pass_context +def apply_patch(ctx, patch_file_path, format, dry_run, verbose): + """Apply given patch of updates to Config. A patch is a JsonPatch which follows rfc6902. + This command can be used do partial updates to the config with minimum disruption to running processes. + It allows addition as well as deletion of configs. The patch file represents a diff of ConfigDb(ABNF) + format or SonicYang format. + + : Path to the patch file on the file-system.""" + try: + with open(patch_file_path, 'r') as fh: + text = fh.read() + patch_as_json = json.loads(text) + patch = jsonpatch.JsonPatch(patch_as_json) + + config_format = ConfigFormat[format.upper()] + + GenericUpdater().apply_patch(patch, config_format, verbose, dry_run) + + click.secho("Patch applied successfully.", fg="cyan", underline=True) + except Exception as ex: + click.secho("Failed to apply patch", fg="red", underline=True, err=True) + ctx.fail(ex) + +@config.command() +@click.argument('target-file-path', type=str, required=True) +@click.option('-f', '--format', type=click.Choice([e.name for e in ConfigFormat]), + default=ConfigFormat.CONFIGDB.name, + help='format of target config is either ConfigDb(ABNF) or SonicYang') +@click.option('-d', '--dry-run', is_flag=True, default=False, help='test out the command without affecting config state') +@click.option('-v', '--verbose', is_flag=True, default=False, help='print additional details of what the operation is doing') +@click.pass_context +def replace(ctx, target_file_path, format, dry_run, verbose): + """Replace the whole config with the specified config. The config is replaced with minimum disruption e.g. + if ACL config is different between current and target config only ACL config is updated, and other config/services + such as DHCP will not be affected. + + **WARNING** The target config file should be the whole config, not just the part intended to be updated. + + : Path to the target file on the file-system.""" + try: + with open(target_file_path, 'r') as fh: + target_config_as_text = fh.read() + target_config = json.loads(target_config_as_text) + + config_format = ConfigFormat[format.upper()] + + GenericUpdater().replace(target_config, config_format, verbose, dry_run) + + click.secho("Config replaced successfully.", fg="cyan", underline=True) + except Exception as ex: + click.secho("Failed to replace config", fg="red", underline=True, err=True) + ctx.fail(ex) + +@config.command() +@click.argument('checkpoint-name', type=str, required=True) +@click.option('-d', '--dry-run', is_flag=True, default=False, help='test out the command without affecting config state') +@click.option('-v', '--verbose', is_flag=True, default=False, help='print additional details of what the operation is doing') +@click.pass_context +def rollback(ctx, checkpoint_name, dry_run, verbose): + """Rollback the whole config to the specified checkpoint. The config is rolled back with minimum disruption e.g. + if ACL config is different between current and checkpoint config only ACL config is updated, and other config/services + such as DHCP will not be affected. + + : The checkpoint name, use `config list-checkpoints` command to see available checkpoints.""" + try: + GenericUpdater().rollback(checkpoint_name, verbose, dry_run) + + click.secho("Config rolled back successfully.", fg="cyan", underline=True) + except Exception as ex: + click.secho("Failed to rollback config", fg="red", underline=True, err=True) + ctx.fail(ex) + +@config.command() +@click.argument('checkpoint-name', type=str, required=True) +@click.option('-v', '--verbose', is_flag=True, default=False, help='print additional details of what the operation is doing') +@click.pass_context +def checkpoint(ctx, checkpoint_name, verbose): + """Take a checkpoint of the whole current config with the specified checkpoint name. + + : The checkpoint name, use `config list-checkpoints` command to see available checkpoints.""" + try: + GenericUpdater().checkpoint(checkpoint_name, verbose) + + click.secho("Checkpoint created successfully.", fg="cyan", underline=True) + except Exception as ex: + click.secho("Failed to create a config checkpoint", fg="red", underline=True, err=True) + ctx.fail(ex) + +@config.command('delete-checkpoint') +@click.argument('checkpoint-name', type=str, required=True) +@click.option('-v', '--verbose', is_flag=True, default=False, help='print additional details of what the operation is doing') +@click.pass_context +def delete_checkpoint(ctx, checkpoint_name, verbose): + """Delete a checkpoint with the specified checkpoint name. + + : The checkpoint name, use `config list-checkpoints` command to see available checkpoints.""" + try: + GenericUpdater().delete_checkpoint(checkpoint_name, verbose) + + click.secho("Checkpoint deleted successfully.", fg="cyan", underline=True) + except Exception as ex: + click.secho("Failed to delete config checkpoint", fg="red", underline=True, err=True) + ctx.fail(ex) + +@config.command('list-checkpoints') +@click.option('-v', '--verbose', is_flag=True, default=False, help='print additional details of what the operation is doing') +@click.pass_context +def list_checkpoints(ctx, verbose): + """List the config checkpoints available.""" + try: + checkpoints_list = GenericUpdater().list_checkpoints(verbose) + formatted_output = json.dumps(checkpoints_list, indent=4) + click.echo(formatted_output) + except Exception as ex: + click.secho("Failed to list config checkpoints", fg="red", underline=True, err=True) + ctx.fail(ex) @config.command() @click.option('-y', '--yes', is_flag=True) @@ -2581,8 +2706,8 @@ def add(ctx, interface_name, ip_addr, gw): if interface_name is None: ctx.fail("'interface_name' is None!") - # Add a validation to check this interface is not a member in vlan before - # changing it to a router port + # Add a validation to check this interface is not a member in vlan before + # changing it to a router port vlan_member_table = config_db.get_table('VLAN_MEMBER') if (interface_is_in_vlan(vlan_member_table, interface_name)): click.echo("Interface {} is a member of vlan\nAborting!".format(interface_name)) diff --git a/generic_config_updater/__init__.py b/generic_config_updater/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/generic_config_updater/generic_updater.py b/generic_config_updater/generic_updater.py new file mode 100644 index 000000000000..079d7ab74208 --- /dev/null +++ b/generic_config_updater/generic_updater.py @@ -0,0 +1,339 @@ +import json +import os +from enum import Enum +from .gu_common import GenericConfigUpdaterError, ConfigWrapper, \ + DryRunConfigWrapper, PatchWrapper + +CHECKPOINTS_DIR = "/etc/sonic/checkpoints" +CHECKPOINT_EXT = ".cp.json" + +class ConfigLock: + def acquire_lock(self): + # TODO: Implement ConfigLock + pass + + def release_lock(self): + # TODO: Implement ConfigLock + pass + +class PatchSorter: + def sort(self, patch): + # TODO: Implement patch sorter + raise NotImplementedError("PatchSorter.sort(patch) is not implemented yet") + +class ChangeApplier: + def apply(self, change): + # TODO: Implement change applier + raise NotImplementedError("ChangeApplier.apply(change) is not implemented yet") + +class ConfigFormat(Enum): + CONFIGDB = 1 + SONICYANG = 2 + +class PatchApplier: + def __init__(self, + patchsorter=None, + changeapplier=None, + config_wrapper=None, + patch_wrapper=None): + self.patchsorter = patchsorter if patchsorter is not None else PatchSorter() + self.changeapplier = changeapplier if changeapplier is not None else ChangeApplier() + self.config_wrapper = config_wrapper if config_wrapper is not None else ConfigWrapper() + self.patch_wrapper = patch_wrapper if patch_wrapper is not None else PatchWrapper() + + def apply(self, patch): + # validate patch is only updating tables with yang models + if not(self.patch_wrapper.validate_config_db_patch_has_yang_models(patch)): + raise ValueError(f"Given patch is not valid because it has changes to tables without YANG models") + + # Get old config + old_config = self.config_wrapper.get_config_db_as_json() + + # Generate target config + target_config = self.patch_wrapper.simulate_patch(patch, old_config) + + # Validate target config + if not(self.config_wrapper.validate_config_db_config(target_config)): + raise ValueError(f"Given patch is not valid because it will result in an invalid config") + + # Generate list of changes to apply + changes = self.patchsorter.sort(patch) + + # Apply changes in order + for change in changes: + self.changeapplier.apply(change) + + # Validate config updated successfully + new_config = self.config_wrapper.get_config_db_as_json() + if not(self.patch_wrapper.verify_same_json(target_config, new_config)): + raise GenericConfigUpdaterError(f"After applying patch to config, there are still some parts not updated") + +class ConfigReplacer: + def __init__(self, patch_applier=None, config_wrapper=None, patch_wrapper=None): + self.patch_applier = patch_applier if patch_applier is not None else PatchApplier() + self.config_wrapper = config_wrapper if config_wrapper is not None else ConfigWrapper() + self.patch_wrapper = patch_wrapper if patch_wrapper is not None else PatchWrapper() + + def replace(self, target_config): + if not(self.config_wrapper.validate_config_db_config(target_config)): + raise ValueError(f"The given target config is not valid") + + old_config = self.config_wrapper.get_config_db_as_json() + patch = self.patch_wrapper.generate_patch(old_config, target_config) + + self.patch_applier.apply(patch) + + new_config = self.config_wrapper.get_config_db_as_json() + if not(self.patch_wrapper.verify_same_json(target_config, new_config)): + raise GenericConfigUpdaterError(f"After replacing config, there is still some parts not updated") + +class FileSystemConfigRollbacker: + def __init__(self, + checkpoints_dir=CHECKPOINTS_DIR, + config_replacer=None, + config_wrapper=None): + self.checkpoints_dir = checkpoints_dir + self.config_replacer = config_replacer if config_replacer is not None else ConfigReplacer() + self.config_wrapper = config_wrapper if config_wrapper is not None else ConfigWrapper() + + def rollback(self, checkpoint_name): + if not self._check_checkpoint_exists(checkpoint_name): + raise ValueError(f"Checkpoint '{checkpoint_name}' does not exist") + + target_config = self._get_checkpoint_content(checkpoint_name) + + self.config_replacer.replace(target_config) + + def checkpoint(self, checkpoint_name): + json_content = self.config_wrapper.get_config_db_as_json() + + if not self.config_wrapper.validate_config_db_config(json_content): + raise ValueError(f"Running configs on the device are not valid.") + + path = self._get_checkpoint_full_path(checkpoint_name) + + self._ensure_checkpoints_dir_exists() + + self._save_json_file(path, json_content) + + def list_checkpoints(self): + if not self._checkpoints_dir_exist(): + return [] + + return self._get_checkpoint_names() + + def delete_checkpoint(self, checkpoint_name): + if not self._check_checkpoint_exists(checkpoint_name): + raise ValueError(f"Checkpoint '{checkpoint_name}' does not exist") + + self._delete_checkpoint(checkpoint_name) + + def _ensure_checkpoints_dir_exists(self): + os.makedirs(self.checkpoints_dir, exist_ok=True) + + def _save_json_file(self, path, json_content): + with open(path, "w") as fh: + fh.write(json.dumps(json_content)) + + def _get_checkpoint_content(self, checkpoint_name): + path = self._get_checkpoint_full_path(checkpoint_name) + with open(path) as fh: + text = fh.read() + return json.loads(text) + + def _get_checkpoint_full_path(self, name): + return os.path.join(self.checkpoints_dir, f"{name}{CHECKPOINT_EXT}") + + def _get_checkpoint_names(self): + file_names = [] + for file_name in os.listdir(self.checkpoints_dir): + if file_name.endswith(CHECKPOINT_EXT): + # Remove extension from file name. + # Example assuming ext is '.cp.json', then 'checkpoint1.cp.json' becomes 'checkpoint1' + file_names.append(file_name[:-len(CHECKPOINT_EXT)]) + + return file_names + + def _checkpoints_dir_exist(self): + return os.path.isdir(self.checkpoints_dir) + + def _check_checkpoint_exists(self, name): + path = self._get_checkpoint_full_path(name) + return os.path.isfile(path) + + def _delete_checkpoint(self, name): + path = self._get_checkpoint_full_path(name) + return os.remove(path) + +class Decorator(PatchApplier, ConfigReplacer, FileSystemConfigRollbacker): + def __init__(self, decorated_patch_applier=None, decorated_config_replacer=None, decorated_config_rollbacker=None): + # initing base classes to make LGTM happy + PatchApplier.__init__(self) + ConfigReplacer.__init__(self) + FileSystemConfigRollbacker.__init__(self) + + self.decorated_patch_applier = decorated_patch_applier + self.decorated_config_replacer = decorated_config_replacer + self.decorated_config_rollbacker = decorated_config_rollbacker + + def apply(self, patch): + self.decorated_patch_applier.apply(patch) + + def replace(self, target_config): + self.decorated_config_replacer.replace(target_config) + + def rollback(self, checkpoint_name): + self.decorated_config_rollbacker.rollback(checkpoint_name) + + def checkpoint(self, checkpoint_name): + self.decorated_config_rollbacker.checkpoint(checkpoint_name) + + def list_checkpoints(self): + return self.decorated_config_rollbacker.list_checkpoints() + + def delete_checkpoint(self, checkpoint_name): + self.decorated_config_rollbacker.delete_checkpoint(checkpoint_name) + +class SonicYangDecorator(Decorator): + def __init__(self, patch_wrapper, config_wrapper, decorated_patch_applier=None, decorated_config_replacer=None): + Decorator.__init__(self, decorated_patch_applier, decorated_config_replacer) + + self.patch_wrapper = patch_wrapper + self.config_wrapper = config_wrapper + + def apply(self, patch): + config_db_patch = self.patch_wrapper.convert_sonic_yang_patch_to_config_db_patch(patch) + Decorator.apply(self, config_db_patch) + + def replace(self, target_config): + config_db_target_config = self.config_wrapper.convert_sonic_yang_to_config_db(target_config) + Decorator.replace(self, config_db_target_config) + +class ConfigLockDecorator(Decorator): + def __init__(self, + decorated_patch_applier=None, + decorated_config_replacer=None, + decorated_config_rollbacker=None, + config_lock = ConfigLock()): + Decorator.__init__(self, decorated_patch_applier, decorated_config_replacer, decorated_config_rollbacker) + + self.config_lock = config_lock + + def apply(self, patch): + self.execute_write_action(Decorator.apply, self, patch) + + def replace(self, target_config): + self.execute_write_action(Decorator.replace, self, target_config) + + def rollback(self, checkpoint_name): + self.execute_write_action(Decorator.rollback, self, checkpoint_name) + + def checkpoint(self, checkpoint_name): + self.execute_write_action(Decorator.checkpoint, self, checkpoint_name) + + def execute_write_action(self, action, *args): + self.config_lock.acquire_lock() + action(*args) + self.config_lock.release_lock() + +class GenericUpdateFactory: + def create_patch_applier(self, config_format, verbose, dry_run): + self.init_verbose_logging(verbose) + + config_wrapper = self.get_config_wrapper(dry_run) + + patch_applier = PatchApplier(config_wrapper=config_wrapper) + + patch_wrapper = PatchWrapper(config_wrapper) + + if config_format == ConfigFormat.CONFIGDB: + pass + elif config_format == ConfigFormat.SONICYANG: + patch_applier = SonicYangDecorator( + decorated_patch_applier = patch_applier, patch_wrapper=patch_wrapper, config_wrapper=config_wrapper) + else: + raise ValueError(f"config-format '{config_format}' is not supported") + + if not dry_run: + patch_applier = ConfigLockDecorator(decorated_patch_applier = patch_applier) + + return patch_applier + + def create_config_replacer(self, config_format, verbose, dry_run): + self.init_verbose_logging(verbose) + + config_wrapper = self.get_config_wrapper(dry_run) + + patch_applier = PatchApplier(config_wrapper=config_wrapper) + + patch_wrapper = PatchWrapper(config_wrapper) + + config_replacer = ConfigReplacer(patch_applier=patch_applier, config_wrapper=config_wrapper) + if config_format == ConfigFormat.CONFIGDB: + pass + elif config_format == ConfigFormat.SONICYANG: + config_replacer = SonicYangDecorator( + decorated_config_replacer = config_replacer, patch_wrapper=patch_wrapper, config_wrapper=config_wrapper) + else: + raise ValueError(f"config-format '{config_format}' is not supported") + + if not dry_run: + config_replacer = ConfigLockDecorator(decorated_config_replacer = config_replacer) + + return config_replacer + + def create_config_rollbacker(self, verbose, dry_run=False): + self.init_verbose_logging(verbose) + + config_wrapper = self.get_config_wrapper(dry_run) + + patch_applier = PatchApplier(config_wrapper=config_wrapper) + config_replacer = ConfigReplacer(config_wrapper=config_wrapper, patch_applier=patch_applier) + config_rollbacker = FileSystemConfigRollbacker(config_wrapper = config_wrapper, config_replacer = config_replacer) + + if not dry_run: + config_rollbacker = ConfigLockDecorator(decorated_config_rollbacker = config_rollbacker) + + return config_rollbacker + + def init_verbose_logging(self, verbose): + # TODO: implement verbose logging + # Usually logs have levels such as: error, warning, info, debug. + # By default all log levels should show up to the user, except debug. + # By allowing verbose logging, debug msgs will also be shown to the user. + pass + + def get_config_wrapper(self, dry_run): + if dry_run: + return DryRunConfigWrapper() + else: + return ConfigWrapper() + +class GenericUpdater: + def __init__(self, generic_update_factory=None): + self.generic_update_factory = \ + generic_update_factory if generic_update_factory is not None else GenericUpdateFactory() + + def apply_patch(self, patch, config_format, verbose, dry_run): + patch_applier = self.generic_update_factory.create_patch_applier(config_format, verbose, dry_run) + patch_applier.apply(patch) + + def replace(self, target_config, config_format, verbose, dry_run): + config_replacer = self.generic_update_factory.create_config_replacer(config_format, verbose, dry_run) + config_replacer.replace(target_config) + + def rollback(self, checkpoint_name, verbose, dry_run): + config_rollbacker = self.generic_update_factory.create_config_rollbacker(verbose, dry_run) + config_rollbacker.rollback(checkpoint_name) + + def checkpoint(self, checkpoint_name, verbose): + config_rollbacker = self.generic_update_factory.create_config_rollbacker(verbose) + config_rollbacker.checkpoint(checkpoint_name) + + def delete_checkpoint(self, checkpoint_name, verbose): + config_rollbacker = self.generic_update_factory.create_config_rollbacker(verbose) + config_rollbacker.delete_checkpoint(checkpoint_name) + + def list_checkpoints(self, verbose): + config_rollbacker = self.generic_update_factory.create_config_rollbacker(verbose) + return config_rollbacker.list_checkpoints() diff --git a/generic_config_updater/gu_common.py b/generic_config_updater/gu_common.py new file mode 100644 index 000000000000..2aa6a36d8a52 --- /dev/null +++ b/generic_config_updater/gu_common.py @@ -0,0 +1,176 @@ +import json +import jsonpatch +import sonic_yang +import subprocess +import copy + +YANG_DIR = "/usr/local/yang-models" + +class GenericConfigUpdaterError(Exception): + pass + +class JsonChange: + # TODO: Implement JsonChange + pass + +class ConfigWrapper: + def __init__(self, yang_dir = YANG_DIR): + self.yang_dir = YANG_DIR + + def get_config_db_as_json(self): + text = self._get_config_db_as_text() + return json.loads(text) + + def _get_config_db_as_text(self): + # TODO: Getting configs from CLI is very slow, need to get it from sonic-cffgen directly + cmd = "show runningconfiguration all" + result = subprocess.Popen(cmd, shell=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + text, err = result.communicate() + return_code = result.returncode + if return_code: # non-zero means failure + raise GenericConfigUpdaterError(f"Failed to get running config, Return code: {return_code}, Error: {err}") + return text + + def get_sonic_yang_as_json(self): + config_db_json = self.get_config_db_as_json() + return self.convert_config_db_to_sonic_yang(config_db_json) + + def convert_config_db_to_sonic_yang(self, config_db_as_json): + sy = sonic_yang.SonicYang(self.yang_dir) + sy.loadYangModel() + + # Crop config_db tables that do not have sonic yang models + cropped_config_db_as_json = self.crop_tables_without_yang(config_db_as_json) + + sonic_yang_as_json = dict() + + sy._xlateConfigDBtoYang(cropped_config_db_as_json, sonic_yang_as_json) + + return sonic_yang_as_json + + def convert_sonic_yang_to_config_db(self, sonic_yang_as_json): + sy = sonic_yang.SonicYang(self.yang_dir) + sy.loadYangModel() + + # replace container of the format 'module:table' with just 'table' + new_sonic_yang_json = {} + for module_top in sonic_yang_as_json: + new_sonic_yang_json[module_top] = {} + for container in sonic_yang_as_json[module_top]: + tokens = container.split(':') + if len(tokens) > 2: + raise ValueError(f"Expecting ':' or '
', found {container}") + table = container if len(tokens) == 1 else tokens[1] + new_sonic_yang_json[module_top][table] = sonic_yang_as_json[module_top][container] + + config_db_as_json = dict() + sy.xlateJson = new_sonic_yang_json + sy.revXlateJson = config_db_as_json + sy._revXlateYangtoConfigDB(new_sonic_yang_json, config_db_as_json) + + return config_db_as_json + + def validate_sonic_yang_config(self, sonic_yang_as_json): + config_db_as_json = self.convert_sonic_yang_to_config_db(sonic_yang_as_json) + + sy = sonic_yang.SonicYang(self.yang_dir) + sy.loadYangModel() + + try: + sy.loadData(config_db_as_json) + + sy.validate_data_tree() + return True + except sonic_yang.SonicYangException as ex: + return False + + def validate_config_db_config(self, config_db_as_json): + sy = sonic_yang.SonicYang(self.yang_dir) + sy.loadYangModel() + + try: + tmp_config_db_as_json = copy.deepcopy(config_db_as_json) + + sy.loadData(tmp_config_db_as_json) + + sy.validate_data_tree() + return True + except sonic_yang.SonicYangException as ex: + return False + + def crop_tables_without_yang(self, config_db_as_json): + sy = sonic_yang.SonicYang(self.yang_dir) + sy.loadYangModel() + + sy.jIn = copy.deepcopy(config_db_as_json) + + sy.tablesWithOutYang = dict() + + sy._cropConfigDB() + + return sy.jIn + + def _create_and_connect_config_db(self): + if self.default_config_db_connector != None: + return self.default_config_db_connector + + config_db = ConfigDBConnector() + config_db.connect() + return config_db + +class DryRunConfigWrapper(ConfigWrapper): + # TODO: implement DryRunConfigWrapper + # This class will simulate all read/write operations to ConfigDB on a virtual storage unit. + pass + +class PatchWrapper: + def __init__(self, config_wrapper=None): + self.config_wrapper = config_wrapper if config_wrapper is not None else ConfigWrapper() + + def validate_config_db_patch_has_yang_models(self, patch): + config_db = {} + for operation in patch: + tokens = operation['path'].split('/')[1:] + if len(tokens) == 0: # Modifying whole config_db + tables_dict = {table_name: {} for table_name in operation['value']} + config_db.update(tables_dict) + elif not tokens[0]: # Not empty + raise ValueError("Table name in patch cannot be empty") + else: + config_db[tokens[0]] = {} + + cropped_config_db = self.config_wrapper.crop_tables_without_yang(config_db) + + # valid if no tables dropped during cropping + return len(cropped_config_db.keys()) == len(config_db.keys()) + + def verify_same_json(self, expected, actual): + # patch will be [] if no diff, [] evaluates to False + return not jsonpatch.make_patch(expected, actual) + + def generate_patch(self, current, target): + return jsonpatch.make_patch(current, target) + + def simulate_patch(self, patch, jsonconfig): + return patch.apply(jsonconfig) + + def convert_config_db_patch_to_sonic_yang_patch(self, patch): + if not(self.validate_config_db_patch_has_yang_models(patch)): + raise ValueError(f"Given patch is not valid") + + current_config_db = self.config_wrapper.get_config_db_as_json() + target_config_db = self.simulate_patch(patch, current_config_db) + + current_yang = self.config_wrapper.convert_config_db_to_sonic_yang(current_config_db) + target_yang = self.config_wrapper.convert_config_db_to_sonic_yang(target_config_db) + + return self.generate_patch(current_yang, target_yang) + + def convert_sonic_yang_patch_to_config_db_patch(self, patch): + current_yang = self.config_wrapper.get_sonic_yang_as_json() + target_yang = self.simulate_patch(patch, current_yang) + + current_config_db = self.config_wrapper.convert_sonic_yang_to_config_db(current_yang) + target_config_db = self.config_wrapper.convert_sonic_yang_to_config_db(target_yang) + + return self.generate_patch(current_config_db, target_config_db) diff --git a/setup.py b/setup.py index 02a8d53e381a..d070827667cb 100644 --- a/setup.py +++ b/setup.py @@ -30,6 +30,7 @@ 'counterpoll', 'crm', 'debug', + 'generic_config_updater', 'pfcwd', 'sfputil', 'ssdutil', @@ -157,6 +158,7 @@ 'click==7.0', 'ipaddress==1.0.23', 'jsondiff==1.2.0', + 'jsonpatch==1.32.0', 'm2crypto==0.31.0', 'natsort==6.2.1', # 6.2.1 is the last version which supports Python 2. Can update once we no longer support Python 2 'netaddr==0.8.0', @@ -164,12 +166,13 @@ 'pexpect==4.8.0', 'pyroute2==0.5.14', 'requests==2.25.0', + 'sonic-config-engine', 'sonic-platform-common', 'sonic-py-common', 'sonic-yang-mgmt', 'swsssdk>=2.0.1', 'tabulate==0.8.2', - 'xmltodict==0.12.0' + 'xmltodict==0.12.0', ], setup_requires= [ 'pytest-runner', @@ -178,7 +181,6 @@ tests_require = [ 'pytest', 'mockredispy>=2.9.3', - 'sonic-config-engine', 'deepdiff==5.2.3' ], classifiers=[ diff --git a/tests/config_test.py b/tests/config_test.py index 381ca803041b..32ecc5bdef35 100644 --- a/tests/config_test.py +++ b/tests/config_test.py @@ -3,6 +3,9 @@ import os import traceback import json +import jsonpatch +import sys +import unittest from unittest import mock import click @@ -11,6 +14,10 @@ from sonic_py_common import device_info from utilities_common.db import Db +from generic_config_updater.generic_updater import ConfigFormat + +import config.main as config + load_minigraph_command_output="""\ Stopping SONiC target ... Running command: /usr/local/bin/sonic-cfggen -H -m --write-to-db @@ -150,3 +157,615 @@ def teardown_class(cls): from .mock_tables import mock_single_asic importlib.reload(mock_single_asic) dbconnector.load_namespace_config() + +class TestGenericUpdateCommands(unittest.TestCase): + def setUp(self): + os.environ['UTILITIES_UNIT_TESTING'] = "1" + self.runner = CliRunner() + self.any_patch_as_json = [{"op":"remove", "path":"/PORT"}] + self.any_patch = jsonpatch.JsonPatch(self.any_patch_as_json) + self.any_patch_as_text = json.dumps(self.any_patch_as_json) + self.any_path = '/usr/admin/patch.json-patch' + self.any_target_config = {"PORT": {}} + self.any_target_config_as_text = json.dumps(self.any_target_config) + self.any_checkpoint_name = "any_checkpoint_name" + self.any_checkpoints_list = ["checkpoint1", "checkpoint2", "checkpoint3"] + self.any_checkpoints_list_as_text = json.dumps(self.any_checkpoints_list, indent=4) + + def test_apply_patch__no_params__get_required_params_error_msg(self): + # Arrange + unexpected_exit_code = 0 + expected_output = "Error: Missing argument \"PATCH_FILE_PATH\"" + + # Act + result = self.runner.invoke(config.config.commands["apply-patch"]) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_apply_patch__help__gets_help_msg(self): + # Arrange + expected_exit_code = 0 + expected_output = "Options:" # this indicates the options are listed + + # Act + result = self.runner.invoke(config.config.commands["apply-patch"], ['--help']) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_apply_patch__only_required_params__default_values_used_for_optional_params(self): + # Arrange + expected_exit_code = 0 + expected_output = "Patch applied successfully" + expected_call_with_default_values = mock.call(self.any_patch, ConfigFormat.CONFIGDB, False, False) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + with mock.patch('builtins.open', mock.mock_open(read_data=self.any_patch_as_text)): + + # Act + result = self.runner.invoke(config.config.commands["apply-patch"], [self.any_path], catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.apply_patch.assert_called_once() + mock_generic_updater.apply_patch.assert_has_calls([expected_call_with_default_values]) + + def test_apply_patch__all_optional_params_non_default__non_default_values_used(self): + # Arrange + expected_exit_code = 0 + expected_output = "Patch applied successfully" + expected_call_with_non_default_values = mock.call(self.any_patch, ConfigFormat.SONICYANG, True, True) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + with mock.patch('builtins.open', mock.mock_open(read_data=self.any_patch_as_text)): + + # Act + result = self.runner.invoke(config.config.commands["apply-patch"], + [self.any_path, + "--format", ConfigFormat.SONICYANG.name, + "--dry-run", + "--verbose"], + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.apply_patch.assert_called_once() + mock_generic_updater.apply_patch.assert_has_calls([expected_call_with_non_default_values]) + + def test_apply_patch__exception_thrown__error_displayed_error_code_returned(self): + # Arrange + unexpected_exit_code = 0 + any_error_message = "any_error_message" + mock_generic_updater = mock.Mock() + mock_generic_updater.apply_patch.side_effect = Exception(any_error_message) + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + with mock.patch('builtins.open', mock.mock_open(read_data=self.any_patch_as_text)): + + # Act + result = self.runner.invoke(config.config.commands["apply-patch"], + [self.any_path], + catch_exceptions=False) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(any_error_message in result.output) + + def test_apply_patch__optional_parameters_passed_correctly(self): + self.validate_apply_patch_optional_parameter( + ["--format", ConfigFormat.SONICYANG.name], + mock.call(self.any_patch, ConfigFormat.SONICYANG, False, False)) + self.validate_apply_patch_optional_parameter( + ["--verbose"], + mock.call(self.any_patch, ConfigFormat.CONFIGDB, True, False)) + self.validate_apply_patch_optional_parameter( + ["--dry-run"], + mock.call(self.any_patch, ConfigFormat.CONFIGDB, False, True)) + + def validate_apply_patch_optional_parameter(self, param_args, expected_call): + # Arrange + expected_exit_code = 0 + expected_output = "Patch applied successfully" + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + with mock.patch('builtins.open', mock.mock_open(read_data=self.any_patch_as_text)): + + # Act + result = self.runner.invoke(config.config.commands["apply-patch"], + [self.any_path] + param_args, + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.apply_patch.assert_called_once() + mock_generic_updater.apply_patch.assert_has_calls([expected_call]) + + def test_replace__no_params__get_required_params_error_msg(self): + # Arrange + unexpected_exit_code = 0 + expected_output = "Error: Missing argument \"TARGET_FILE_PATH\"" + + # Act + result = self.runner.invoke(config.config.commands["replace"]) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_replace__help__gets_help_msg(self): + # Arrange + expected_exit_code = 0 + expected_output = "Options:" # this indicates the options are listed + + # Act + result = self.runner.invoke(config.config.commands["replace"], ['--help']) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_replace__only_required_params__default_values_used_for_optional_params(self): + # Arrange + expected_exit_code = 0 + expected_output = "Config replaced successfully" + expected_call_with_default_values = mock.call(self.any_target_config, ConfigFormat.CONFIGDB, False, False) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + with mock.patch('builtins.open', mock.mock_open(read_data=self.any_target_config_as_text)): + + # Act + result = self.runner.invoke(config.config.commands["replace"], [self.any_path], catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.replace.assert_called_once() + mock_generic_updater.replace.assert_has_calls([expected_call_with_default_values]) + + def test_replace__all_optional_params_non_default__non_default_values_used(self): + # Arrange + expected_exit_code = 0 + expected_output = "Config replaced successfully" + expected_call_with_non_default_values = mock.call(self.any_target_config, ConfigFormat.SONICYANG, True, True) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + with mock.patch('builtins.open', mock.mock_open(read_data=self.any_target_config_as_text)): + + # Act + result = self.runner.invoke(config.config.commands["replace"], + [self.any_path, + "--format", ConfigFormat.SONICYANG.name, + "--dry-run", + "--verbose"], + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.replace.assert_called_once() + mock_generic_updater.replace.assert_has_calls([expected_call_with_non_default_values]) + + def test_replace__exception_thrown__error_displayed_error_code_returned(self): + # Arrange + unexpected_exit_code = 0 + any_error_message = "any_error_message" + mock_generic_updater = mock.Mock() + mock_generic_updater.replace.side_effect = Exception(any_error_message) + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + with mock.patch('builtins.open', mock.mock_open(read_data=self.any_target_config_as_text)): + + # Act + result = self.runner.invoke(config.config.commands["replace"], + [self.any_path], + catch_exceptions=False) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(any_error_message in result.output) + + def test_replace__optional_parameters_passed_correctly(self): + self.validate_replace_optional_parameter( + ["--format", ConfigFormat.SONICYANG.name], + mock.call(self.any_target_config, ConfigFormat.SONICYANG, False, False)) + self.validate_replace_optional_parameter( + ["--verbose"], + mock.call(self.any_target_config, ConfigFormat.CONFIGDB, True, False)) + self.validate_replace_optional_parameter( + ["--dry-run"], + mock.call(self.any_target_config, ConfigFormat.CONFIGDB, False, True)) + + def validate_replace_optional_parameter(self, param_args, expected_call): + # Arrange + expected_exit_code = 0 + expected_output = "Config replaced successfully" + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + with mock.patch('builtins.open', mock.mock_open(read_data=self.any_target_config_as_text)): + + # Act + result = self.runner.invoke(config.config.commands["replace"], + [self.any_path] + param_args, + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.replace.assert_called_once() + mock_generic_updater.replace.assert_has_calls([expected_call]) + + def test_rollback__no_params__get_required_params_error_msg(self): + # Arrange + unexpected_exit_code = 0 + expected_output = "Error: Missing argument \"CHECKPOINT_NAME\"" + + # Act + result = self.runner.invoke(config.config.commands["rollback"]) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_rollback__help__gets_help_msg(self): + # Arrange + expected_exit_code = 0 + expected_output = "Options:" # this indicates the options are listed + + # Act + result = self.runner.invoke(config.config.commands["rollback"], ['--help']) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_rollback__only_required_params__default_values_used_for_optional_params(self): + # Arrange + expected_exit_code = 0 + expected_output = "Config rolled back successfully" + expected_call_with_default_values = mock.call(self.any_checkpoint_name, False, False) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + # Act + result = self.runner.invoke(config.config.commands["rollback"], [self.any_checkpoint_name], catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.rollback.assert_called_once() + mock_generic_updater.rollback.assert_has_calls([expected_call_with_default_values]) + + def test_rollback__all_optional_params_non_default__non_default_values_used(self): + # Arrange + expected_exit_code = 0 + expected_output = "Config rolled back successfully" + expected_call_with_non_default_values = mock.call(self.any_checkpoint_name, True, True) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + + # Act + result = self.runner.invoke(config.config.commands["rollback"], + [self.any_checkpoint_name, + "--dry-run", + "--verbose"], + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.rollback.assert_called_once() + mock_generic_updater.rollback.assert_has_calls([expected_call_with_non_default_values]) + + def test_rollback__exception_thrown__error_displayed_error_code_returned(self): + # Arrange + unexpected_exit_code = 0 + any_error_message = "any_error_message" + mock_generic_updater = mock.Mock() + mock_generic_updater.rollback.side_effect = Exception(any_error_message) + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + + # Act + result = self.runner.invoke(config.config.commands["rollback"], + [self.any_checkpoint_name], + catch_exceptions=False) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(any_error_message in result.output) + + def test_rollback__optional_parameters_passed_correctly(self): + self.validate_rollback_optional_parameter( + ["--verbose"], + mock.call(self.any_checkpoint_name, True, False)) + self.validate_rollback_optional_parameter( + ["--dry-run"], + mock.call(self.any_checkpoint_name, False, True)) + + def validate_rollback_optional_parameter(self, param_args, expected_call): + # Arrange + expected_exit_code = 0 + expected_output = "Config rolled back successfully" + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + # Act + result = self.runner.invoke(config.config.commands["rollback"], + [self.any_checkpoint_name] + param_args, + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.rollback.assert_called_once() + mock_generic_updater.rollback.assert_has_calls([expected_call]) + + def test_checkpoint__no_params__get_required_params_error_msg(self): + # Arrange + unexpected_exit_code = 0 + expected_output = "Error: Missing argument \"CHECKPOINT_NAME\"" + + # Act + result = self.runner.invoke(config.config.commands["checkpoint"]) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_checkpoint__help__gets_help_msg(self): + # Arrange + expected_exit_code = 0 + expected_output = "Options:" # this indicates the options are listed + + # Act + result = self.runner.invoke(config.config.commands["checkpoint"], ['--help']) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_checkpoint__only_required_params__default_values_used_for_optional_params(self): + # Arrange + expected_exit_code = 0 + expected_output = "Checkpoint created successfully" + expected_call_with_default_values = mock.call(self.any_checkpoint_name, False) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + # Act + result = self.runner.invoke(config.config.commands["checkpoint"], [self.any_checkpoint_name], catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.checkpoint.assert_called_once() + mock_generic_updater.checkpoint.assert_has_calls([expected_call_with_default_values]) + + def test_checkpoint__all_optional_params_non_default__non_default_values_used(self): + # Arrange + expected_exit_code = 0 + expected_output = "Checkpoint created successfully" + expected_call_with_non_default_values = mock.call(self.any_checkpoint_name, True) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + + # Act + result = self.runner.invoke(config.config.commands["checkpoint"], + [self.any_checkpoint_name, + "--verbose"], + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.checkpoint.assert_called_once() + mock_generic_updater.checkpoint.assert_has_calls([expected_call_with_non_default_values]) + + def test_checkpoint__exception_thrown__error_displayed_error_code_returned(self): + # Arrange + unexpected_exit_code = 0 + any_error_message = "any_error_message" + mock_generic_updater = mock.Mock() + mock_generic_updater.checkpoint.side_effect = Exception(any_error_message) + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + + # Act + result = self.runner.invoke(config.config.commands["checkpoint"], + [self.any_checkpoint_name], + catch_exceptions=False) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(any_error_message in result.output) + + def test_checkpoint__optional_parameters_passed_correctly(self): + self.validate_checkpoint_optional_parameter( + ["--verbose"], + mock.call(self.any_checkpoint_name, True)) + + def validate_checkpoint_optional_parameter(self, param_args, expected_call): + # Arrange + expected_exit_code = 0 + expected_output = "Checkpoint created successfully" + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + # Act + result = self.runner.invoke(config.config.commands["checkpoint"], + [self.any_checkpoint_name] + param_args, + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.checkpoint.assert_called_once() + mock_generic_updater.checkpoint.assert_has_calls([expected_call]) + + def test_delete_checkpoint__no_params__get_required_params_error_msg(self): + # Arrange + unexpected_exit_code = 0 + expected_output = "Error: Missing argument \"CHECKPOINT_NAME\"" + + # Act + result = self.runner.invoke(config.config.commands["delete-checkpoint"]) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_delete_checkpoint__help__gets_help_msg(self): + # Arrange + expected_exit_code = 0 + expected_output = "Options:" # this indicates the options are listed + + # Act + result = self.runner.invoke(config.config.commands["delete-checkpoint"], ['--help']) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_delete_checkpoint__only_required_params__default_values_used_for_optional_params(self): + # Arrange + expected_exit_code = 0 + expected_output = "Checkpoint deleted successfully" + expected_call_with_default_values = mock.call(self.any_checkpoint_name, False) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + # Act + result = self.runner.invoke(config.config.commands["delete-checkpoint"], [self.any_checkpoint_name], catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.delete_checkpoint.assert_called_once() + mock_generic_updater.delete_checkpoint.assert_has_calls([expected_call_with_default_values]) + + def test_delete_checkpoint__all_optional_params_non_default__non_default_values_used(self): + # Arrange + expected_exit_code = 0 + expected_output = "Checkpoint deleted successfully" + expected_call_with_non_default_values = mock.call(self.any_checkpoint_name, True) + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + + # Act + result = self.runner.invoke(config.config.commands["delete-checkpoint"], + [self.any_checkpoint_name, + "--verbose"], + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.delete_checkpoint.assert_called_once() + mock_generic_updater.delete_checkpoint.assert_has_calls([expected_call_with_non_default_values]) + + def test_delete_checkpoint__exception_thrown__error_displayed_error_code_returned(self): + # Arrange + unexpected_exit_code = 0 + any_error_message = "any_error_message" + mock_generic_updater = mock.Mock() + mock_generic_updater.delete_checkpoint.side_effect = Exception(any_error_message) + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + + # Act + result = self.runner.invoke(config.config.commands["delete-checkpoint"], + [self.any_checkpoint_name], + catch_exceptions=False) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(any_error_message in result.output) + + def test_delete_checkpoint__optional_parameters_passed_correctly(self): + self.validate_delete_checkpoint_optional_parameter( + ["--verbose"], + mock.call(self.any_checkpoint_name, True)) + + def validate_delete_checkpoint_optional_parameter(self, param_args, expected_call): + # Arrange + expected_exit_code = 0 + expected_output = "Checkpoint deleted successfully" + mock_generic_updater = mock.Mock() + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + # Act + result = self.runner.invoke(config.config.commands["delete-checkpoint"], + [self.any_checkpoint_name] + param_args, + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.delete_checkpoint.assert_called_once() + mock_generic_updater.delete_checkpoint.assert_has_calls([expected_call]) + + def test_list_checkpoints__help__gets_help_msg(self): + # Arrange + expected_exit_code = 0 + expected_output = "Options:" # this indicates the options are listed + + # Act + result = self.runner.invoke(config.config.commands["list-checkpoints"], ['--help']) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + + def test_list_checkpoints__all_optional_params_non_default__non_default_values_used(self): + # Arrange + expected_exit_code = 0 + expected_output = self.any_checkpoints_list_as_text + expected_call_with_non_default_values = mock.call(True) + mock_generic_updater = mock.Mock() + mock_generic_updater.list_checkpoints.return_value = self.any_checkpoints_list + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + + # Act + result = self.runner.invoke(config.config.commands["list-checkpoints"], + ["--verbose"], + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.list_checkpoints.assert_called_once() + mock_generic_updater.list_checkpoints.assert_has_calls([expected_call_with_non_default_values]) + + def test_list_checkpoints__exception_thrown__error_displayed_error_code_returned(self): + # Arrange + unexpected_exit_code = 0 + any_error_message = "any_error_message" + mock_generic_updater = mock.Mock() + mock_generic_updater.list_checkpoints.side_effect = Exception(any_error_message) + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + + # Act + result = self.runner.invoke(config.config.commands["list-checkpoints"], + catch_exceptions=False) + + # Assert + self.assertNotEqual(unexpected_exit_code, result.exit_code) + self.assertTrue(any_error_message in result.output) + + def test_list_checkpoints__optional_parameters_passed_correctly(self): + self.validate_list_checkpoints_optional_parameter( + ["--verbose"], + mock.call(True)) + + def validate_list_checkpoints_optional_parameter(self, param_args, expected_call): + # Arrange + expected_exit_code = 0 + expected_output = self.any_checkpoints_list_as_text + mock_generic_updater = mock.Mock() + mock_generic_updater.list_checkpoints.return_value = self.any_checkpoints_list + with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater): + # Act + result = self.runner.invoke(config.config.commands["list-checkpoints"], + param_args, + catch_exceptions=False) + + # Assert + self.assertEqual(expected_exit_code, result.exit_code) + self.assertTrue(expected_output in result.output) + mock_generic_updater.list_checkpoints.assert_called_once() + mock_generic_updater.list_checkpoints.assert_has_calls([expected_call]) diff --git a/tests/generic_config_updater/__init__.py b/tests/generic_config_updater/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/generic_config_updater/files/config_db_after_multi_patch.json b/tests/generic_config_updater/files/config_db_after_multi_patch.json new file mode 100644 index 000000000000..042bf1d51b40 --- /dev/null +++ b/tests/generic_config_updater/files/config_db_after_multi_patch.json @@ -0,0 +1,122 @@ +{ + "VLAN_MEMBER": { + "Vlan1000|Ethernet0": { + "tagging_mode": "untagged" + }, + "Vlan1000|Ethernet4": { + "tagging_mode": "untagged" + }, + "Vlan1000|Ethernet8": { + "tagging_mode": "untagged" + }, + "Vlan100|Ethernet2": { + "tagging_mode": "untagged" + }, + "Vlan100|Ethernet3": { + "tagging_mode": "untagged" + }, + "Vlan100|Ethernet1": { + "tagging_mode": "untagged" + } + }, + "VLAN": { + "Vlan1000": { + "vlanid": "1000", + "dhcp_servers": [ + "192.0.0.1", + "192.0.0.2", + "192.0.0.3", + "192.0.0.4" + ] + } + }, + "ACL_TABLE": { + "NO-NSW-PACL-V4": { + "type": "L3", + "policy_desc": "NO-NSW-PACL-V4", + "ports": [ + "Ethernet0", + "Ethernet1", + "Ethernet2", + "Ethernet3" + ] + }, + "DATAACL": { + "policy_desc": "DATAACL", + "ports": [ + "Ethernet4" + ], + "stage": "ingress", + "type": "L3" + }, + "EVERFLOW": { + "policy_desc": "EVERFLOW", + "ports": [ + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRROR" + }, + "EVERFLOWV6": { + "policy_desc": "EVERFLOWV6", + "ports": [ + "Ethernet4", + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRRORV6" + } + }, + "PORT": { + "Ethernet0": { + "alias": "Eth1/1", + "lanes": "65", + "description": "", + "speed": "10000" + }, + "Ethernet4": { + "admin_status": "up", + "alias": "fortyGigE0/4", + "description": "Servers0:eth0", + "index": "1", + "lanes": "29,30,31,32", + "mtu": "9100", + "pfc_asym": "off", + "speed": "40000" + }, + "Ethernet8": { + "admin_status": "up", + "alias": "fortyGigE0/8", + "description": "Servers1:eth0", + "index": "2", + "lanes": "33,34,35,36", + "mtu": "9100", + "pfc_asym": "off", + "speed": "40000" + }, + "Ethernet3": { + "alias": "Eth1/4", + "lanes": "68", + "description": "", + "speed": "10000" + }, + "Ethernet1": { + "alias": "Eth1/2", + "lanes": "66", + "description": "", + "speed": "10000" + }, + "Ethernet2": { + "alias": "Eth1/3", + "lanes": "67", + "description": "", + "speed": "10000" + } + }, + "TABLE_WITHOUT_YANG": { + "Item1": { + "key11": "value11", + "key12": "value12" + } + } +} \ No newline at end of file diff --git a/tests/generic_config_updater/files/config_db_as_json.json b/tests/generic_config_updater/files/config_db_as_json.json new file mode 100644 index 000000000000..02fb7c7e6af6 --- /dev/null +++ b/tests/generic_config_updater/files/config_db_as_json.json @@ -0,0 +1,92 @@ +{ + "VLAN_MEMBER": { + "Vlan1000|Ethernet0": { + "tagging_mode": "untagged" + }, + "Vlan1000|Ethernet4": { + "tagging_mode": "untagged" + }, + "Vlan1000|Ethernet8": { + "tagging_mode": "untagged" + } + }, + "VLAN": { + "Vlan1000": { + "vlanid": "1000", + "dhcp_servers": [ + "192.0.0.1", + "192.0.0.2", + "192.0.0.3", + "192.0.0.4" + ] + } + }, + "ACL_TABLE": { + "NO-NSW-PACL-V4": { + "type": "L3", + "policy_desc": "NO-NSW-PACL-V4", + "ports": [ + "Ethernet0" + ] + }, + "DATAACL": { + "policy_desc": "DATAACL", + "ports": [ + "Ethernet4" + ], + "stage": "ingress", + "type": "L3" + }, + "EVERFLOW": { + "policy_desc": "EVERFLOW", + "ports": [ + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRROR" + }, + "EVERFLOWV6": { + "policy_desc": "EVERFLOWV6", + "ports": [ + "Ethernet4", + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRRORV6" + } + }, + "PORT": { + "Ethernet0": { + "alias": "Eth1", + "lanes": "65, 66, 67, 68", + "description": "Ethernet0 100G link", + "speed": "100000" + }, + "Ethernet4": { + "admin_status": "up", + "alias": "fortyGigE0/4", + "description": "Servers0:eth0", + "index": "1", + "lanes": "29,30,31,32", + "mtu": "9100", + "pfc_asym": "off", + "speed": "40000" + }, + "Ethernet8": { + "admin_status": "up", + "alias": "fortyGigE0/8", + "description": "Servers1:eth0", + "index": "2", + "lanes": "33,34,35,36", + "mtu": "9100", + "pfc_asym": "off", + "speed": "40000" + } + }, + "TABLE_WITHOUT_YANG": { + "Item1": { + "key11": "value11", + "key12": "value12" + } + } +} diff --git a/tests/generic_config_updater/files/config_db_as_json_invalid.json b/tests/generic_config_updater/files/config_db_as_json_invalid.json new file mode 100644 index 000000000000..a2cfdc91df2b --- /dev/null +++ b/tests/generic_config_updater/files/config_db_as_json_invalid.json @@ -0,0 +1,7 @@ +{ + "VLAN_MEMBER": { + "Vlan1000|Ethernet8": { + "tagging_mode": "untagged" + } + } +} diff --git a/tests/generic_config_updater/files/cropped_config_db_as_json.json b/tests/generic_config_updater/files/cropped_config_db_as_json.json new file mode 100644 index 000000000000..261e912c7124 --- /dev/null +++ b/tests/generic_config_updater/files/cropped_config_db_as_json.json @@ -0,0 +1,86 @@ +{ + "VLAN_MEMBER": { + "Vlan1000|Ethernet0": { + "tagging_mode": "untagged" + }, + "Vlan1000|Ethernet4": { + "tagging_mode": "untagged" + }, + "Vlan1000|Ethernet8": { + "tagging_mode": "untagged" + } + }, + "VLAN": { + "Vlan1000": { + "vlanid": "1000", + "dhcp_servers": [ + "192.0.0.1", + "192.0.0.2", + "192.0.0.3", + "192.0.0.4" + ] + } + }, + "ACL_TABLE": { + "NO-NSW-PACL-V4": { + "type": "L3", + "policy_desc": "NO-NSW-PACL-V4", + "ports": [ + "Ethernet0" + ] + }, + "DATAACL": { + "policy_desc": "DATAACL", + "ports": [ + "Ethernet4" + ], + "stage": "ingress", + "type": "L3" + }, + "EVERFLOW": { + "policy_desc": "EVERFLOW", + "ports": [ + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRROR" + }, + "EVERFLOWV6": { + "policy_desc": "EVERFLOWV6", + "ports": [ + "Ethernet4", + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRRORV6" + } + }, + "PORT": { + "Ethernet0": { + "alias": "Eth1", + "lanes": "65, 66, 67, 68", + "description": "Ethernet0 100G link", + "speed": "100000" + }, + "Ethernet4": { + "admin_status": "up", + "alias": "fortyGigE0/4", + "description": "Servers0:eth0", + "index": "1", + "lanes": "29,30,31,32", + "mtu": "9100", + "pfc_asym": "off", + "speed": "40000" + }, + "Ethernet8": { + "admin_status": "up", + "alias": "fortyGigE0/8", + "description": "Servers1:eth0", + "index": "2", + "lanes": "33,34,35,36", + "mtu": "9100", + "pfc_asym": "off", + "speed": "40000" + } + } +} diff --git a/tests/generic_config_updater/files/multi_operation_config_db_patch.json-patch b/tests/generic_config_updater/files/multi_operation_config_db_patch.json-patch new file mode 100644 index 000000000000..8eddd7a19dad --- /dev/null +++ b/tests/generic_config_updater/files/multi_operation_config_db_patch.json-patch @@ -0,0 +1,88 @@ +[ + { + "op": "add", + "path": "/PORT/Ethernet3", + "value": { + "alias": "Eth1/4", + "lanes": "68", + "description": "", + "speed": "10000" + } + }, + { + "op": "add", + "path": "/PORT/Ethernet1", + "value": { + "alias": "Eth1/2", + "lanes": "66", + "description": "", + "speed": "10000" + } + }, + { + "op": "add", + "path": "/PORT/Ethernet2", + "value": { + "alias": "Eth1/3", + "lanes": "67", + "description": "", + "speed": "10000" + } + }, + { + "op": "replace", + "path": "/PORT/Ethernet0/lanes", + "value": "65" + }, + { + "op": "replace", + "path": "/PORT/Ethernet0/alias", + "value": "Eth1/1" + }, + { + "op": "replace", + "path": "/PORT/Ethernet0/description", + "value": "" + }, + { + "op": "replace", + "path": "/PORT/Ethernet0/speed", + "value": "10000" + }, + { + "op": "add", + "path": "/VLAN_MEMBER/Vlan100|Ethernet2", + "value": { + "tagging_mode": "untagged" + } + }, + { + "op": "add", + "path": "/VLAN_MEMBER/Vlan100|Ethernet3", + "value": { + "tagging_mode": "untagged" + } + }, + { + "op": "add", + "path": "/VLAN_MEMBER/Vlan100|Ethernet1", + "value": { + "tagging_mode": "untagged" + } + }, + { + "op": "add", + "path": "/ACL_TABLE/NO-NSW-PACL-V4/ports/1", + "value": "Ethernet1" + }, + { + "op": "add", + "path": "/ACL_TABLE/NO-NSW-PACL-V4/ports/2", + "value": "Ethernet2" + }, + { + "op": "add", + "path": "/ACL_TABLE/NO-NSW-PACL-V4/ports/3", + "value": "Ethernet3" + } +] diff --git a/tests/generic_config_updater/files/multi_operation_sonic_yang_patch.json-patch b/tests/generic_config_updater/files/multi_operation_sonic_yang_patch.json-patch new file mode 100644 index 000000000000..f7005bb4a047 --- /dev/null +++ b/tests/generic_config_updater/files/multi_operation_sonic_yang_patch.json-patch @@ -0,0 +1,97 @@ +[ + { + "op": "add", + "path": "/sonic-vlan:sonic-vlan/sonic-vlan:VLAN_MEMBER/VLAN_MEMBER_LIST/3", + "value": { + "name": "Vlan100", + "port": "Ethernet2", + "tagging_mode": "untagged" + } + }, + { + "op": "add", + "path": "/sonic-vlan:sonic-vlan/sonic-vlan:VLAN_MEMBER/VLAN_MEMBER_LIST/4", + "value": { + "name": "Vlan100", + "port": "Ethernet3", + "tagging_mode": "untagged" + } + }, + { + "op": "add", + "path": "/sonic-vlan:sonic-vlan/sonic-vlan:VLAN_MEMBER/VLAN_MEMBER_LIST/5", + "value": { + "name": "Vlan100", + "port": "Ethernet1", + "tagging_mode": "untagged" + } + }, + { + "op": "replace", + "path": "/sonic-port:sonic-port/sonic-port:PORT/PORT_LIST/0/lanes", + "value": "65" + }, + { + "op": "replace", + "path": "/sonic-port:sonic-port/sonic-port:PORT/PORT_LIST/0/alias", + "value": "Eth1/1" + }, + { + "op": "replace", + "path": "/sonic-port:sonic-port/sonic-port:PORT/PORT_LIST/0/speed", + "value": 10000 + }, + { + "op": "replace", + "path": "/sonic-port:sonic-port/sonic-port:PORT/PORT_LIST/0/description", + "value": "" + }, + { + "op": "add", + "path": "/sonic-port:sonic-port/sonic-port:PORT/PORT_LIST/3", + "value": { + "name": "Ethernet3", + "alias": "Eth1/4", + "lanes": "68", + "description": "", + "speed": 10000 + } + }, + { + "op": "add", + "path": "/sonic-port:sonic-port/sonic-port:PORT/PORT_LIST/4", + "value": { + "name": "Ethernet1", + "alias": "Eth1/2", + "lanes": "66", + "description": "", + "speed": 10000 + } + }, + { + "op": "add", + "path": "/sonic-port:sonic-port/sonic-port:PORT/PORT_LIST/5", + "value": { + "name": "Ethernet2", + "alias": "Eth1/3", + "lanes": "67", + "description": "", + "speed": 10000 + } + }, + { + "op": "add", + "path": "/sonic-acl:sonic-acl/sonic-acl:ACL_TABLE/ACL_TABLE_LIST/0/ports/1", + "value": "Ethernet1" + }, + { + "op": "add", + "path": "/sonic-acl:sonic-acl/sonic-acl:ACL_TABLE/ACL_TABLE_LIST/0/ports/2", + "value": "Ethernet2" + }, + { + "op": "add", + "path": "/sonic-acl:sonic-acl/sonic-acl:ACL_TABLE/ACL_TABLE_LIST/0/ports/3", + "value": "Ethernet3" + } +] diff --git a/tests/generic_config_updater/files/single_operation_config_db_patch.json-patch b/tests/generic_config_updater/files/single_operation_config_db_patch.json-patch new file mode 100644 index 000000000000..7cc0967bf046 --- /dev/null +++ b/tests/generic_config_updater/files/single_operation_config_db_patch.json-patch @@ -0,0 +1,6 @@ +[ + { + "op": "remove", + "path": "/VLAN_MEMBER/Vlan1000|Ethernet8" + } +] diff --git a/tests/generic_config_updater/files/single_operation_sonic_yang_patch.json-patch b/tests/generic_config_updater/files/single_operation_sonic_yang_patch.json-patch new file mode 100644 index 000000000000..5a4656049617 --- /dev/null +++ b/tests/generic_config_updater/files/single_operation_sonic_yang_patch.json-patch @@ -0,0 +1,6 @@ +[ + { + "op": "remove", + "path": "/sonic-vlan:sonic-vlan/sonic-vlan:VLAN_MEMBER/VLAN_MEMBER_LIST/2" + } +] diff --git a/tests/generic_config_updater/files/sonic_yang_after_multi_patch.json b/tests/generic_config_updater/files/sonic_yang_after_multi_patch.json new file mode 100644 index 000000000000..0c9ddd4546a0 --- /dev/null +++ b/tests/generic_config_updater/files/sonic_yang_after_multi_patch.json @@ -0,0 +1,153 @@ +{ + "sonic-vlan:sonic-vlan": { + "sonic-vlan:VLAN_MEMBER": { + "VLAN_MEMBER_LIST": [ + { + "name": "Vlan1000", + "port": "Ethernet0", + "tagging_mode": "untagged" + }, + { + "name": "Vlan1000", + "port": "Ethernet4", + "tagging_mode": "untagged" + }, + { + "name": "Vlan1000", + "port": "Ethernet8", + "tagging_mode": "untagged" + }, + { + "name": "Vlan100", + "port": "Ethernet2", + "tagging_mode": "untagged" + }, + { + "name": "Vlan100", + "port": "Ethernet3", + "tagging_mode": "untagged" + }, + { + "name": "Vlan100", + "port": "Ethernet1", + "tagging_mode": "untagged" + } + ] + }, + "sonic-vlan:VLAN": { + "VLAN_LIST": [ + { + "name": "Vlan1000", + "vlanid": 1000, + "dhcp_servers": [ + "192.0.0.1", + "192.0.0.2", + "192.0.0.3", + "192.0.0.4" + ] + } + ] + } + }, + "sonic-acl:sonic-acl": { + "sonic-acl:ACL_TABLE": { + "ACL_TABLE_LIST": [ + { + "ACL_TABLE_NAME": "NO-NSW-PACL-V4", + "type": "L3", + "policy_desc": "NO-NSW-PACL-V4", + "ports": [ + "Ethernet0", + "Ethernet1", + "Ethernet2", + "Ethernet3" + ] + }, + { + "ACL_TABLE_NAME": "DATAACL", + "policy_desc": "DATAACL", + "ports": [ + "Ethernet4" + ], + "stage": "ingress", + "type": "L3" + }, + { + "ACL_TABLE_NAME": "EVERFLOW", + "policy_desc": "EVERFLOW", + "ports": [ + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRROR" + }, + { + "ACL_TABLE_NAME": "EVERFLOWV6", + "policy_desc": "EVERFLOWV6", + "ports": [ + "Ethernet4", + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRRORV6" + } + ] + } + }, + "sonic-port:sonic-port": { + "sonic-port:PORT": { + "PORT_LIST": [ + { + "name": "Ethernet0", + "alias": "Eth1/1", + "lanes": "65", + "description": "", + "speed": 10000 + }, + { + "name": "Ethernet4", + "admin_status": "up", + "alias": "fortyGigE0/4", + "description": "Servers0:eth0", + "index": 1, + "lanes": "29,30,31,32", + "mtu": 9100, + "pfc_asym": "off", + "speed": 40000 + }, + { + "name": "Ethernet8", + "admin_status": "up", + "alias": "fortyGigE0/8", + "description": "Servers1:eth0", + "index": 2, + "lanes": "33,34,35,36", + "mtu": 9100, + "pfc_asym": "off", + "speed": 40000 + }, + { + "name": "Ethernet3", + "alias": "Eth1/4", + "lanes": "68", + "description": "", + "speed": 10000 + }, + { + "name": "Ethernet1", + "alias": "Eth1/2", + "lanes": "66", + "description": "", + "speed": 10000 + }, + { + "name": "Ethernet2", + "alias": "Eth1/3", + "lanes": "67", + "description": "", + "speed": 10000 + } + ] + } + } +} diff --git a/tests/generic_config_updater/files/sonic_yang_as_json.json b/tests/generic_config_updater/files/sonic_yang_as_json.json new file mode 100644 index 000000000000..37f0fe6ba779 --- /dev/null +++ b/tests/generic_config_updater/files/sonic_yang_as_json.json @@ -0,0 +1,114 @@ +{ + "sonic-vlan:sonic-vlan": { + "sonic-vlan:VLAN_MEMBER": { + "VLAN_MEMBER_LIST": [ + { + "name": "Vlan1000", + "port": "Ethernet0", + "tagging_mode": "untagged" + }, + { + "name": "Vlan1000", + "port": "Ethernet4", + "tagging_mode": "untagged" + }, + { + "name": "Vlan1000", + "port": "Ethernet8", + "tagging_mode": "untagged" + } + ] + }, + "sonic-vlan:VLAN": { + "VLAN_LIST": [ + { + "name": "Vlan1000", + "vlanid": 1000, + "dhcp_servers": [ + "192.0.0.1", + "192.0.0.2", + "192.0.0.3", + "192.0.0.4" + ] + } + ] + } + }, + "sonic-acl:sonic-acl": { + "sonic-acl:ACL_TABLE": { + "ACL_TABLE_LIST": [ + { + "ACL_TABLE_NAME": "NO-NSW-PACL-V4", + "type": "L3", + "policy_desc": "NO-NSW-PACL-V4", + "ports": [ + "Ethernet0" + ] + }, + { + "ACL_TABLE_NAME": "DATAACL", + "policy_desc": "DATAACL", + "ports": [ + "Ethernet4" + ], + "stage": "ingress", + "type": "L3" + }, + { + "ACL_TABLE_NAME": "EVERFLOW", + "policy_desc": "EVERFLOW", + "ports": [ + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRROR" + }, + { + "ACL_TABLE_NAME": "EVERFLOWV6", + "policy_desc": "EVERFLOWV6", + "ports": [ + "Ethernet4", + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRRORV6" + } + ] + } + }, + "sonic-port:sonic-port": { + "sonic-port:PORT": { + "PORT_LIST": [ + { + "name": "Ethernet0", + "alias": "Eth1", + "lanes": "65, 66, 67, 68", + "description": "Ethernet0 100G link", + "speed": 100000 + }, + { + "name": "Ethernet4", + "admin_status": "up", + "alias": "fortyGigE0/4", + "description": "Servers0:eth0", + "index": 1, + "lanes": "29,30,31,32", + "mtu": 9100, + "pfc_asym": "off", + "speed": 40000 + }, + { + "name": "Ethernet8", + "admin_status": "up", + "alias": "fortyGigE0/8", + "description": "Servers1:eth0", + "index": 2, + "lanes": "33,34,35,36", + "mtu": 9100, + "pfc_asym": "off", + "speed": 40000 + } + ] + } + } +} diff --git a/tests/generic_config_updater/files/sonic_yang_as_json_invalid.json b/tests/generic_config_updater/files/sonic_yang_as_json_invalid.json new file mode 100644 index 000000000000..4f67d7e6a6b2 --- /dev/null +++ b/tests/generic_config_updater/files/sonic_yang_as_json_invalid.json @@ -0,0 +1,13 @@ +{ + "sonic-vlan:sonic-vlan": { + "sonic-vlan:VLAN_MEMBER": { + "VLAN_MEMBER_LIST": [ + { + "name": "Vlan1000", + "port": "Ethernet4", + "tagging_mode": "untagged" + } + ] + } + } +} diff --git a/tests/generic_config_updater/files/sonic_yang_as_json_with_unexpected_colons.json b/tests/generic_config_updater/files/sonic_yang_as_json_with_unexpected_colons.json new file mode 100644 index 000000000000..aac97da42b14 --- /dev/null +++ b/tests/generic_config_updater/files/sonic_yang_as_json_with_unexpected_colons.json @@ -0,0 +1,114 @@ +{ + "sonic-vlan:sonic-vlan": { + "sonic-vlan::VLAN_MEMBER": { + "VLAN_MEMBER_LIST": [ + { + "name": "Vlan1000", + "port": "Ethernet0", + "tagging_mode": "untagged" + }, + { + "name": "Vlan1000", + "port": "Ethernet4", + "tagging_mode": "untagged" + }, + { + "name": "Vlan1000", + "port": "Ethernet8", + "tagging_mode": "untagged" + } + ] + }, + "sonic-vlan::VLAN": { + "VLAN_LIST": [ + { + "name": "Vlan1000", + "vlanid": 1000, + "dhcp_servers": [ + "192.0.0.1", + "192.0.0.2", + "192.0.0.3", + "192.0.0.4" + ] + } + ] + } + }, + "sonic-acl:sonic-acl": { + "sonic-vlan::ACL_TABLE": { + "ACL_TABLE_LIST": [ + { + "ACL_TABLE_NAME": "NO-NSW-PACL-V4", + "type": "L3", + "policy_desc": "NO-NSW-PACL-V4", + "ports": [ + "Ethernet0" + ] + }, + { + "ACL_TABLE_NAME": "DATAACL", + "policy_desc": "DATAACL", + "ports": [ + "Ethernet4" + ], + "stage": "ingress", + "type": "L3" + }, + { + "ACL_TABLE_NAME": "EVERFLOW", + "policy_desc": "EVERFLOW", + "ports": [ + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRROR" + }, + { + "ACL_TABLE_NAME": "EVERFLOWV6", + "policy_desc": "EVERFLOWV6", + "ports": [ + "Ethernet4", + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRRORV6" + } + ] + } + }, + "sonic-port:sonic-port": { + "sonic-vlan::PORT": { + "PORT_LIST": [ + { + "name": "Ethernet0", + "alias": "Eth1", + "lanes": "65, 66, 67, 68", + "description": "Ethernet0 100G link", + "speed": 100000 + }, + { + "name": "Ethernet4", + "admin_status": "up", + "alias": "fortyGigE0/4", + "description": "Servers0:eth0", + "index": 1, + "lanes": "29,30,31,32", + "mtu": 9100, + "pfc_asym": "off", + "speed": 40000 + }, + { + "name": "Ethernet8", + "admin_status": "up", + "alias": "fortyGigE0/8", + "description": "Servers1:eth0", + "index": 2, + "lanes": "33,34,35,36", + "mtu": 9100, + "pfc_asym": "off", + "speed": 40000 + } + ] + } + } +} diff --git a/tests/generic_config_updater/files/sonic_yang_as_json_without_colons.json b/tests/generic_config_updater/files/sonic_yang_as_json_without_colons.json new file mode 100644 index 000000000000..ad4ab15f4abb --- /dev/null +++ b/tests/generic_config_updater/files/sonic_yang_as_json_without_colons.json @@ -0,0 +1,114 @@ +{ + "sonic-vlan:sonic-vlan": { + "VLAN_MEMBER": { + "VLAN_MEMBER_LIST": [ + { + "name": "Vlan1000", + "port": "Ethernet0", + "tagging_mode": "untagged" + }, + { + "name": "Vlan1000", + "port": "Ethernet4", + "tagging_mode": "untagged" + }, + { + "name": "Vlan1000", + "port": "Ethernet8", + "tagging_mode": "untagged" + } + ] + }, + "VLAN": { + "VLAN_LIST": [ + { + "name": "Vlan1000", + "vlanid": 1000, + "dhcp_servers": [ + "192.0.0.1", + "192.0.0.2", + "192.0.0.3", + "192.0.0.4" + ] + } + ] + } + }, + "sonic-acl:sonic-acl": { + "ACL_TABLE": { + "ACL_TABLE_LIST": [ + { + "ACL_TABLE_NAME": "NO-NSW-PACL-V4", + "type": "L3", + "policy_desc": "NO-NSW-PACL-V4", + "ports": [ + "Ethernet0" + ] + }, + { + "ACL_TABLE_NAME": "DATAACL", + "policy_desc": "DATAACL", + "ports": [ + "Ethernet4" + ], + "stage": "ingress", + "type": "L3" + }, + { + "ACL_TABLE_NAME": "EVERFLOW", + "policy_desc": "EVERFLOW", + "ports": [ + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRROR" + }, + { + "ACL_TABLE_NAME": "EVERFLOWV6", + "policy_desc": "EVERFLOWV6", + "ports": [ + "Ethernet4", + "Ethernet8" + ], + "stage": "ingress", + "type": "MIRRORV6" + } + ] + } + }, + "sonic-port:sonic-port": { + "PORT": { + "PORT_LIST": [ + { + "name": "Ethernet0", + "alias": "Eth1", + "lanes": "65, 66, 67, 68", + "description": "Ethernet0 100G link", + "speed": 100000 + }, + { + "name": "Ethernet4", + "admin_status": "up", + "alias": "fortyGigE0/4", + "description": "Servers0:eth0", + "index": 1, + "lanes": "29,30,31,32", + "mtu": 9100, + "pfc_asym": "off", + "speed": 40000 + }, + { + "name": "Ethernet8", + "admin_status": "up", + "alias": "fortyGigE0/8", + "description": "Servers1:eth0", + "index": 2, + "lanes": "33,34,35,36", + "mtu": 9100, + "pfc_asym": "off", + "speed": 40000 + } + ] + } + } +} diff --git a/tests/generic_config_updater/generic_updater_test.py b/tests/generic_config_updater/generic_updater_test.py new file mode 100644 index 000000000000..f201280062a5 --- /dev/null +++ b/tests/generic_config_updater/generic_updater_test.py @@ -0,0 +1,766 @@ +import json +import os +import shutil +import unittest +from unittest.mock import MagicMock, Mock, call +from .gutest_helpers import create_side_effect_dict, Files + +import generic_config_updater.generic_updater as gu + +# import sys +# sys.path.insert(0,'../../generic_config_updater') +# import generic_updater as gu + +class TestPatchApplier(unittest.TestCase): + def test_apply__invalid_patch_updating_tables_without_yang_models__failure(self): + # Arrange + patch_applier = self.__create_patch_applier(valid_patch_only_tables_with_yang_models=False) + + # Act and assert + self.assertRaises(ValueError, patch_applier.apply, Files.MULTI_OPERATION_CONFIG_DB_PATCH) + + def test_apply__invalid_config_db__failure(self): + # Arrange + patch_applier = self.__create_patch_applier(valid_config_db=False) + + # Act and assert + self.assertRaises(ValueError, patch_applier.apply, Files.MULTI_OPERATION_CONFIG_DB_PATCH) + + def test_apply__json_not_fully_updated__failure(self): + # Arrange + patch_applier = self.__create_patch_applier(verified_same_config=False) + + # Act and assert + self.assertRaises(gu.GenericConfigUpdaterError, patch_applier.apply, Files.MULTI_OPERATION_CONFIG_DB_PATCH) + + def test_apply__no_errors__update_successful(self): + # Arrange + changes = [Mock(), Mock()] + patch_applier = self.__create_patch_applier(changes) + + # Act + patch_applier.apply(Files.MULTI_OPERATION_CONFIG_DB_PATCH) + + # Assert + patch_applier.patch_wrapper.validate_config_db_patch_has_yang_models.assert_has_calls( + [call(Files.MULTI_OPERATION_CONFIG_DB_PATCH)]) + patch_applier.config_wrapper.get_config_db_as_json.assert_has_calls([call(), call()]) + patch_applier.patch_wrapper.simulate_patch.assert_has_calls( + [call(Files.MULTI_OPERATION_CONFIG_DB_PATCH, Files.CONFIG_DB_AS_JSON)]) + patch_applier.config_wrapper.validate_config_db_config.assert_has_calls( + [call(Files.CONFIG_DB_AFTER_MULTI_PATCH)]) + patch_applier.patchsorter.sort.assert_has_calls([call(Files.MULTI_OPERATION_CONFIG_DB_PATCH)]) + patch_applier.changeapplier.apply.assert_has_calls([call(changes[0]), call(changes[1])]) + patch_applier.patch_wrapper.verify_same_json.assert_has_calls( + [call(Files.CONFIG_DB_AFTER_MULTI_PATCH, Files.CONFIG_DB_AFTER_MULTI_PATCH)]) + + def __create_patch_applier(self, + changes=None, + valid_patch_only_tables_with_yang_models=True, + valid_config_db=True, + verified_same_config=True): + config_wrapper = Mock() + config_wrapper.get_config_db_as_json.side_effect = \ + [Files.CONFIG_DB_AS_JSON, Files.CONFIG_DB_AFTER_MULTI_PATCH] + config_wrapper.validate_config_db_config.side_effect = \ + create_side_effect_dict({(str(Files.CONFIG_DB_AFTER_MULTI_PATCH),): valid_config_db}) + + patch_wrapper = Mock() + patch_wrapper.validate_config_db_patch_has_yang_models.side_effect = \ + create_side_effect_dict( + {(str(Files.MULTI_OPERATION_CONFIG_DB_PATCH),): valid_patch_only_tables_with_yang_models}) + patch_wrapper.simulate_patch.side_effect = \ + create_side_effect_dict( + {(str(Files.MULTI_OPERATION_CONFIG_DB_PATCH), str(Files.CONFIG_DB_AS_JSON)): + Files.CONFIG_DB_AFTER_MULTI_PATCH}) + patch_wrapper.verify_same_json.side_effect = \ + create_side_effect_dict( + {(str(Files.CONFIG_DB_AFTER_MULTI_PATCH), str(Files.CONFIG_DB_AFTER_MULTI_PATCH)): + verified_same_config}) + + changes = [Mock(), Mock()] if not changes else changes + patchsorter = Mock() + patchsorter.sort.side_effect = \ + create_side_effect_dict({(str(Files.MULTI_OPERATION_CONFIG_DB_PATCH),): changes}) + + changeapplier = Mock() + changeapplier.apply.side_effect = create_side_effect_dict({(str(changes[0]),): 0, (str(changes[1]),): 0}) + + return gu.PatchApplier(patchsorter, changeapplier, config_wrapper, patch_wrapper) + +class TestConfigReplacer(unittest.TestCase): + def test_replace__invalid_config_db__failure(self): + # Arrange + config_replacer = self.__create_config_replacer(valid_config_db=False) + + # Act and assert + self.assertRaises(ValueError, config_replacer.replace, Files.CONFIG_DB_AFTER_MULTI_PATCH) + + def test_replace__json_not_fully_updated__failure(self): + # Arrange + config_replacer = self.__create_config_replacer(verified_same_config=False) + + # Act and assert + self.assertRaises(gu.GenericConfigUpdaterError, config_replacer.replace, Files.CONFIG_DB_AFTER_MULTI_PATCH) + + def test_replace__no_errors__update_successful(self): + # Arrange + config_replacer = self.__create_config_replacer() + + # Act + config_replacer.replace(Files.CONFIG_DB_AFTER_MULTI_PATCH) + + # Assert + config_replacer.config_wrapper.validate_config_db_config.assert_has_calls( + [call(Files.CONFIG_DB_AFTER_MULTI_PATCH)]) + config_replacer.config_wrapper.get_config_db_as_json.assert_has_calls([call(), call()]) + config_replacer.patch_wrapper.generate_patch.assert_has_calls( + [call(Files.CONFIG_DB_AS_JSON, Files.CONFIG_DB_AFTER_MULTI_PATCH)]) + config_replacer.patch_applier.apply.assert_has_calls([call(Files.MULTI_OPERATION_CONFIG_DB_PATCH)]) + config_replacer.patch_wrapper.verify_same_json.assert_has_calls( + [call(Files.CONFIG_DB_AFTER_MULTI_PATCH, Files.CONFIG_DB_AFTER_MULTI_PATCH)]) + + def __create_config_replacer(self, changes=None, valid_config_db=True, verified_same_config=True): + config_wrapper = Mock() + config_wrapper.validate_config_db_config.side_effect = \ + create_side_effect_dict({(str(Files.CONFIG_DB_AFTER_MULTI_PATCH),): valid_config_db}) + config_wrapper.get_config_db_as_json.side_effect = \ + [Files.CONFIG_DB_AS_JSON, Files.CONFIG_DB_AFTER_MULTI_PATCH] + + patch_wrapper = Mock() + patch_wrapper.generate_patch.side_effect = \ + create_side_effect_dict( + {(str(Files.CONFIG_DB_AS_JSON), str(Files.CONFIG_DB_AFTER_MULTI_PATCH)): + Files.MULTI_OPERATION_CONFIG_DB_PATCH}) + patch_wrapper.verify_same_json.side_effect = \ + create_side_effect_dict( + {(str(Files.CONFIG_DB_AFTER_MULTI_PATCH), str(Files.CONFIG_DB_AFTER_MULTI_PATCH)): \ + verified_same_config}) + + changes = [Mock(), Mock()] if not changes else changes + patchsorter = Mock() + patchsorter.sort.side_effect = create_side_effect_dict({(str(Files.MULTI_OPERATION_CONFIG_DB_PATCH),): \ + changes}) + + patch_applier = Mock() + patch_applier.apply.side_effect = create_side_effect_dict({(str(Files.MULTI_OPERATION_CONFIG_DB_PATCH),): 0}) + + return gu.ConfigReplacer(patch_applier, config_wrapper, patch_wrapper) + +class TestFileSystemConfigRollbacker(unittest.TestCase): + def setUp(self): + self.checkpoints_dir = os.path.join(os.getcwd(),"checkpoints") + self.checkpoint_ext = ".cp.json" + self.any_checkpoint_name = "anycheckpoint" + self.any_other_checkpoint_name = "anyothercheckpoint" + self.any_config = {} + self.clean_up() + + def tearDown(self): + self.clean_up() + + def test_rollback__checkpoint_does_not_exist__failure(self): + # Arrange + rollbacker = self.create_rollbacker() + + # Act and assert + self.assertRaises(ValueError, rollbacker.rollback, "NonExistingCheckpoint") + + def test_rollback__no_errors__success(self): + # Arrange + self.create_checkpoints_dir() + self.add_checkpoint(self.any_checkpoint_name, self.any_config) + rollbacker = self.create_rollbacker() + + # Act + rollbacker.rollback(self.any_checkpoint_name) + + # Assert + rollbacker.config_replacer.replace.assert_has_calls([call(self.any_config)]) + + def test_checkpoint__checkpoints_dir_does_not_exist__checkpoint_created(self): + # Arrange + rollbacker = self.create_rollbacker() + self.assertFalse(os.path.isdir(self.checkpoints_dir)) + + # Act + rollbacker.checkpoint(self.any_checkpoint_name) + + # Assert + self.assertTrue(os.path.isdir(self.checkpoints_dir)) + self.assertEqual(self.any_config, self.get_checkpoint(self.any_checkpoint_name)) + + def test_checkpoint__config_not_valid__failure(self): + # Arrange + rollbacker = self.create_rollbacker(valid_config=False) + + # Act and assert + self.assertRaises(ValueError, rollbacker.checkpoint, self.any_checkpoint_name) + + def test_checkpoint__checkpoints_dir_exists__checkpoint_created(self): + # Arrange + self.create_checkpoints_dir() + rollbacker = self.create_rollbacker() + + # Act + rollbacker.checkpoint(self.any_checkpoint_name) + + # Assert + self.assertEqual(self.any_config, self.get_checkpoint(self.any_checkpoint_name)) + + def test_list_checkpoints__checkpoints_dir_does_not_exist__empty_list(self): + # Arrange + rollbacker = self.create_rollbacker() + self.assertFalse(os.path.isdir(self.checkpoints_dir)) + expected = [] + + # Act + actual = rollbacker.list_checkpoints() + + # Assert + # 'assertCountEqual' does check same count, same elements ignoring order + self.assertCountEqual(expected, actual) + + def test_list_checkpoints__checkpoints_dir_exist_but_no_files__empty_list(self): + # Arrange + self.create_checkpoints_dir() + rollbacker = self.create_rollbacker() + expected = [] + + # Act + actual = rollbacker.list_checkpoints() + + # Assert + # 'assertCountEqual' does check same count, same elements ignoring order + self.assertCountEqual(expected, actual) + + def test_list_checkpoints__checkpoints_dir_has_multiple_files__multiple_files(self): + # Arrange + self.create_checkpoints_dir() + self.add_checkpoint(self.any_checkpoint_name, self.any_config) + self.add_checkpoint(self.any_other_checkpoint_name, self.any_config) + rollbacker = self.create_rollbacker() + expected = [self.any_checkpoint_name, self.any_other_checkpoint_name] + + # Act + actual = rollbacker.list_checkpoints() + + # Assert + # 'assertCountEqual' does check same count, same elements ignoring order + self.assertCountEqual(expected, actual) + + def test_list_checkpoints__checkpoints_names_have_special_characters__multiple_files(self): + # Arrange + self.create_checkpoints_dir() + self.add_checkpoint("check.point1", self.any_config) + self.add_checkpoint(".checkpoint2", self.any_config) + self.add_checkpoint("checkpoint3.", self.any_config) + rollbacker = self.create_rollbacker() + expected = ["check.point1", ".checkpoint2", "checkpoint3."] + + # Act + actual = rollbacker.list_checkpoints() + + # Assert + # 'assertCountEqual' does check same count, same elements ignoring order + self.assertCountEqual(expected, actual) + + def test_delete_checkpoint__checkpoint_does_not_exist__failure(self): + # Arrange + rollbacker = self.create_rollbacker() + + # Act and assert + self.assertRaises(ValueError, rollbacker.delete_checkpoint, self.any_checkpoint_name) + + def test_delete_checkpoint__checkpoint_exist__success(self): + # Arrange + self.create_checkpoints_dir() + self.add_checkpoint(self.any_checkpoint_name, self.any_config) + rollbacker = self.create_rollbacker() + + # Act + rollbacker.delete_checkpoint(self.any_checkpoint_name) + + # Assert + self.assertFalse(self.check_checkpoint_exists(self.any_checkpoint_name)) + + def test_multiple_operations(self): + rollbacker = self.create_rollbacker() + + # 'assertCountEqual' does check same count, same elements ignoring order + self.assertCountEqual([], rollbacker.list_checkpoints()) + + rollbacker.checkpoint(self.any_checkpoint_name) + self.assertCountEqual([self.any_checkpoint_name], rollbacker.list_checkpoints()) + self.assertEqual(self.any_config, self.get_checkpoint(self.any_checkpoint_name)) + + rollbacker.rollback(self.any_checkpoint_name) + rollbacker.config_replacer.replace.assert_has_calls([call(self.any_config)]) + + rollbacker.checkpoint(self.any_other_checkpoint_name) + self.assertCountEqual([self.any_checkpoint_name, self.any_other_checkpoint_name], rollbacker.list_checkpoints()) + self.assertEqual(self.any_config, self.get_checkpoint(self.any_other_checkpoint_name)) + + rollbacker.delete_checkpoint(self.any_checkpoint_name) + self.assertCountEqual([self.any_other_checkpoint_name], rollbacker.list_checkpoints()) + + rollbacker.delete_checkpoint(self.any_other_checkpoint_name) + self.assertCountEqual([], rollbacker.list_checkpoints()) + + def clean_up(self): + if os.path.isdir(self.checkpoints_dir): + shutil.rmtree(self.checkpoints_dir) + + def create_checkpoints_dir(self): + os.makedirs(self.checkpoints_dir) + + def add_checkpoint(self, name, json_content): + path=os.path.join(self.checkpoints_dir, f"{name}{self.checkpoint_ext}") + with open(path, "w") as fh: + fh.write(json.dumps(json_content)) + + def get_checkpoint(self, name): + path=os.path.join(self.checkpoints_dir, f"{name}{self.checkpoint_ext}") + with open(path) as fh: + text = fh.read() + return json.loads(text) + + def check_checkpoint_exists(self, name): + path=os.path.join(self.checkpoints_dir, f"{name}{self.checkpoint_ext}") + return os.path.isfile(path) + + def create_rollbacker(self, valid_config=True): + replacer = Mock() + replacer.replace.side_effect = create_side_effect_dict({(str(self.any_config),): 0}) + + config_wrapper = Mock() + config_wrapper.get_config_db_as_json.return_value = self.any_config + config_wrapper.validate_config_db_config.return_value = valid_config + + return gu.FileSystemConfigRollbacker(checkpoints_dir=self.checkpoints_dir, + config_replacer=replacer, + config_wrapper=config_wrapper) + +class TestGenericUpdateFactory(unittest.TestCase): + def setUp(self): + self.any_verbose=True + self.any_dry_run=True + + def test_create_patch_applier__invalid_config_format__failure(self): + # Arrange + factory = gu.GenericUpdateFactory() + + # Act and assert + self.assertRaises( + ValueError, factory.create_patch_applier, "INVALID_FORMAT", self.any_verbose, self.any_dry_run) + + def test_create_patch_applier__different_options(self): + # Arrange + options = [ + {"verbose": {True: None, False: None}}, + {"dry_run": {True: None, False: gu.ConfigLockDecorator}}, + { + "config_format": { + gu.ConfigFormat.SONICYANG: gu.SonicYangDecorator, + gu.ConfigFormat.CONFIGDB: None, + } + }, + ] + + # Act and assert + self.recursively_test_create_func(options, 0, {}, [], self.validate_create_patch_applier) + + def test_create_config_replacer__invalid_config_format__failure(self): + # Arrange + factory = gu.GenericUpdateFactory() + + # Act and assert + self.assertRaises( + ValueError, factory.create_config_replacer, "INVALID_FORMAT", self.any_verbose, self.any_dry_run) + + def test_create_config_replacer__different_options(self): + # Arrange + options = [ + {"verbose": {True: None, False: None}}, + {"dry_run": {True: None, False: gu.ConfigLockDecorator}}, + { + "config_format": { + gu.ConfigFormat.SONICYANG: gu.SonicYangDecorator, + gu.ConfigFormat.CONFIGDB: None, + } + }, + ] + + # Act and assert + self.recursively_test_create_func(options, 0, {}, [], self.validate_create_config_replacer) + + def test_create_config_rollbacker__different_options(self): + # Arrange + options = [ + {"verbose": {True: None, False: None}}, + {"dry_run": {True: None, False: gu.ConfigLockDecorator}} + ] + + # Act and assert + self.recursively_test_create_func(options, 0, {}, [], self.validate_create_config_rollbacker) + + def recursively_test_create_func(self, options, cur_option, params, expected_decorators, create_func): + if cur_option == len(options): + create_func(params, expected_decorators) + return + + param = list(options[cur_option].keys())[0] + for key in options[cur_option][param]: + params[param] = key + decorator = options[cur_option][param][key] + if decorator != None: + expected_decorators.append(decorator) + self.recursively_test_create_func(options, cur_option+1, params, expected_decorators, create_func) + if decorator != None: + expected_decorators.pop() + + def validate_create_patch_applier(self, params, expected_decorators): + factory = gu.GenericUpdateFactory() + patch_applier = factory.create_patch_applier(params["config_format"], params["verbose"], params["dry_run"]) + for decorator_type in expected_decorators: + self.assertIsInstance(patch_applier, decorator_type) + + patch_applier = patch_applier.decorated_patch_applier + + self.assertIsInstance(patch_applier, gu.PatchApplier) + if params["dry_run"]: + self.assertIsInstance(patch_applier.config_wrapper, gu.DryRunConfigWrapper) + else: + self.assertIsInstance(patch_applier.config_wrapper, gu.ConfigWrapper) + + def validate_create_config_replacer(self, params, expected_decorators): + factory = gu.GenericUpdateFactory() + config_replacer = factory.create_config_replacer(params["config_format"], params["verbose"], params["dry_run"]) + for decorator_type in expected_decorators: + self.assertIsInstance(config_replacer, decorator_type) + + config_replacer = config_replacer.decorated_config_replacer + + self.assertIsInstance(config_replacer, gu.ConfigReplacer) + if params["dry_run"]: + self.assertIsInstance(config_replacer.config_wrapper, gu.DryRunConfigWrapper) + self.assertIsInstance(config_replacer.patch_applier.config_wrapper, gu.DryRunConfigWrapper) + else: + self.assertIsInstance(config_replacer.config_wrapper, gu.ConfigWrapper) + self.assertIsInstance(config_replacer.patch_applier.config_wrapper, gu.ConfigWrapper) + + def validate_create_config_rollbacker(self, params, expected_decorators): + factory = gu.GenericUpdateFactory() + config_rollbacker = factory.create_config_rollbacker(params["verbose"], params["dry_run"]) + for decorator_type in expected_decorators: + self.assertIsInstance(config_rollbacker, decorator_type) + + config_rollbacker = config_rollbacker.decorated_config_rollbacker + + self.assertIsInstance(config_rollbacker, gu.FileSystemConfigRollbacker) + if params["dry_run"]: + self.assertIsInstance(config_rollbacker.config_wrapper, gu.DryRunConfigWrapper) + self.assertIsInstance(config_rollbacker.config_replacer.config_wrapper, gu.DryRunConfigWrapper) + self.assertIsInstance( + config_rollbacker.config_replacer.patch_applier.config_wrapper, gu.DryRunConfigWrapper) + else: + self.assertIsInstance(config_rollbacker.config_wrapper, gu.ConfigWrapper) + self.assertIsInstance(config_rollbacker.config_replacer.config_wrapper, gu.ConfigWrapper) + self.assertIsInstance( + config_rollbacker.config_replacer.patch_applier.config_wrapper, gu.ConfigWrapper) + +class TestGenericUpdater(unittest.TestCase): + def setUp(self): + self.any_checkpoint_name = "anycheckpoint" + self.any_other_checkpoint_name = "anyothercheckpoint" + self.any_checkpoints_list = [self.any_checkpoint_name, self.any_other_checkpoint_name] + self.any_config_format = gu.ConfigFormat.SONICYANG + self.any_verbose = True + self.any_dry_run = True + + def test_apply_patch__creates_applier_and_apply(self): + # Arrange + patch_applier = Mock() + patch_applier.apply.side_effect = create_side_effect_dict({(str(Files.SINGLE_OPERATION_SONIC_YANG_PATCH),): 0}) + + factory = Mock() + factory.create_patch_applier.side_effect = \ + create_side_effect_dict( + {(str(self.any_config_format), str(self.any_verbose), str(self.any_dry_run),): patch_applier}) + + generic_updater = gu.GenericUpdater(factory) + + # Act + generic_updater.apply_patch( + Files.SINGLE_OPERATION_SONIC_YANG_PATCH, self.any_config_format, self.any_verbose, self.any_dry_run) + + # Assert + patch_applier.apply.assert_has_calls([call(Files.SINGLE_OPERATION_SONIC_YANG_PATCH)]) + + def test_replace__creates_replacer_and_replace(self): + # Arrange + config_replacer = Mock() + config_replacer.replace.side_effect = create_side_effect_dict({(str(Files.SONIC_YANG_AS_JSON),): 0}) + + factory = Mock() + factory.create_config_replacer.side_effect = \ + create_side_effect_dict( + {(str(self.any_config_format), str(self.any_verbose), str(self.any_dry_run),): config_replacer}) + + generic_updater = gu.GenericUpdater(factory) + + # Act + generic_updater.replace(Files.SONIC_YANG_AS_JSON, self.any_config_format, self.any_verbose, self.any_dry_run) + + # Assert + config_replacer.replace.assert_has_calls([call(Files.SONIC_YANG_AS_JSON)]) + + def test_rollback__creates_rollbacker_and_rollback(self): + # Arrange + config_rollbacker = Mock() + config_rollbacker.rollback.side_effect = create_side_effect_dict({(self.any_checkpoint_name,): 0}) + + factory = Mock() + factory.create_config_rollbacker.side_effect = \ + create_side_effect_dict({(str(self.any_verbose), str(self.any_dry_run),): config_rollbacker}) + + generic_updater = gu.GenericUpdater(factory) + + # Act + generic_updater.rollback(self.any_checkpoint_name, self.any_verbose, self.any_dry_run) + + # Assert + config_rollbacker.rollback.assert_has_calls([call(self.any_checkpoint_name)]) + + def test_checkpoint__creates_rollbacker_and_checkpoint(self): + # Arrange + config_rollbacker = Mock() + config_rollbacker.checkpoint.side_effect = create_side_effect_dict({(self.any_checkpoint_name,): 0}) + + factory = Mock() + factory.create_config_rollbacker.side_effect = \ + create_side_effect_dict({(str(self.any_verbose),): config_rollbacker}) + + generic_updater = gu.GenericUpdater(factory) + + # Act + generic_updater.checkpoint(self.any_checkpoint_name, self.any_verbose) + + # Assert + config_rollbacker.checkpoint.assert_has_calls([call(self.any_checkpoint_name)]) + + def test_delete_checkpoint__creates_rollbacker_and_deletes_checkpoint(self): + # Arrange + config_rollbacker = Mock() + config_rollbacker.delete_checkpoint.side_effect = create_side_effect_dict({(self.any_checkpoint_name,): 0}) + + factory = Mock() + factory.create_config_rollbacker.side_effect = \ + create_side_effect_dict({(str(self.any_verbose),): config_rollbacker}) + + generic_updater = gu.GenericUpdater(factory) + + # Act + generic_updater.delete_checkpoint(self.any_checkpoint_name, self.any_verbose) + + # Assert + config_rollbacker.delete_checkpoint.assert_has_calls([call(self.any_checkpoint_name)]) + + def test_list_checkpoints__creates_rollbacker_and_list_checkpoints(self): + # Arrange + config_rollbacker = Mock() + config_rollbacker.list_checkpoints.return_value = self.any_checkpoints_list + + factory = Mock() + factory.create_config_rollbacker.side_effect = \ + create_side_effect_dict({(str(self.any_verbose),): config_rollbacker}) + + generic_updater = gu.GenericUpdater(factory) + + expected = self.any_checkpoints_list + + # Act + actual = generic_updater.list_checkpoints(self.any_verbose) + + # Assert + self.assertCountEqual(expected, actual) + +class TestDecorator(unittest.TestCase): + def setUp(self): + self.decorated_patch_applier = Mock() + self.decorated_config_replacer = Mock() + self.decorated_config_rollbacker = Mock() + + self.any_checkpoint_name = "anycheckpoint" + self.any_other_checkpoint_name = "anyothercheckpoint" + self.any_checkpoints_list = [self.any_checkpoint_name, self.any_other_checkpoint_name] + self.decorated_config_rollbacker.list_checkpoints.return_value = self.any_checkpoints_list + + self.decorator = gu.Decorator( + self.decorated_patch_applier, self.decorated_config_replacer, self.decorated_config_rollbacker) + + def test_apply__calls_decorated_applier(self): + # Act + self.decorator.apply(Files.SINGLE_OPERATION_SONIC_YANG_PATCH) + + # Assert + self.decorated_patch_applier.apply.assert_has_calls([call(Files.SINGLE_OPERATION_SONIC_YANG_PATCH)]) + + def test_replace__calls_decorated_replacer(self): + # Act + self.decorator.replace(Files.SONIC_YANG_AS_JSON) + + # Assert + self.decorated_config_replacer.replace.assert_has_calls([call(Files.SONIC_YANG_AS_JSON)]) + + def test_rollback__calls_decorated_rollbacker(self): + # Act + self.decorator.rollback(self.any_checkpoint_name) + + # Assert + self.decorated_config_rollbacker.rollback.assert_has_calls([call(self.any_checkpoint_name)]) + + def test_checkpoint__calls_decorated_rollbacker(self): + # Act + self.decorator.checkpoint(self.any_checkpoint_name) + + # Assert + self.decorated_config_rollbacker.checkpoint.assert_has_calls([call(self.any_checkpoint_name)]) + + def test_delete_checkpoint__calls_decorated_rollbacker(self): + # Act + self.decorator.delete_checkpoint(self.any_checkpoint_name) + + # Assert + self.decorated_config_rollbacker.delete_checkpoint.assert_has_calls([call(self.any_checkpoint_name)]) + + def test_list_checkpoints__calls_decorated_rollbacker(self): + # Arrange + expected = self.any_checkpoints_list + + # Act + actual = self.decorator.list_checkpoints() + + # Assert + self.decorated_config_rollbacker.list_checkpoints.assert_called_once() + self.assertListEqual(expected, actual) + +class TestSonicYangDecorator(unittest.TestCase): + def test_apply__converts_to_config_db_and_calls_decorated_class(self): + # Arrange + sonic_yang_decorator = self.__create_sonic_yang_decorator() + + # Act + sonic_yang_decorator.apply(Files.SINGLE_OPERATION_SONIC_YANG_PATCH) + + # Assert + sonic_yang_decorator.patch_wrapper.convert_sonic_yang_patch_to_config_db_patch.assert_has_calls( + [call(Files.SINGLE_OPERATION_SONIC_YANG_PATCH)]) + sonic_yang_decorator.decorated_patch_applier.apply.assert_has_calls( + [call(Files.SINGLE_OPERATION_CONFIG_DB_PATCH)]) + + def test_replace__converts_to_config_db_and_calls_decorated_class(self): + # Arrange + sonic_yang_decorator = self.__create_sonic_yang_decorator() + + # Act + sonic_yang_decorator.replace(Files.SONIC_YANG_AS_JSON) + + # Assert + sonic_yang_decorator.config_wrapper.convert_sonic_yang_to_config_db.assert_has_calls( + [call(Files.SONIC_YANG_AS_JSON)]) + sonic_yang_decorator.decorated_config_replacer.replace.assert_has_calls([call(Files.CONFIG_DB_AS_JSON)]) + + def __create_sonic_yang_decorator(self): + patch_applier = Mock() + patch_applier.apply.side_effect = create_side_effect_dict({(str(Files.SINGLE_OPERATION_CONFIG_DB_PATCH),): 0}) + + patch_wrapper = Mock() + patch_wrapper.convert_sonic_yang_patch_to_config_db_patch.side_effect = \ + create_side_effect_dict({(str(Files.SINGLE_OPERATION_SONIC_YANG_PATCH),): \ + Files.SINGLE_OPERATION_CONFIG_DB_PATCH}) + + config_replacer = Mock() + config_replacer.replace.side_effect = create_side_effect_dict({(str(Files.CONFIG_DB_AS_JSON),): 0}) + + config_wrapper = Mock() + config_wrapper.convert_sonic_yang_to_config_db.side_effect = \ + create_side_effect_dict({(str(Files.SONIC_YANG_AS_JSON),): Files.CONFIG_DB_AS_JSON}) + + return gu.SonicYangDecorator(decorated_patch_applier=patch_applier, + decorated_config_replacer=config_replacer, + patch_wrapper=patch_wrapper, + config_wrapper=config_wrapper) + +class TestConfigLockDecorator(unittest.TestCase): + def setUp(self): + self.any_checkpoint_name = "anycheckpoint" + + def test_apply__lock_config(self): + # Arrange + config_lock_decorator = self.__create_config_lock_decorator() + + # Act + config_lock_decorator.apply(Files.SINGLE_OPERATION_SONIC_YANG_PATCH) + + # Assert + config_lock_decorator.config_lock.acquire_lock.assert_called_once() + config_lock_decorator.decorated_patch_applier.apply.assert_has_calls( + [call(Files.SINGLE_OPERATION_SONIC_YANG_PATCH)]) + config_lock_decorator.config_lock.release_lock.assert_called_once() + + def test_replace__lock_config(self): + # Arrange + config_lock_decorator = self.__create_config_lock_decorator() + + # Act + config_lock_decorator.replace(Files.SONIC_YANG_AS_JSON) + + # Assert + config_lock_decorator.config_lock.acquire_lock.assert_called_once() + config_lock_decorator.decorated_config_replacer.replace.assert_has_calls([call(Files.SONIC_YANG_AS_JSON)]) + config_lock_decorator.config_lock.release_lock.assert_called_once() + + def test_rollback__lock_config(self): + # Arrange + config_lock_decorator = self.__create_config_lock_decorator() + + # Act + config_lock_decorator.rollback(self.any_checkpoint_name) + + # Assert + config_lock_decorator.config_lock.acquire_lock.assert_called_once() + config_lock_decorator.decorated_config_rollbacker.rollback.assert_has_calls([call(self.any_checkpoint_name)]) + config_lock_decorator.config_lock.release_lock.assert_called_once() + + def test_checkpoint__lock_config(self): + # Arrange + config_lock_decorator = self.__create_config_lock_decorator() + + # Act + config_lock_decorator.checkpoint(self.any_checkpoint_name) + + # Assert + config_lock_decorator.config_lock.acquire_lock.assert_called_once() + config_lock_decorator.decorated_config_rollbacker.checkpoint.assert_has_calls([call(self.any_checkpoint_name)]) + config_lock_decorator.config_lock.release_lock.assert_called_once() + + def __create_config_lock_decorator(self): + config_lock = Mock() + + patch_applier = Mock() + patch_applier.apply.side_effect = create_side_effect_dict({(str(Files.SINGLE_OPERATION_SONIC_YANG_PATCH),): 0}) + + config_replacer = Mock() + config_replacer.replace.side_effect = create_side_effect_dict({(str(Files.SONIC_YANG_AS_JSON),): 0}) + + config_rollbacker = Mock() + config_rollbacker.rollback.side_effect = create_side_effect_dict({(self.any_checkpoint_name,): 0}) + config_rollbacker.checkpoint.side_effect = create_side_effect_dict({(self.any_checkpoint_name,): 0}) + + config_rollbacker.delete_checkpoint.side_effect = create_side_effect_dict({(self.any_checkpoint_name,): 0}) + + return gu.ConfigLockDecorator(config_lock=config_lock, + decorated_patch_applier=patch_applier, + decorated_config_replacer=config_replacer, + decorated_config_rollbacker=config_rollbacker) diff --git a/tests/generic_config_updater/gu_common_test.py b/tests/generic_config_updater/gu_common_test.py new file mode 100644 index 000000000000..f18ad4579902 --- /dev/null +++ b/tests/generic_config_updater/gu_common_test.py @@ -0,0 +1,335 @@ +import json +import jsonpatch +import unittest +from unittest.mock import MagicMock, Mock +from .gutest_helpers import create_side_effect_dict, Files + +import generic_config_updater.gu_common as gu_common + +# import sys +# sys.path.insert(0,'../../generic_config_updater') +# import gu_common + +class TestConfigWrapper(unittest.TestCase): + def setUp(self): + self.config_wrapper_mock = gu_common.ConfigWrapper() + self.config_wrapper_mock.get_config_db_as_json=MagicMock(return_value=Files.CONFIG_DB_AS_JSON) + + def test_ctor__default_values_set(self): + config_wrapper = gu_common.ConfigWrapper() + + self.assertEqual("/usr/local/yang-models", gu_common.YANG_DIR) + + def test_get_sonic_yang_as_json__returns_sonic_yang_as_json(self): + # Arrange + config_wrapper = self.config_wrapper_mock + expected = Files.SONIC_YANG_AS_JSON + + # Act + actual = config_wrapper.get_sonic_yang_as_json() + + # Assert + self.assertDictEqual(expected, actual) + + def test_convert_config_db_to_sonic_yang__empty_config_db__returns_empty_sonic_yang(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = {} + + # Act + actual = config_wrapper.convert_config_db_to_sonic_yang({}) + + # Assert + self.assertDictEqual(expected, actual) + + def test_convert_config_db_to_sonic_yang__non_empty_config_db__returns_sonic_yang_as_json(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = Files.SONIC_YANG_AS_JSON + + # Act + actual = config_wrapper.convert_config_db_to_sonic_yang(Files.CONFIG_DB_AS_JSON) + + # Assert + self.assertDictEqual(expected, actual) + + def test_convert_sonic_yang_to_config_db__empty_sonic_yang__returns_empty_config_db(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = {} + + # Act + actual = config_wrapper.convert_sonic_yang_to_config_db({}) + + # Assert + self.assertDictEqual(expected, actual) + + def test_convert_sonic_yang_to_config_db__non_empty_sonic_yang__returns_config_db_as_json(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = Files.CROPPED_CONFIG_DB_AS_JSON + + # Act + actual = config_wrapper.convert_sonic_yang_to_config_db(Files.SONIC_YANG_AS_JSON) + + # Assert + self.assertDictEqual(expected, actual) + + def test_convert_sonic_yang_to_config_db__table_name_without_colons__returns_config_db_as_json(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = Files.CROPPED_CONFIG_DB_AS_JSON + + # Act + actual = config_wrapper.convert_sonic_yang_to_config_db(Files.SONIC_YANG_AS_JSON_WITHOUT_COLONS) + + # Assert + self.assertDictEqual(expected, actual) + + def test_convert_sonic_yang_to_config_db__table_name_with_unexpected_colons__returns_config_db_as_json(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = Files.CROPPED_CONFIG_DB_AS_JSON + + # Act and assert + self.assertRaises(ValueError, + config_wrapper.convert_sonic_yang_to_config_db, + Files.SONIC_YANG_AS_JSON_WITH_UNEXPECTED_COLONS) + + def test_validate_sonic_yang_config__valid_config__returns_true(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = True + + # Act + actual = config_wrapper.validate_sonic_yang_config(Files.SONIC_YANG_AS_JSON) + + # Assert + self.assertEqual(expected, actual) + + def test_validate_sonic_yang_config__invvalid_config__returns_false(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = False + + # Act + actual = config_wrapper.validate_sonic_yang_config(Files.SONIC_YANG_AS_JSON_INVALID) + + # Assert + self.assertEqual(expected, actual) + + def test_validate_config_db_config__valid_config__returns_true(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = True + + # Act + actual = config_wrapper.validate_config_db_config(Files.CONFIG_DB_AS_JSON) + + # Assert + self.assertEqual(expected, actual) + + def test_validate_config_db_config__invalid_config__returns_false(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = False + + # Act + actual = config_wrapper.validate_config_db_config(Files.CONFIG_DB_AS_JSON_INVALID) + + # Assert + self.assertEqual(expected, actual) + + def test_crop_tables_without_yang__returns_cropped_config_db_as_json(self): + # Arrange + config_wrapper = gu_common.ConfigWrapper() + expected = Files.CROPPED_CONFIG_DB_AS_JSON + + # Act + actual = config_wrapper.crop_tables_without_yang(Files.CONFIG_DB_AS_JSON) + + # Assert + self.assertDictEqual(expected, actual) + +class TestPatchWrapper(unittest.TestCase): + def setUp(self): + self.config_wrapper_mock = gu_common.ConfigWrapper() + self.config_wrapper_mock.get_config_db_as_json=MagicMock(return_value=Files.CONFIG_DB_AS_JSON) + + def test_validate_config_db_patch_has_yang_models__table_without_yang_model__returns_false(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + patch = [ { 'op': 'remove', 'path': '/TABLE_WITHOUT_YANG' } ] + expected = False + + # Act + actual = patch_wrapper.validate_config_db_patch_has_yang_models(patch) + + # Assert + self.assertEqual(expected, actual) + + def test_validate_config_db_patch_has_yang_models__table_with_yang_model__returns_true(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + patch = [ { 'op': 'remove', 'path': '/ACL_TABLE' } ] + expected = True + + # Act + actual = patch_wrapper.validate_config_db_patch_has_yang_models(patch) + + # Assert + self.assertEqual(expected, actual) + + def test_convert_config_db_patch_to_sonic_yang_patch__invalid_config_db_patch__failure(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + patch = [ { 'op': 'remove', 'path': '/TABLE_WITHOUT_YANG' } ] + + # Act and Assert + self.assertRaises(ValueError, patch_wrapper.convert_config_db_patch_to_sonic_yang_patch, patch) + + def test_same_patch__no_diff__returns_true(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + + # Act and Assert + self.assertTrue(patch_wrapper.verify_same_json(Files.CONFIG_DB_AS_JSON, Files.CONFIG_DB_AS_JSON)) + + def test_same_patch__diff__returns_false(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + + # Act and Assert + self.assertFalse(patch_wrapper.verify_same_json(Files.CONFIG_DB_AS_JSON, Files.CROPPED_CONFIG_DB_AS_JSON)) + + def test_generate_patch__no_diff__empty_patch(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + + # Act + patch = patch_wrapper.generate_patch(Files.CONFIG_DB_AS_JSON, Files.CONFIG_DB_AS_JSON) + + # Assert + self.assertFalse(patch) + + def test_simulate_patch__empty_patch__no_changes(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + patch = jsonpatch.JsonPatch([]) + expected = Files.CONFIG_DB_AS_JSON + + # Act + actual = patch_wrapper.simulate_patch(patch, Files.CONFIG_DB_AS_JSON) + + # Assert + self.assertDictEqual(expected, actual) + + def test_simulate_patch__non_empty_patch__changes_applied(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + patch = Files.SINGLE_OPERATION_CONFIG_DB_PATCH + expected = Files.SINGLE_OPERATION_CONFIG_DB_PATCH.apply(Files.CONFIG_DB_AS_JSON) + + # Act + actual = patch_wrapper.simulate_patch(patch, Files.CONFIG_DB_AS_JSON) + + # Assert + self.assertDictEqual(expected, actual) + + def test_generate_patch__diff__non_empty_patch(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper() + after_update_json = Files.SINGLE_OPERATION_CONFIG_DB_PATCH.apply(Files.CONFIG_DB_AS_JSON) + expected = Files.SINGLE_OPERATION_CONFIG_DB_PATCH + + # Act + actual = patch_wrapper.generate_patch(Files.CONFIG_DB_AS_JSON, after_update_json) + + # Assert + self.assertTrue(actual) + self.assertEqual(expected, actual) + + def test_convert_config_db_patch_to_sonic_yang_patch__empty_patch__returns_empty_patch(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper(config_wrapper = self.config_wrapper_mock) + patch = jsonpatch.JsonPatch([]) + expected = jsonpatch.JsonPatch([]) + + # Act + actual = patch_wrapper.convert_config_db_patch_to_sonic_yang_patch(patch) + + # Assert + self.assertEqual(expected, actual) + + def test_convert_config_db_patch_to_sonic_yang_patch__single_operation_patch__returns_sonic_yang_patch(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper(config_wrapper = self.config_wrapper_mock) + patch = Files.SINGLE_OPERATION_CONFIG_DB_PATCH + expected = Files.SINGLE_OPERATION_SONIC_YANG_PATCH + + # Act + actual = patch_wrapper.convert_config_db_patch_to_sonic_yang_patch(patch) + + # Assert + self.assertEqual(expected, actual) + + def test_convert_config_db_patch_to_sonic_yang_patch__multiple_operations_patch__returns_sonic_yang_patch(self): + # Arrange + config_wrapper = self.config_wrapper_mock + patch_wrapper = gu_common.PatchWrapper(config_wrapper = config_wrapper) + config_db_patch = Files.MULTI_OPERATION_CONFIG_DB_PATCH + + # Act + sonic_yang_patch = patch_wrapper.convert_config_db_patch_to_sonic_yang_patch(config_db_patch) + + # Assert + self.__assert_same_patch(config_db_patch, sonic_yang_patch, config_wrapper, patch_wrapper) + + def test_convert_sonic_yang_patch_to_config_db_patch__empty_patch__returns_empty_patch(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper(config_wrapper = self.config_wrapper_mock) + patch = jsonpatch.JsonPatch([]) + expected = jsonpatch.JsonPatch([]) + + # Act + actual = patch_wrapper.convert_sonic_yang_patch_to_config_db_patch(patch) + + # Assert + self.assertEqual(expected, actual) + + def test_convert_sonic_yang_patch_to_config_db_patch__single_operation_patch__returns_config_db_patch(self): + # Arrange + patch_wrapper = gu_common.PatchWrapper(config_wrapper = self.config_wrapper_mock) + patch = Files.SINGLE_OPERATION_SONIC_YANG_PATCH + expected = Files.SINGLE_OPERATION_CONFIG_DB_PATCH + + # Act + actual = patch_wrapper.convert_sonic_yang_patch_to_config_db_patch(patch) + + # Assert + self.assertEqual(expected, actual) + + def test_convert_sonic_yang_patch_to_config_db_patch__multiple_operations_patch__returns_config_db_patch(self): + # Arrange + config_wrapper = self.config_wrapper_mock + patch_wrapper = gu_common.PatchWrapper(config_wrapper = config_wrapper) + sonic_yang_patch = Files.MULTI_OPERATION_SONIC_YANG_PATCH + + # Act + config_db_patch = patch_wrapper.convert_sonic_yang_patch_to_config_db_patch(sonic_yang_patch) + + # Assert + self.__assert_same_patch(config_db_patch, sonic_yang_patch, config_wrapper, patch_wrapper) + + def __assert_same_patch(self, config_db_patch, sonic_yang_patch, config_wrapper, patch_wrapper): + sonic_yang = config_wrapper.get_sonic_yang_as_json() + config_db = config_wrapper.get_config_db_as_json() + + after_update_sonic_yang = patch_wrapper.simulate_patch(sonic_yang_patch, sonic_yang) + after_update_config_db = patch_wrapper.simulate_patch(config_db_patch, config_db) + after_update_config_db_cropped = config_wrapper.crop_tables_without_yang(after_update_config_db) + + after_update_sonic_yang_as_config_db = \ + config_wrapper.convert_sonic_yang_to_config_db(after_update_sonic_yang) + + self.assertTrue(patch_wrapper.verify_same_json(after_update_config_db_cropped, after_update_sonic_yang_as_config_db)) diff --git a/tests/generic_config_updater/gutest_helpers.py b/tests/generic_config_updater/gutest_helpers.py new file mode 100644 index 000000000000..2e8984ad683c --- /dev/null +++ b/tests/generic_config_updater/gutest_helpers.py @@ -0,0 +1,53 @@ +import json +import jsonpatch +import os +import shutil +import sys +import unittest +from unittest.mock import MagicMock, Mock, call + +class MockSideEffectDict: + def __init__(self, map): + self.map = map + + def side_effect_func(self, *args): + l = [str(arg) for arg in args] + key = tuple(l) + value = self.map.get(key) + if value is None: + raise ValueError(f"Given arguments were not found in arguments map.\n Arguments: {key}\n Map: {self.map}") + + return value + +def create_side_effect_dict(map): + return MockSideEffectDict(map).side_effect_func + +class FilesLoader: + def __init__(self): + self.files_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "files") + self.cache = {} + + def __getattr__(self, attr): + return self._load(attr) + + def _load(self, file_name): + normalized_file_name = file_name.lower() + + # Try load json file + json_file_path = os.path.join(self.files_path, f"{normalized_file_name}.json") + if os.path.isfile(json_file_path): + with open(json_file_path) as fh: + text = fh.read() + return json.loads(text) + + # Try load json-patch file + jsonpatch_file_path = os.path.join(self.files_path, f"{normalized_file_name}.json-patch") + if os.path.isfile(jsonpatch_file_path): + with open(jsonpatch_file_path) as fh: + text = fh.read() + return jsonpatch.JsonPatch(json.loads(text)) + + raise ValueError(f"There is no file called '{file_name}' in 'files/' directory") + +# Files.File_Name will look for a file called "file_name" in the "files/" directory +Files = FilesLoader()