diff --git a/changelogs/fragments/fix_pylint_issues.yaml b/changelogs/fragments/fix_pylint_issues.yaml
new file mode 100644
index 0000000..6e278f6
--- /dev/null
+++ b/changelogs/fragments/fix_pylint_issues.yaml
@@ -0,0 +1,3 @@
+---
+trivial:
+ - Fix pylint failures seen in sanity tests.
diff --git a/plugins/cliconf/vyos.py b/plugins/cliconf/vyos.py
index d63c677..339aed9 100644
--- a/plugins/cliconf/vyos.py
+++ b/plugins/cliconf/vyos.py
@@ -1,361 +1,361 @@
#
# (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
cliconf: 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
default: []
vars:
- name: ansible_vyos_config_commands
"""
import re
import json
from ansible.errors import AnsibleConnectionFailure
from ansible.module_utils._text import to_text
from ansible.module_utils.common._collections_compat import Mapping
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.plugins.cliconf import CliconfBase
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)
match = re.search(r"HW 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, comment=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)
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,
- output=None,
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):
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
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 entry.startswith(item) 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", "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/sanity/ignore-2.12.txt b/tests/sanity/ignore-2.12.txt
index d753480..496cf86 100644
--- a/tests/sanity/ignore-2.12.txt
+++ b/tests/sanity/ignore-2.12.txt
@@ -1,12 +1,12 @@
plugins/action/vyos.py action-plugin-docs # base class for deprecated network platform modules using `connection: local`
plugins/module_utils/network/vyos/config/ospf_interfaces/ospf_interfaces.py compile-2.6!skip
plugins/module_utils/network/vyos/config/ospf_interfaces/ospf_interfaces.py import-2.6!skip
plugins/module_utils/network/vyos/config/route_maps/route_maps.py compile-2.6!skip
plugins/module_utils/network/vyos/config/route_maps/route_maps.py import-2.6!skip
plugins/modules/vyos_route_maps.py import-2.6!skip
plugins/modules/vyos_prefix_lists.py import-2.6!skip
plugins/module_utils/network/vyos/config/prefix_lists/prefix_lists.py import-2.6!skip
plugins/module_utils/network/vyos/config/prefix_lists/prefix_lists.py compile-2.6!skip
plugins/modules/vyos_logging_global.py import-2.6!skip
plugins/module_utils/network/vyos/config/logging_global/logging_global.py import-2.6!skip
-plugins/module_utils/network/vyos/config/logging_global/logging_global.py compile-2.6!skip
\ No newline at end of file
+plugins/module_utils/network/vyos/config/logging_global/logging_global.py compile-2.6!skip
diff --git a/tests/unit/mock/loader.py b/tests/unit/mock/loader.py
index c21188e..aea5df5 100644
--- a/tests/unit/mock/loader.py
+++ b/tests/unit/mock/loader.py
@@ -1,116 +1,116 @@
# (c) 2012-2014, Michael DeHaan
#
# 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
import os
from ansible.errors import AnsibleParserError
from ansible.parsing.dataloader import DataLoader
from ansible.module_utils._text import to_bytes, to_text
class DictDataLoader(DataLoader):
def __init__(self, file_mapping=None):
file_mapping = {} if file_mapping is None else file_mapping
assert type(file_mapping) == dict
super(DictDataLoader, self).__init__()
self._file_mapping = file_mapping
self._build_known_directories()
self._vault_secrets = None
def load_from_file(self, path, cache=True, unsafe=False):
path = to_text(path)
if path in self._file_mapping:
return self.load(self._file_mapping[path], path)
return None
# TODO: the real _get_file_contents returns a bytestring, so we actually convert the
# unicode/text it's created with to utf-8
- def _get_file_contents(self, path):
- path = to_text(path)
+ def _get_file_contents(self, file_name):
+ path = to_text(file_name)
if path in self._file_mapping:
- return (to_bytes(self._file_mapping[path]), False)
+ return (to_bytes(self._file_mapping[file_name]), False)
else:
- raise AnsibleParserError("file not found: %s" % path)
+ raise AnsibleParserError("file not found: %s" % file_name)
def path_exists(self, path):
path = to_text(path)
return path in self._file_mapping or path in self._known_directories
def is_file(self, path):
path = to_text(path)
return path in self._file_mapping
def is_directory(self, path):
path = to_text(path)
return path in self._known_directories
def list_directory(self, path):
ret = []
path = to_text(path)
for x in list(self._file_mapping.keys()) + self._known_directories:
if x.startswith(path):
if os.path.dirname(x) == path:
ret.append(os.path.basename(x))
return ret
def is_executable(self, path):
# FIXME: figure out a way to make paths return true for this
return False
def _add_known_directory(self, directory):
if directory not in self._known_directories:
self._known_directories.append(directory)
def _build_known_directories(self):
self._known_directories = []
for path in self._file_mapping:
dirname = os.path.dirname(path)
while dirname not in ("/", ""):
self._add_known_directory(dirname)
dirname = os.path.dirname(dirname)
def push(self, path, content):
rebuild_dirs = False
if path not in self._file_mapping:
rebuild_dirs = True
self._file_mapping[path] = content
if rebuild_dirs:
self._build_known_directories()
def pop(self, path):
if path in self._file_mapping:
del self._file_mapping[path]
self._build_known_directories()
def clear(self):
self._file_mapping = dict()
self._known_directories = []
def get_basedir(self):
return os.getcwd()
def set_vault_secrets(self, vault_secrets):
self._vault_secrets = vault_secrets
diff --git a/tests/unit/modules/network/vyos/test_vyos_interface.py b/tests/unit/modules/network/vyos/test_vyos_interface.py
index 1194e3b..e0e4bfc 100644
--- a/tests/unit/modules/network/vyos/test_vyos_interface.py
+++ b/tests/unit/modules/network/vyos/test_vyos_interface.py
@@ -1,288 +1,288 @@
# (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 ansible_collections.vyos.vyos.tests.unit.compat.mock import patch
from ansible_collections.vyos.vyos.plugins.modules import vyos_interface
from ansible_collections.vyos.vyos.tests.unit.modules.utils import (
set_module_args,
)
from .vyos_module import TestVyosModule, load_fixture
class TestVyosInterfaceModule(TestVyosModule):
module = vyos_interface
def setUp(self):
super(TestVyosInterfaceModule, self).setUp()
self.mock_get_config = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_interface.get_config"
)
self.get_config = self.mock_get_config.start()
self.mock_load_config = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_interface.load_config"
)
self.load_config = self.mock_load_config.start()
self.mock_execute_interfaces_command = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_interface.get_interfaces_data"
)
self.execute_interfaces_command = (
self.mock_execute_interfaces_command.start()
)
self.mock_execute_lldp_command = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_interface.get_lldp_neighbor"
)
self.execute_lldp_command = self.mock_execute_lldp_command.start()
# self.mock_get_config = patch(
# "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config"
# )
# self.get_config = self.mock_get_config.start()
# self.mock_load_config = patch(
# "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config"
# )
# self.load_config = self.mock_load_config.start()
# self.mock_get_resource_connection_config = patch(
# "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection"
# )
# self.get_resource_connection_config = (
# self.mock_get_resource_connection_config.start()
# )
# self.mock_get_resource_connection_facts = patch(
# "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection"
# )
# self.get_resource_connection_facts = (
# self.mock_get_resource_connection_facts.start()
# )
def tearDown(self):
super(TestVyosInterfaceModule, self).tearDown()
self.mock_get_config.stop()
self.mock_load_config.stop()
self.mock_execute_lldp_command.stop()
self.mock_execute_interfaces_command.stop()
- def load_fixtures(self, commands=None, transport="cli"):
+ def load_fixtures(self, commands=None, transport="cli", filename=None):
self.get_config.return_value = load_fixture(
"vyos_interface_config.cfg"
)
self.execute_interfaces_command.return_value = [
0,
load_fixture("vyos_interface_config.cfg"),
None,
]
self.execute_lldp_command.return_value = [
0,
load_fixture("vyos_lldp_neighbor_config.cfg"),
None,
]
self.load_config.return_value = dict(diff=None, session="session")
def test_vyos_setup_int(self):
set_module_args(
dict(
name="eth1",
enabled=True,
state="present",
speed="100",
duplex="half",
)
)
commands = [
"set interfaces ethernet eth1 speed 100",
"set interfaces ethernet eth1 duplex half",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_setup_required_params(self):
set_module_args(
dict(
name="eth1",
enabled=True,
state="present",
speed="100",
)
)
result = self.execute_module(failed=True)
self.assertIn(
"parameters are required together: speed, duplex", result["msg"]
)
def test_vyos_setup_int_idempotent(self):
set_module_args(
dict(
name="eth1",
enabled=True,
state="present",
)
)
self.execute_module(changed=False, commands=[])
def test_vyos_disable_int(self):
set_module_args(
dict(
name="eth1",
state="absent",
)
)
commands = ["delete interfaces ethernet eth1"]
self.execute_module(changed=True, commands=commands)
def test_vyos_setup_int_aggregate(self):
set_module_args(
dict(
aggregate=[
dict(
name="eth1",
enabled=True,
state="present",
mtu="512",
duplex="half",
speed="100",
),
dict(
name="eth2",
enabled=True,
state="present",
speed="1000",
duplex="full",
mtu="256",
),
]
)
)
commands = [
"set interfaces ethernet eth1 speed 100",
"set interfaces ethernet eth1 duplex half",
"set interfaces ethernet eth1 mtu 512",
"set interfaces ethernet eth2 speed 1000",
"set interfaces ethernet eth2 duplex full",
"set interfaces ethernet eth2 mtu 256",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_delete_int_aggregate(self):
set_module_args(
dict(
aggregate=[
dict(
name="eth1",
state="absent",
),
dict(
name="eth2",
state="absent",
),
]
)
)
commands = [
"delete interfaces ethernet eth1",
"delete interfaces ethernet eth2",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_disable_int_aggregate(self):
set_module_args(
dict(
aggregate=[
dict(
name="eth1",
enabled=False,
),
dict(
name="eth2",
enabled=False,
),
]
)
)
commands = [
"set interfaces ethernet eth1 disable",
"set interfaces ethernet eth2 disable",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_intent_wrongport(self):
set_module_args(
dict(
name="eth0",
neighbors=[dict(port="dummy_port", host="dummy_host")],
)
)
result = self.execute_module(failed=True)
self.assertIn(
"One or more conditional statements have not been satisfied",
result["msg"],
)
def test_vyos_intent_neighbor_fail(self):
set_module_args(
dict(
name="eth0",
neighbors=[
dict(
port="eth0",
)
],
)
)
result = self.execute_module(failed=True)
self.assertIn(
"One or more conditional statements have not been satisfied",
result["msg"],
)
def test_vyos_intent_neighbor(self):
set_module_args(
dict(
name="eth1",
neighbors=[
dict(
port="eth0",
)
],
)
)
self.execute_module(failed=False)
def test_vyos_intent_neighbor_aggregate(self):
set_module_args(
dict(
aggregate=[
dict(
name="eth1",
neighbors=[
dict(
port="eth0",
)
],
)
]
)
)
self.execute_module(failed=False)
diff --git a/tests/unit/modules/network/vyos/test_vyos_ping.py b/tests/unit/modules/network/vyos/test_vyos_ping.py
index f3ba2ee..60407ad 100644
--- a/tests/unit/modules/network/vyos/test_vyos_ping.py
+++ b/tests/unit/modules/network/vyos/test_vyos_ping.py
@@ -1,107 +1,107 @@
# (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 ansible_collections.vyos.vyos.tests.unit.compat.mock import patch
from ansible_collections.vyos.vyos.plugins.modules import vyos_ping
from ansible_collections.vyos.vyos.tests.unit.modules.utils import (
set_module_args,
)
from .vyos_module import TestVyosModule, load_fixture
class TestVyosPingModule(TestVyosModule):
module = vyos_ping
def setUp(self):
super(TestVyosPingModule, self).setUp()
self.mock_run_commands = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_ping.run_commands"
)
self.run_commands = self.mock_run_commands.start()
def tearDown(self):
super(TestVyosPingModule, self).tearDown()
self.mock_run_commands.stop()
def load_fixtures(self, commands=None):
def load_from_file(*args, **kwargs):
commands = kwargs["commands"]
output = list()
for command in commands:
- filename = str(command).split(" | ")[0].replace(" ", "_")
+ filename = str(command).split(" | ", 1)[0].replace(" ", "_")
output.append(load_fixture("vyos_ping_%s" % filename))
return output
self.run_commands.side_effect = load_from_file
def test_vyos_ping_expected_success(self):
"""Test for successful pings when destination should be reachable"""
set_module_args(dict(count=2, dest="10.10.10.10"))
self.execute_module()
def test_vyos_ping_expected_failure(self):
"""Test for unsuccessful pings when destination should not be reachable"""
set_module_args(dict(count=4, dest="10.10.10.20", state="absent"))
self.execute_module()
def test_vyos_ping_unexpected_success(self):
"""Test for successful pings when destination should not be reachable - FAIL."""
set_module_args(dict(count=2, dest="10.10.10.10", state="absent"))
self.execute_module(failed=True)
def test_vyos_ping_unexpected_failure(self):
"""Test for unsuccessful pings when destination should be reachable - FAIL."""
set_module_args(dict(count=4, dest="10.10.10.20"))
self.execute_module(failed=True)
def test_vyos_ping_failure_stats(self):
"""Test for asserting stats when ping fails"""
set_module_args(dict(count=4, dest="10.10.10.20"))
result = self.execute_module(failed=True)
self.assertEqual(result["packet_loss"], "100%")
self.assertEqual(result["packets_rx"], 0)
self.assertEqual(result["packets_tx"], 4)
def test_vyos_ping_success_stats(self):
"""Test for asserting stats when ping passes"""
set_module_args(dict(count=2, dest="10.10.10.10"))
result = self.execute_module()
self.assertEqual(result["packet_loss"], "0%")
self.assertEqual(result["packets_rx"], 2)
self.assertEqual(result["packets_tx"], 2)
self.assertEqual(result["rtt"]["min"], 12)
self.assertEqual(result["rtt"]["avg"], 17)
self.assertEqual(result["rtt"]["max"], 22)
self.assertEqual(result["rtt"]["mdev"], 10)
def test_vyos_ping_success_stats_with_options(self):
set_module_args(dict(count=10, ttl=128, size=512, dest="10.10.10.11"))
result = self.execute_module()
self.assertEqual(result["packet_loss"], "0%")
self.assertEqual(result["packets_rx"], 10)
self.assertEqual(result["packets_tx"], 10)
self.assertEqual(result["rtt"]["min"], 1)
self.assertEqual(result["rtt"]["avg"], 3)
self.assertEqual(result["rtt"]["max"], 21)
self.assertEqual(result["rtt"]["mdev"], 5)
diff --git a/tests/unit/modules/network/vyos/test_vyos_static_route.py b/tests/unit/modules/network/vyos/test_vyos_static_route.py
index 21f1139..00985c6 100644
--- a/tests/unit/modules/network/vyos/test_vyos_static_route.py
+++ b/tests/unit/modules/network/vyos/test_vyos_static_route.py
@@ -1,73 +1,73 @@
# (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 ansible_collections.vyos.vyos.tests.unit.compat.mock import patch
from ansible_collections.vyos.vyos.plugins.modules import vyos_static_route
from ansible_collections.vyos.vyos.tests.unit.modules.utils import (
set_module_args,
)
from .vyos_module import TestVyosModule
class TestVyosStaticRouteModule(TestVyosModule):
module = vyos_static_route
def setUp(self):
super(TestVyosStaticRouteModule, self).setUp()
self.mock_get_config = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_static_route.get_config"
)
self.get_config = self.mock_get_config.start()
self.mock_load_config = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_static_route.load_config"
)
self.load_config = self.mock_load_config.start()
def tearDown(self):
super(TestVyosStaticRouteModule, self).tearDown()
self.mock_get_config.stop()
self.mock_load_config.stop()
- def load_fixtures(self, commands=None, transport="cli"):
+ def load_fixtures(self, commands=None, transport="cli", filename=None):
self.get_config.return_value = ""
self.load_config.return_value = dict(diff=None, session="session")
def test_vyos_static_route_present(self):
set_module_args(
dict(
prefix="172.26.0.0/16",
next_hop="172.26.4.1",
admin_distance="1",
)
)
result = self.execute_module(changed=True)
self.assertEqual(
result["commands"],
[
"set protocols static route 172.26.0.0/16 next-hop 172.26.4.1 distance 1"
],
)
diff --git a/tests/unit/modules/network/vyos/test_vyos_user.py b/tests/unit/modules/network/vyos/test_vyos_user.py
index d4c2dbe..616ad09 100644
--- a/tests/unit/modules/network/vyos/test_vyos_user.py
+++ b/tests/unit/modules/network/vyos/test_vyos_user.py
@@ -1,139 +1,139 @@
# (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 ansible_collections.vyos.vyos.tests.unit.compat.mock import patch
from ansible_collections.vyos.vyos.plugins.modules import vyos_user
from ansible_collections.vyos.vyos.tests.unit.modules.utils import (
set_module_args,
)
from .vyos_module import TestVyosModule, load_fixture
class TestVyosUserModule(TestVyosModule):
module = vyos_user
def setUp(self):
super(TestVyosUserModule, self).setUp()
self.mock_get_config = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_user.get_config"
)
self.get_config = self.mock_get_config.start()
self.mock_load_config = patch(
"ansible_collections.vyos.vyos.plugins.modules.vyos_user.load_config"
)
self.load_config = self.mock_load_config.start()
def tearDown(self):
super(TestVyosUserModule, self).tearDown()
self.mock_get_config.stop()
self.mock_load_config.stop()
- def load_fixtures(self, commands=None, transport="cli"):
+ def load_fixtures(self, commands=None, transport="cli", filename=None):
self.get_config.return_value = load_fixture("vyos_user_config.cfg")
self.load_config.return_value = dict(diff=None, session="session")
def test_vyos_user_password(self):
set_module_args(dict(name="ansible", configured_password="test"))
result = self.execute_module(changed=True)
self.assertEqual(
result["commands"],
[
"set system login user ansible authentication plaintext-password test"
],
)
def test_vyos_user_delete(self):
set_module_args(dict(name="ansible", state="absent"))
result = self.execute_module(changed=True)
self.assertEqual(
result["commands"], ["delete system login user ansible"]
)
def test_vyos_user_level(self):
set_module_args(dict(name="ansible", level="operator"))
result = self.execute_module(changed=True)
self.assertEqual(
result["commands"],
["set system login user ansible level operator"],
)
def test_vyos_user_level_invalid(self):
set_module_args(dict(name="ansible", level="sysadmin"))
self.execute_module(failed=True)
def test_vyos_user_purge(self):
set_module_args(dict(purge=True))
result = self.execute_module(changed=True)
self.assertEqual(
sorted(result["commands"]),
sorted(
[
"delete system login user ansible",
"delete system login user admin",
]
),
)
def test_vyos_user_update_password_changed(self):
set_module_args(
dict(
name="test",
configured_password="test",
update_password="on_create",
)
)
result = self.execute_module(changed=True)
self.assertEqual(
result["commands"],
[
"set system login user test authentication plaintext-password test"
],
)
def test_vyos_user_update_password_on_create_ok(self):
set_module_args(
dict(
name="ansible",
configured_password="test",
update_password="on_create",
)
)
self.execute_module()
def test_vyos_user_update_password_always(self):
set_module_args(
dict(
name="ansible",
configured_password="test",
update_password="always",
)
)
result = self.execute_module(changed=True)
self.assertEqual(
result["commands"],
[
"set system login user ansible authentication plaintext-password test"
],
)