diff --git a/plugins/cliconf/vyos.py b/plugins/cliconf/vyos.py
index 87bb1d03..9e631a82 100644
--- a/plugins/cliconf/vyos.py
+++ b/plugins/cliconf/vyos.py
@@ -1,364 +1,370 @@
# (c) 2017 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 .
#
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = """
author: Ansible Networking Team (@ansible-network)
name: vyos
short_description: Use vyos cliconf to run command on VyOS platform
description:
- This vyos plugin provides low level abstraction apis for sending and receiving CLI
commands from VyOS network devices.
version_added: 1.0.0
options:
config_commands:
description:
- Specifies a list of commands that can make configuration changes
to the target device.
- When `ansible_network_single_user_mode` is enabled, if a command sent
to the device is present in this list, the existing cache is invalidated.
version_added: 2.0.0
type: list
elements: str
default: []
vars:
- name: ansible_vyos_config_commands
"""
import json
import re
from ansible.errors import AnsibleConnectionFailure
from ansible.module_utils._text import to_text
from ansible.module_utils.common._collections_compat import Mapping
from ansible.plugins.cliconf import CliconfBase
from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.config import (
NetworkConfig,
)
from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.utils import to_list
from ansible_collections.vyos.vyos.plugins.cliconf_utils.vyosconf import VyosConf
class Cliconf(CliconfBase):
__rpc__ = CliconfBase.__rpc__ + [
"commit",
"discard_changes",
"get_diff",
"run_commands",
]
def __init__(self, *args, **kwargs):
super(Cliconf, self).__init__(*args, **kwargs)
self._device_info = {}
def get_device_info(self):
if not self._device_info:
device_info = {}
device_info["network_os"] = "vyos"
reply = self.get("show version")
data = to_text(reply, errors="surrogate_or_strict").strip()
match = re.search(r"Version:\s*(.*)", data)
if match:
device_info["network_os_version"] = match.group(1)
if device_info["network_os_version"]:
match = re.search(r"VyOS\s*(\d+\.\d+)", device_info["network_os_version"])
if match:
device_info["network_os_major_version"] = match.group(1)
match = re.search(r"(?:HW|Hardware) model:\s*(\S+)", data)
if match:
device_info["network_os_model"] = match.group(1)
reply = self.get("show host name")
device_info["network_os_hostname"] = to_text(
reply,
errors="surrogate_or_strict",
).strip()
self._device_info = device_info
return self._device_info
def get_config(self, flags=None, format=None):
if format:
option_values = self.get_option_values()
if format not in option_values["format"]:
raise ValueError(
"'format' value %s is invalid. Valid values of format are %s"
% (format, ", ".join(option_values["format"])),
)
if not flags:
flags = []
if format == "text":
command = "show configuration"
else:
command = "show configuration commands"
command += " ".join(to_list(flags))
command = command.strip()
out = self.send_command(command)
return out
def edit_config(
- self, candidate=None, commit=True, replace=None, diff=False, comment=None, confirm=None
+ self,
+ candidate=None,
+ commit=True,
+ replace=None,
+ diff=False,
+ comment=None,
+ confirm=None,
):
resp = {}
operations = self.get_device_operations()
self.check_edit_config_capability(operations, candidate, commit, replace, comment)
results = []
requests = []
self.send_command("configure")
for cmd in to_list(candidate):
if not isinstance(cmd, Mapping):
cmd = {"command": cmd}
results.append(self.send_command(**cmd))
requests.append(cmd["command"])
out = self.get("compare")
out = to_text(out, errors="surrogate_or_strict")
diff_config = out if not out.startswith("No changes") else None
if diff_config:
if commit:
try:
self.commit(comment, confirm)
except AnsibleConnectionFailure as e:
msg = "commit failed: %s" % e.message
self.discard_changes()
raise AnsibleConnectionFailure(msg)
else:
self.send_command("exit")
else:
self.discard_changes()
else:
self.send_command("exit")
if (
to_text(self._connection.get_prompt(), errors="surrogate_or_strict")
.strip()
.endswith("#")
):
self.discard_changes()
if diff_config:
resp["diff"] = diff_config
resp["response"] = results
resp["request"] = requests
return resp
def get(
self,
command=None,
prompt=None,
answer=None,
sendonly=False,
newline=True,
output=None,
check_all=False,
):
if not command:
raise ValueError("must provide value of command to execute")
if output:
raise ValueError("'output' value %s is not supported for get" % output)
return self.send_command(
command=command,
prompt=prompt,
answer=answer,
sendonly=sendonly,
newline=newline,
check_all=check_all,
)
def commit(self, comment=None, confirm=None):
if confirm:
if comment:
command = 'commit-confirm {0} comment "{1}"'.format(confirm, comment)
else:
- command = 'commit-confirm {0}'.format(confirm)
+ command = "commit-confirm {0}".format(confirm)
self.send_command(command, "Proceed?", "\n")
else:
if comment:
command = 'commit comment "{0}"'.format(comment)
else:
command = "commit"
self.send_command(command)
def discard_changes(self):
self.send_command("exit discard")
def get_diff(
self,
candidate=None,
running=None,
diff_match="line",
diff_ignore_lines=None,
path=None,
diff_replace=None,
):
diff = {}
device_operations = self.get_device_operations()
option_values = self.get_option_values()
if candidate is None and device_operations["supports_generate_diff"]:
raise ValueError("candidate configuration is required to generate diff")
if diff_match not in option_values["diff_match"]:
raise ValueError(
"'match' value %s in invalid, valid values are %s"
% (diff_match, ", ".join(option_values["diff_match"])),
)
if diff_replace:
raise ValueError("'replace' in diff is not supported")
if diff_ignore_lines:
raise ValueError("'diff_ignore_lines' in diff is not supported")
if path:
raise ValueError("'path' in diff is not supported")
set_format = candidate.startswith("set") or candidate.startswith("delete")
candidate_obj = NetworkConfig(indent=4, contents=candidate)
if not set_format:
config = [c.line for c in candidate_obj.items]
commands = list()
# this filters out less specific lines
for item in config:
for index, entry in enumerate(commands):
if item.startswith(entry):
del commands[index]
break
commands.append(item)
candidate_commands = ["set %s" % cmd.replace(" {", "") for cmd in commands]
else:
candidate_commands = str(candidate).strip().split("\n")
if diff_match == "none":
diff["config_diff"] = list(candidate_commands)
return diff
if diff_match == "smart":
- running_conf = VyosConf(running.splitlines())
- candidate_conf = VyosConf(candidate_commands)
+ running_conf = VyosConf([line for line in running.splitlines() if line.strip()])
+ candidate_conf = VyosConf([line for line in candidate_commands if line.strip()])
diff["config_diff"] = running_conf.diff_commands_to(candidate_conf)
return diff
running_commands = [str(c).replace("'", "") for c in running.splitlines()]
updates = list()
visited = set()
for line in candidate_commands:
item = str(line).replace("'", "")
if not item.startswith("set") and not item.startswith("delete"):
raise ValueError("line must start with either `set` or `delete`")
elif item.startswith("set") and item not in running_commands:
updates.append(line)
elif item.startswith("delete"):
if not running_commands:
updates.append(line)
else:
item = re.sub(r"delete", "set", item)
for entry in running_commands:
if re.match(rf"^{re.escape(item)}\b", entry) and line not in visited:
updates.append(line)
visited.add(line)
diff["config_diff"] = list(updates)
return diff
def run_commands(self, commands=None, check_rc=True):
if commands is None:
raise ValueError("'commands' value is required")
responses = list()
for cmd in to_list(commands):
if not isinstance(cmd, Mapping):
cmd = {"command": cmd}
output = cmd.pop("output", None)
if output:
raise ValueError("'output' value %s is not supported for run_commands" % output)
try:
out = self.send_command(**cmd)
except AnsibleConnectionFailure as e:
if check_rc:
raise
out = getattr(e, "err", e)
responses.append(out)
return responses
def get_device_operations(self):
return {
"supports_diff_replace": False,
"supports_commit": True,
"supports_rollback": False,
"supports_defaults": False,
"supports_onbox_diff": True,
"supports_commit_comment": True,
"supports_multiline_delimiter": False,
"supports_diff_match": True,
"supports_diff_ignore_lines": False,
"supports_generate_diff": False,
"supports_replace": False,
}
def get_option_values(self):
return {
"format": ["text", "set"],
"diff_match": ["line", "smart", "none"],
"diff_replace": [],
"output": [],
}
def get_capabilities(self):
result = super(Cliconf, self).get_capabilities()
result["device_operations"] = self.get_device_operations()
result.update(self.get_option_values())
return json.dumps(result)
def set_cli_prompt_context(self):
"""
Make sure we are in the operational cli mode
:return: None
"""
if self._connection.connected:
self._update_cli_prompt_context(config_context="#", exit_command="exit discard")
diff --git a/tests/unit/modules/network/vyos/test_vyos_config.py b/tests/unit/modules/network/vyos/test_vyos_config.py
index 67fb431f..28e954e1 100644
--- a/tests/unit/modules/network/vyos/test_vyos_config.py
+++ b/tests/unit/modules/network/vyos/test_vyos_config.py
@@ -1,212 +1,219 @@
#
# (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 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)
response = self.cliconf_obj.get_diff(
candidate,
self.running_config,
diff_match="smart",
)
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()
+
+ def test_vyos_config_match_smart_blank_lines(self):
+ """smart diff must not raise IndexError on blank lines in running config."""
+ running_with_blanks = self.running_config + "\n\n"
+ candidate = "set interfaces ethernet eth0 address 1.2.3.4/24"
+ response = self.cliconf_obj.get_diff(candidate, running_with_blanks, diff_match="smart")
+ self.assertIn("config_diff", response)