diff --git a/.gitignore b/.gitignore index 44b5f79c..1dbe8b0e 100644 --- a/.gitignore +++ b/.gitignore @@ -1,129 +1,128 @@ # CML/virl lab cache .virl/ # A collection directory, resulting from the use of the pytest-ansible-units plugin collections/ # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] *$py.class # C extensions *.so # Distribution / packaging .Python build/ develop-eggs/ dist/ downloads/ eggs/ .eggs/ lib/ lib64/ parts/ sdist/ var/ wheels/ *.egg-info/ .installed.cfg *.egg MANIFEST # PyInstaller # Usually these files are written by a python script from a template # before PyInstaller builds the exe, so as to inject date/other infos into it. *.manifest *.spec # Installer logs pip-log.txt pip-delete-this-directory.txt # Unit test / coverage reports htmlcov/ .tox/ .coverage .coverage.* .cache nosetests.xml coverage.xml *.cover .hypothesis/ .pytest_cache/ # Translations *.mo *.pot # Django stuff: *.log local_settings.py db.sqlite3 # Flask stuff: instance/ .webassets-cache # Scrapy stuff: .scrapy # Sphinx documentation docs/_build/ # PyBuilder target/ # Jupyter Notebook .ipynb_checkpoints # pyenv .python-version # celery beat schedule file celerybeat-schedule # SageMath parsed files *.sage.py # Environments .env .venv env/ venv/ ENV/ env.bak/ venv.bak/ -*.bak # Spyder project settings .spyderproject .spyproject # Rope project settings .ropeproject # mkdocs documentation /site # mypy .mypy_cache/ # ide *.code-workspace .vscode/ .DS_Store changelogs/.plugin-cache.yaml # inventory for testing inventory.network *.bak # Git worktrees .worktrees/ # Claude Code .claude/ diff --git a/plugins/cliconf_utils/vyosconf.py b/plugins/cliconf_utils/vyosconf.py index 29ee404c..fab58112 100644 --- a/plugins/cliconf_utils/vyosconf.py +++ b/plugins/cliconf_utils/vyosconf.py @@ -1,235 +1,235 @@ # # This file is part of Ansible # # Ansible is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # Ansible is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Ansible. If not, see . # from __future__ import absolute_import, division, print_function __metaclass__ = type import re KEEP_EXISTING_VALUES = "..." class VyosConf: def __init__(self, commands=None): self.config = {} if isinstance(commands, list): self.run_commands(commands) def set_entry(self, path, leaf): """ This function sets a value in the configuration given a path. :param path: list of strings to traverse in the config :param leaf: value to set at the destination :return: dict """ target = self.config path = path + [leaf] for key in path: if key not in target or not isinstance(target[key], dict): target[key] = {} target = target[key] return self.config def del_entry(self, path, leaf): """ This function deletes a value from the configuration given a path and also removes all the parents that are now empty. :param path: list of strings to traverse in the config :param leaf: value to delete at the destination :return: dict """ target = self.config first_no_sibling_key = None for key in path: if key not in target: return self.config if len(target[key]) <= 1: if first_no_sibling_key is None: first_no_sibling_key = [target, key] else: first_no_sibling_key = None target = target[key] if first_no_sibling_key is None: first_no_sibling_key = [target, leaf] target = first_no_sibling_key[0] target_key = first_no_sibling_key[1] del target[target_key] return self.config def check_entry(self, path, leaf): """ This function checks if a value exists in the config. :param path: list of strings to traverse in the config :param leaf: value to check for existence :return: bool """ target = self.config path = path + [leaf] existing = [] for key in path: if key not in target or not isinstance(target[key], dict): return False existing.append(key) target = target[key] return True def parse_line(self, line): """ This function parses a given command from string. :param line: line to parse :return: [command, path, leaf] """ line = re.match(r"^('(.*)'|\"(.*)\"|([^#\"']*))*", line).group(0).strip() path = re.findall(r"('.*?'|\".*?\"|\S+)", line) leaf = path[-1] if leaf.startswith('"') and leaf.endswith('"'): leaf = leaf[1:-1] if leaf.startswith("'") and leaf.endswith("'"): leaf = leaf[1:-1] return [path[0], path[1:-1], leaf] def run_command(self, command): """ This function runs a given command string. :param command: command to run :return: dict """ [cmd, path, leaf] = self.parse_line(command) if cmd.startswith("set"): self.set_entry(path, leaf) if cmd.startswith("del"): self.del_entry(path, leaf) return self.config def run_commands(self, commands): """ - This function runs a a list of command strings. + This function runs a list of command strings. :param commands: commands to run :return: dict """ for c in commands: self.run_command(c) return self.config def check_command(self, command): """ This function checks a command for existence in the config. :param command: command to check :return: bool """ [cmd, path, leaf] = self.parse_line(command) if cmd.startswith("set"): return self.check_entry(path, leaf) if cmd.startswith("del"): return not self.check_entry(path, leaf) return True def check_commands(self, commands): """ This function checks a list of commands for existence in the config. :param commands: list of commands to check :return: [bool] """ return [self.check_command(c) for c in commands] def quote_key(self, key): """ This function adds quotes to key if quotes are needed for correct parsing. :param key: str to wrap in quotes if needed :return: str """ if len(key) == 0: return "" if '"' in key: return "'" + key + "'" if "'" in key: return '"' + key + '"' if not re.match(r"^[a-zA-Z0-9./-]*$", key): return "'" + key + "'" return key def build_commands(self, structure=None, nested=False): """ This function builds a list of commands to recreate the current configuration. :return: [str] """ if not isinstance(structure, dict): structure = self.config if len(structure) == 0: return [""] if nested else [] commands = [] for key, value in structure.items(): quoted_key = self.quote_key(key) for c in self.build_commands(value, True): commands.append((quoted_key + " " + c).strip()) if nested: return commands return ["set " + c for c in commands] def diff_to(self, other, structure): if not isinstance(other, dict): other = {} if len(structure) == 0: return ([], [""]) if not isinstance(structure, dict): structure = {} if len(other) == 0: return ([""], []) if len(other) == 0 and len(structure) == 0: return ([], []) toset = [] todel = [] for key in structure.keys(): quoted_key = self.quote_key(key) if key in other: # keys in both configs, pls compare subkeys (subset, subdel) = self.diff_to(other[key], structure[key]) for s in subset: toset.append(quoted_key + " " + s) for d in subdel: todel.append(quoted_key + " " + d) else: # keys only in this, delete if KEEP_EXISTING_VALUES not set if KEEP_EXISTING_VALUES not in other: todel.append(quoted_key) continue # del for key, value in other.items(): if key == KEEP_EXISTING_VALUES: continue quoted_key = self.quote_key(key) if key not in structure: # keys only in other, pls set all subkeys (subset, subdel) = self.diff_to(other[key], None) for s in subset: toset.append(quoted_key + " " + s) return (toset, todel) def diff_commands_to(self, other): """ This function calculates the required commands to change the current into the given configuration. :param other: VyosConf :return: [str] """ (toset, todel) = self.diff_to(other.config, self.config) return ["delete " + c.strip() for c in todel] + ["set " + c.strip() for c in toset] diff --git a/tests/unit/modules/network/vyos/test_vyos_config.py b/tests/unit/modules/network/vyos/test_vyos_config.py index 0198538b..67fb431f 100644 --- a/tests/unit/modules/network/vyos/test_vyos_config.py +++ b/tests/unit/modules/network/vyos/test_vyos_config.py @@ -1,202 +1,212 @@ # # (c) 2016 Red Hat Inc. # # This file is part of Ansible # # Ansible is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # Ansible is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Ansible. If not, see . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type from unittest.mock import MagicMock, patch from ansible_collections.vyos.vyos.plugins.cliconf.vyos import Cliconf from ansible_collections.vyos.vyos.plugins.modules import vyos_config from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosConfigModule(TestVyosModule): module = vyos_config def setUp(self): super(TestVyosConfigModule, self).setUp() self.mock_get_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_config.get_config", ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_config.load_config", ) self.load_config = self.mock_load_config.start() self.mock_run_commands = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_config.run_commands", ) self.run_commands = self.mock_run_commands.start() self.mock_get_connection = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_config.get_connection", ) self.get_connection = self.mock_get_connection.start() self.cliconf_obj = Cliconf(MagicMock()) self.running_config = load_fixture("vyos_config_config.cfg") self.conn = self.get_connection() self.conn.edit_config = MagicMock() self.running_config = load_fixture("vyos_config_config.cfg") def tearDown(self): super(TestVyosConfigModule, self).tearDown() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_run_commands.stop() self.mock_get_connection.stop() def load_fixtures(self, commands=None, filename=None): config_file = "vyos_config_config.cfg" self.get_config.return_value = load_fixture(config_file) self.load_config.return_value = None def test_vyos_config_unchanged(self): src = load_fixture("vyos_config_config.cfg") self.conn.get_diff = MagicMock(return_value=self.cliconf_obj.get_diff(src, src)) set_module_args(dict(src=src)) self.execute_module() def test_vyos_config_src(self): src = load_fixture("vyos_config_src.cfg") set_module_args(dict(src=src)) candidate = "\n".join(self.module.format_commands(src.splitlines())) commands = [ "set system host-name foo", "delete interfaces ethernet eth0 address", ] self.conn.get_diff = MagicMock( return_value=self.cliconf_obj.get_diff(candidate, self.running_config), ) self.execute_module(changed=True, commands=commands) def test_vyos_config_src_brackets(self): src = load_fixture("vyos_config_src_brackets.cfg") set_module_args(dict(src=src)) commands = [ "set interfaces ethernet eth0 address 10.10.10.10/24", "set policy route testroute rule 1 set table 10", "set system host-name foo", ] self.conn.get_diff = MagicMock(side_effect=self.cliconf_obj.get_diff) self.execute_module(changed=True, commands=commands) def test_vyos_config_backup(self): set_module_args(dict(backup=True)) result = self.execute_module() self.assertIn("__backup__", result) def test_vyos_config_lines(self): commands = ["set system host-name foo"] set_module_args(dict(lines=commands)) candidate = "\n".join(commands) self.conn.get_diff = MagicMock( return_value=self.cliconf_obj.get_diff(candidate, self.running_config), ) self.execute_module(changed=True, commands=commands) def test_vyos_config_config(self): config = "set system host-name localhost" new_config = ["set system host-name router"] set_module_args(dict(lines=new_config, config=config)) candidate = "\n".join(new_config) self.conn.get_diff = MagicMock(return_value=self.cliconf_obj.get_diff(candidate, config)) self.execute_module(changed=True, commands=new_config) def test_vyos_config_match_none(self): lines = [ "set system interfaces ethernet eth0 address 1.2.3.4/24", "set system interfaces ethernet eth0 description test string", ] set_module_args(dict(lines=lines, match="none")) candidate = "\n".join(lines) self.conn.get_diff = MagicMock( return_value=self.cliconf_obj.get_diff(candidate, None, diff_match="none"), ) self.execute_module(changed=True, commands=lines, sort=False) def test_vyos_config_match_smart(self): lines = [ - "set system interfaces ethernet eth0 address 1.2.3.4/24", - "set system interfaces ethernet eth0 description test string", + "set interfaces ethernet eth0 address '1.2.3.4/24'", + "set interfaces ethernet eth0 description 'test string'", ] set_module_args(dict(lines=lines, match="smart")) candidate = "\n".join(lines) - self.conn.get_diff = MagicMock( - return_value=self.cliconf_obj.get_diff( - candidate, - self.running_config, - diff_match="smart", - ), + + response = self.cliconf_obj.get_diff( + candidate, + self.running_config, + diff_match="smart", ) - self.execute_module(changed=True, sort=False) + + self.conn.get_diff = MagicMock(return_value=response) + result = self.execute_module(changed=True, sort=False) self.conn.get_diff.assert_called_once_with( candidate=candidate, running=self.running_config, diff_match="smart", ) + expected_config_diff = [ + "delete system", + "delete interfaces ethernet eth1", + ] + self.assertEqual(response["config_diff"], expected_config_diff) + + expected_commands = expected_config_diff + self.assertEqual(result["commands"], expected_commands) + def test_vyos_config_confirm_automatic(self): src = load_fixture("vyos_config_src.cfg") confirm_timeout = 7 set_module_args(dict(src=src, confirm="automatic", confirm_timeout=confirm_timeout)) candidate = "\n".join(self.module.format_commands(src.splitlines())) commands = [ "set system host-name foo", "delete interfaces ethernet eth0 address", ] self.conn.get_diff = MagicMock( return_value=self.cliconf_obj.get_diff(candidate, self.running_config), ) self.execute_module(changed=True, commands=commands) self.assertEqual(self.load_config.call_args[1]["confirm"], confirm_timeout) self.run_commands.assert_called_once() self.assertEqual( ["configure", "confirm", "exit"], self.run_commands.call_args[0][1], ) def test_vyos_config_confirm_manual(self): lines = [ "set system host-name foo", ] confirm_timeout = 12 set_module_args(dict(lines=lines, confirm="manual", confirm_timeout=confirm_timeout)) candidate = "\n".join(lines) self.conn.get_diff = MagicMock( return_value=self.cliconf_obj.get_diff(candidate, self.running_config), ) self.execute_module(changed=True, commands=lines) self.assertEqual(self.load_config.call_args[1]["confirm"], confirm_timeout) self.run_commands.assert_not_called()