Page MenuHomeVyOS Platform
Files F119063688

No OneTemporary
Actions

Size
18 KB
Referenced Files
None
Subscribers
None

View Options

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 <http://www.gnu.org/licenses/>.
#
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 <http://www.gnu.org/licenses/>.
# 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()

File Metadata

Mime Type
text/x-diff
Expires
Fri, Nov 6, 9:00 AM (1 d, 4 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
4653662
Default Alt Text
(18 KB)

Event Timeline