diff --git a/tests/unit/cliconf/test_utils_vyosconf.py b/tests/unit/cliconf/test_utils_vyosconf.py
index 044c2968..dbc296e6 100644
--- a/tests/unit/cliconf/test_utils_vyosconf.py
+++ b/tests/unit/cliconf/test_utils_vyosconf.py
@@ -1,224 +1,217 @@
#
# 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 unittest
-from ansible_collections.vyos.vyos.plugins.cliconf_utils.vyosconf import VyosConf
+from ansible_collections.vyos.vyos.plugins.cliconf_utils.vyosconf import (
+ KEEP_EXISTING_VALUES,
+ VyosConf,
+)
class TestListElements(unittest.TestCase):
def test_add(self):
conf = VyosConf()
conf.set_entry(["a", "b"], "c")
self.assertEqual(conf.config, {"a": {"b": {"c": {}}}})
conf.set_entry(["a", "b"], "d")
self.assertEqual(conf.config, {"a": {"b": {"c": {}, "d": {}}}})
conf.set_entry(["a", "c"], "b")
self.assertEqual(
conf.config,
{"a": {"b": {"c": {}, "d": {}}, "c": {"b": {}}}},
)
conf.set_entry(["a", "c", "b"], "d")
self.assertEqual(
conf.config,
{"a": {"b": {"c": {}, "d": {}}, "c": {"b": {"d": {}}}}},
)
def test_del(self):
conf = VyosConf()
conf.set_entry(["a", "b"], "c")
conf.set_entry(["a", "c", "b"], "d")
conf.set_entry(["a", "b"], "d")
self.assertEqual(
conf.config,
{"a": {"b": {"c": {}, "d": {}}, "c": {"b": {"d": {}}}}},
)
conf.del_entry(["a", "c", "b"], "d")
self.assertEqual(conf.config, {"a": {"b": {"c": {}, "d": {}}}})
conf.set_entry(["a", "b", "c"], "d")
conf.del_entry(["a", "b", "c"], "d")
self.assertEqual(conf.config, {"a": {"b": {"d": {}}}})
def test_del_missing_leaf_is_noop(self):
"""
Deleting a leaf that was never set must leave the config unchanged.
Regression test: del_entry() used to raise KeyError when the leaf's
parent had siblings, and could delete an unrelated ancestor subtree
(or the entire config) when the parent path had no siblings.
"""
# parent has siblings: previously raised KeyError
conf = VyosConf()
conf.set_entry(["a", "b"], "c")
conf.set_entry(["a", "b"], "d")
conf.del_entry(["a", "b"], "nonexistent")
self.assertEqual(conf.config, {"a": {"b": {"c": {}, "d": {}}}})
# parent path is an unbranched chain: previously deleted the
# entire config instead of no-op'ing
conf = VyosConf()
conf.set_entry(["a", "b"], "d")
conf.del_entry(["a", "b"], "c")
self.assertEqual(conf.config, {"a": {"b": {"d": {}}}})
# missing intermediate path element already behaved correctly;
# confirm it still does
conf = VyosConf()
conf.set_entry(["a", "b"], "c")
conf.del_entry(["a", "x"], "c")
self.assertEqual(conf.config, {"a": {"b": {"c": {}}}})
def test_parse(self):
conf = VyosConf()
self.assertListEqual(
conf.parse_line("set a b c"),
["set", ["a", "b"], "c"],
)
self.assertListEqual(
conf.parse_line('set a b "c"'),
["set", ["a", "b"], "c"],
)
self.assertListEqual(
conf.parse_line("set a b 'c d'"),
["set", ["a", "b"], "c d"],
)
self.assertListEqual(
conf.parse_line("set a b 'c'"),
["set", ["a", "b"], "c"],
)
self.assertListEqual(
conf.parse_line("delete a b 'c'"),
["delete", ["a", "b"], "c"],
)
self.assertListEqual(
conf.parse_line("del a b 'c'"),
["del", ["a", "b"], "c"],
)
self.assertListEqual(
conf.parse_line("set a b '\"c'"),
["set", ["a", "b"], '"c'],
)
self.assertListEqual(
conf.parse_line("set a b 'c' #this is a comment"),
["set", ["a", "b"], "c"],
)
self.assertListEqual(
conf.parse_line("set a b '#c'"),
["set", ["a", "b"], "#c"],
)
def test_run_commands(self):
self.assertEqual(
VyosConf(["set a b 'c'", "set a c 'b'"]).config,
{"a": {"b": {"c": {}}, "c": {"b": {}}}},
)
self.assertEqual(
VyosConf(["set a b c 'd'", "set a c 'b'", "del a b c d"]).config,
{"a": {"c": {"b": {}}}},
)
def test_build_commands(self):
self.assertEqual(
sorted(
VyosConf(
[
"set a b 'c a'",
"set a c a",
"set a c b",
"delete a c a",
],
).build_commands(),
),
sorted(["set a b 'c a'", "set a c b"]),
)
self.assertEqual(
sorted(
VyosConf(
[
"set a b 10.0.0.1/24",
"set a c ABCabc123+/=",
"set a d $6$ABC.abc.123.+./=..",
],
).build_commands(),
),
sorted(
[
"set a b 10.0.0.1/24",
"set a c 'ABCabc123+/='",
"set a d '$6$ABC.abc.123.+./=..'",
],
),
)
def test_check_commands(self):
conf = VyosConf(["set a b 'c a'", "set a c b"])
self.assertListEqual(
conf.check_commands(
["set a b 'c a'", "del a c b", "set a b 'c'", "del a a a"],
),
[True, False, False, True],
)
def test_diff_commands_to(self):
conf = VyosConf(["set a b 'c a'", "set a c b"])
self.assertListEqual(
conf.diff_commands_to(VyosConf(["set a c b"])),
["delete a b"],
)
self.assertListEqual(
conf.diff_commands_to(VyosConf(["set a b 'c a'", "set a c b"])),
[],
)
+ # KEEP_EXISTING_VALUES is no longer reachable via 'set'/'delete'
+ # command text (see #6): a literal "..." leaf is now an ordinary
+ # value, not a sentinel, so nothing is suppressed here.
self.assertListEqual(
- conf.diff_commands_to(
- VyosConf(
- [
- "set a b ...",
- ],
- ),
- ),
- ["delete a c"],
- )
- self.assertListEqual(
- conf.diff_commands_to(VyosConf(["set a ...", "set a d e"])),
- ["set a d e"],
- )
- self.assertListEqual(
- conf.diff_commands_to(VyosConf(["set a b", "set a c b"])),
- ["delete a b 'c a'"],
+ conf.diff_commands_to(VyosConf(["set a b ..."])),
+ ["delete a b 'c a'", "delete a c", "set a b ..."],
)
- self.assertListEqual(
- conf.diff_commands_to(VyosConf(["set a b 'a c'", "set a c b"])),
- ["delete a b 'c a'", "set a b 'a c'"],
- )
+ def test_diff_commands_to_keep_existing_values_sentinel(self):
+ # KEEP_EXISTING_VALUES is only reachable via the Python API now.
+ # Build the candidate tree directly to prove diff_to() still
+ # honours it when used that way.
+ conf = VyosConf(["set a b 'c a'", "set a c b"])
+ candidate = VyosConf()
+ candidate.config = {"a": {"b": {KEEP_EXISTING_VALUES: {}}}}
self.assertListEqual(
- VyosConf(
- ["set a b c d", "set a b c e", "set a b d"],
- ).diff_commands_to(VyosConf(["set a b c d", "set a b ..."])),
- ["delete a b c e"],
+ conf.diff_commands_to(candidate),
+ ["delete a c"],
)
if __name__ == "__main__":
unittest.main()
diff --git a/tests/unit/modules/network/vyos/test_vyos_config.py b/tests/unit/modules/network/vyos/test_vyos_config.py
index 0e5bb1c0..df61361c 100644
--- a/tests/unit/modules/network/vyos/test_vyos_config.py
+++ b/tests/unit/modules/network/vyos/test_vyos_config.py
@@ -1,343 +1,412 @@
#
# (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_enforce(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="enforce"))
candidate = "\n".join(lines)
response = self.cliconf_obj.get_diff(
candidate,
self.running_config,
diff_match="enforce",
)
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="enforce",
)
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_enforce_blank_lines(self):
"""enforce 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="enforce")
self.assertIn("config_diff", response)
def test_vyos_config_match_enforce_additions(self):
lines = [
"set interfaces ethernet eth0 address '1.2.3.4/24'",
"set interfaces ethernet eth0 description 'test string'",
"set interfaces ethernet eth2 address '192.0.2.1/24'",
]
set_module_args(dict(lines=lines, match="enforce"))
candidate = "\n".join(lines)
response = self.cliconf_obj.get_diff(
candidate,
self.running_config,
diff_match="enforce",
)
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="enforce",
)
self.assertIn(
"set interfaces ethernet eth2 address 192.0.2.1/24",
response["config_diff"],
)
self.assertEqual(result["commands"], response["config_diff"])
def test_vyos_config_match_enforce_rejects_delete_lines(self):
"""
match=enforce treats the candidate as the complete desired end-state.
A candidate containing 'delete' lines must be rejected rather than
silently producing a diff that removes most/all of the running
config (regression test for a candidate that is a no-op/delete-only
input generating deletes for everything the candidate omits).
"""
lines = ["delete interfaces ethernet eth0 address"]
candidate = "\n".join(lines)
with self.assertRaises(ValueError):
self.cliconf_obj.get_diff(
candidate,
self.running_config,
diff_match="enforce",
)
def test_vyos_config_match_enforce_rejects_empty_candidate(self):
"""
A candidate that is empty, whitespace-only, or comment-only must be
rejected rather than silently treated as an empty desired end-state
(which would generate deletes for the entire running config).
Comment-only candidates are also stripped away entirely by upstream
NetworkConfig parsing before reaching VyosConf, so this is a second,
distinct route to the same mass-deletion failure mode as the
'delete' lines case above.
"""
for candidate in ("", " ", "# just a comment"):
with self.assertRaises(ValueError):
self.cliconf_obj.get_diff(
candidate,
self.running_config,
diff_match="enforce",
)
def test_vyos_config_match_enforce_requires_running(self):
"""
diff_match=enforce with running=None must raise a clear ValueError
instead of falling through to an AttributeError on
running.splitlines().
"""
with self.assertRaises(ValueError):
self.cliconf_obj.get_diff(
"set system host-name foo",
None,
diff_match="enforce",
)
def test_vyos_config_match_enforce_ignores_comment_lines(self):
"""
Comment lines mixed in with 'set' lines must be stripped out rather
than causing the whole candidate to be rejected as not starting
with 'set'.
"""
candidate = "\n".join(
[
"set interfaces ethernet eth0 address '1.2.3.4/24'",
"# a note about this interface",
"set interfaces ethernet eth0 description 'test string'",
],
)
running = "set interfaces ethernet eth0 address '1.2.3.4/24'"
response = self.cliconf_obj.get_diff(
candidate,
running,
diff_match="enforce",
)
self.assertIn(
"set interfaces ethernet eth0 description 'test string'",
response["config_diff"],
)
def test_sanitize_config_filters_password_delete_lines(self):
"""
sanitize_config()/PASSWORD_NEEDLE must filter 'delete ... password'
lines the same way it filters 'set ... password' lines, since
match=enforce can generate deletes for password config the candidate
omits. Without this, allow_password_change=none/plaintext/encrypted
would fail to catch a password-affecting delete.
"""
result = {}
commands = [
"set system host-name foo",
"delete system login user admin authentication encrypted-password",
"set system login user admin authentication plaintext-password 'secret'",
]
vyos_config.sanitize_config(commands, result, allow="none")
self.assertIn(
"delete system login user admin authentication encrypted-password",
result["filtered"],
)
self.assertIn(
"set system login user admin authentication plaintext-password 'secret'",
result["filtered"],
)
self.assertNotIn("set system host-name foo", result["filtered"])
+
+ def test_vyos_config_match_enforce_refuses_ssh_deletion(self):
+ """
+ match=enforce must refuse to generate 'delete service ssh ...'
+ commands, since this could sever the management connection.
+ Regression test for the incident where an enforce candidate that
+ didn't restate 'service ssh' generated a delete for it.
+ """
+ running = "\n".join(
+ [
+ "set service ssh port '22'",
+ "set service lldp",
+ ],
+ )
+ candidate = "set service lldp"
+
+ with self.assertRaises(ValueError):
+ self.cliconf_obj.get_diff(
+ candidate,
+ running,
+ diff_match="enforce",
+ )
+
+ def test_vyos_config_confirm_defaults_to_automatic_for_match_enforce(self):
+ """
+ confirm defaults to 'automatic' when match=enforce and confirm is
+ not explicitly set, since enforce can generate broad deletes and a
+ bad commit should self-revert rather than leave the device
+ unreachable.
+ """
+ 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="enforce"))
+ candidate = "\n".join(lines)
+ response = self.cliconf_obj.get_diff(
+ candidate,
+ self.running_config,
+ diff_match="enforce",
+ )
+ self.conn.get_diff = MagicMock(return_value=response)
+
+ self.execute_module(changed=True, sort=False)
+
+ self.assertEqual(self.load_config.call_args[1]["confirm"], 10)
+ self.run_commands.assert_called_once()
+ self.assertEqual(
+ ["configure", "confirm", "exit"],
+ self.run_commands.call_args[0][1],
+ )
+
+ def test_vyos_config_confirm_stays_none_for_other_match_values(self):
+ """
+ confirm stays 'none' (no confirm kwarg passed, no auto-confirm
+ run_commands call) when match is not 'enforce' and confirm is not
+ explicitly set -- the new conditional default must not change
+ existing behaviour for match=line/none.
+ """
+ lines = ["set system host-name foo"]
+ set_module_args(dict(lines=lines))
+ 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.assertIsNone(self.load_config.call_args[1]["confirm"])
+ self.run_commands.assert_not_called()