sa_8001/offline_kubespray
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

collection 교체

Browse Source
This commit is contained in:
정훈 변
2024-02-23 16:37:40 +09:00
parent b494779b5b
commit 3fd554eee9
38862 changed files with 220204 additions and 6600073 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
collection/ansible/netcommon/tests/unit/plugins/action/cli_parse/fixtures/nxos_empty_parser.yaml
View File

38
collection/ansible/netcommon/tests/unit/plugins/action/cli_parse/fixtures/nxos_show_version.txt
View File

@@ -1,38 +0,0 @@
Cisco Nexus Operating System (NX-OS) Software
TAC support: http://www.cisco.com/tac
Documents: http://www.cisco.com/en/US/products/ps9372/tsd_products_support_series_home.html
Copyright (c) 2002-2018, Cisco Systems, Inc. All rights reserved.
The copyrights to certain works contained herein are owned by
other third parties and are used and distributed under license.
Some parts of this software are covered under the GNU Public
License. A copy of the license is available at
http://www.gnu.org/licenses/gpl.html.
Nexus 9000v is a demo version of the Nexus Operating System
Software
BIOS: version
NXOS: version 9.2(2)
BIOS compile time:
NXOS image file is: bootflash:///nxos.9.2.2.bin
NXOS compile time: 11/4/2018 21:00:00 [11/05/2018 06:11:06]
Hardware
cisco Nexus9000 9000v Chassis
with 8035024 kB of memory.
Processor Board ID 970MUM0NTLV
Device name: nxos101
bootflash: 3509454 kB
Kernel uptime is 33 day(s), 17 hour(s), 29 minute(s), 8 second(s)
Last reset
Reason: Unknown
System version:
Service:
plugin
Core Plugin, Ethernet Plugin
Active Package(s):

5
collection/ansible/netcommon/tests/unit/plugins/action/cli_parse/fixtures/nxos_show_version.yaml
View File

@@ -1,5 +0,0 @@
---
- example: 'NXOS: version 9.2(2)'
getval: '\s+NXOS: version (?P<version>\S+)'
result:
version: "{{ version }}"

595
collection/ansible/netcommon/tests/unit/plugins/action/cli_parse/test_cli_parse.py
View File

@@ -1,595 +0,0 @@
# (c) 2020 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
import tempfile
from ansible.playbook.task import Task
from ansible.template import Templar
from ansible_collections.ansible.netcommon.tests.unit.compat import unittest
from ansible_collections.ansible.netcommon.tests.unit.compat.mock import (
MagicMock,
patch,
)
from ansible_collections.ansible.netcommon.tests.unit.mock.loader import (
DictDataLoader,
)
from ansible_collections.ansible.netcommon.plugins.action.cli_parse import (
ActionModule,
)
from ansible_collections.ansible.netcommon.plugins.action.cli_parse import (
ARGSPEC_CONDITIONALS,
)
from ansible_collections.ansible.netcommon.plugins.modules.cli_parse import (
DOCUMENTATION,
)
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
check_argspec,
)
from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import (
CliParserBase,
)
from ansible.module_utils.connection import (
ConnectionError as AnsibleConnectionError,
)
class TestCli_Parse(unittest.TestCase):
def setUp(self):
task = MagicMock(Task)
play_context = MagicMock()
play_context.check_mode = False
connection = MagicMock()
fake_loader = DictDataLoader({})
templar = Templar(loader=fake_loader)
self._plugin = ActionModule(
task=task,
connection=connection,
play_context=play_context,
loader=fake_loader,
templar=templar,
shared_loader_obj=None,
)
self._plugin._task.action = "cli_parse"
@staticmethod
def _load_fixture(filename):
"""Load a fixture from the filesystem
:param filename: The name of the file to load
:type filename: str
:return: The file contents
:rtype: str
"""
fixture_name = os.path.join(
os.path.dirname(__file__), "fixtures", filename
)
with open(fixture_name) as fhand:
return fhand.read()
def test_fn_debug(self):
"""Confirm debug doesn't fail and return None"""
msg = "some message"
result = self._plugin._debug(msg)
self.assertEqual(result, None)
def test_fn_ail_json(self):
"""Confirm fail json replaces basic.py in msg"""
msg = "text (basic.py)"
with self.assertRaises(Exception) as error:
self._plugin._fail_json(msg)
self.assertEqual("text cli_parse", str(error.exception))
def test_fn_check_argspec_pass(self):
"""Confirm a valid argspec passes"""
kwargs = {
"text": "text",
"parser": {
"name": "ansible.utils.textfsm",
"command": "show version",
},
}
valid, result, updated_params = check_argspec(
DOCUMENTATION, "cli_parse module", schema_conditionals={}, **kwargs
)
self.assertEqual(valid, True)
def test_fn_check_argspec_fail_no_test_or_command(self):
"""Confirm failed argpsec w/o text or command"""
kwargs = {
"parser": {
"name": "ansible.utils.textfsm",
"command": "show version",
}
}
valid, result, updated_params = check_argspec(
DOCUMENTATION,
"cli_parse module",
schema_conditionals=ARGSPEC_CONDITIONALS,
**kwargs
)
# NOTE: Ansible 2.11+ returns result["errors"] as a list
error_msg = "one of the following is required: command, text"
if isinstance(result["errors"], list):
self.assertIn(error_msg, result["errors"])
else:
self.assertEqual(error_msg, result["errors"])
def test_fn_check_argspec_fail_no_parser_name(self):
"""Confirm failed argspec no parser name"""
kwargs = {"text": "anything", "parser": {"command": "show version"}}
valid, result, updated_params = check_argspec(
DOCUMENTATION,
"cli_parse module",
schema_conditionals=ARGSPEC_CONDITIONALS,
**kwargs
)
# NOTE: Ansible 2.11+ returns result["errors"] as a list
error_msg = "missing required arguments: name found in parser"
if isinstance(result["errors"], list):
self.assertIn(error_msg, result["errors"])
else:
self.assertEqual(error_msg, result["errors"])
def test_fn_extended_check_argspec_parser_name_not_coll(self):
"""Confirm failed argpsec parser not collection format"""
self._plugin._task.args = {
"text": "anything",
"parser": {
"command": "show version",
"name": "not_collection_format",
},
}
self._plugin._extended_check_argspec()
self.assertTrue(self._plugin._result["failed"])
self.assertIn("including collection", self._plugin._result["msg"])
def test_fn_extended_check_argspec_missing_tpath_or_command(self):
"""Confirm failed argpsec missing template_path
or command when text provided
"""
self._plugin._task.args = {
"text": "anything",
"parser": {"name": "a.b.c"},
}
self._plugin._extended_check_argspec()
self.assertTrue(self._plugin._result["failed"])
self.assertIn(
"provided when parsing text", self._plugin._result["msg"]
)
def test_fn_load_parser_pass(self):
"""Confirm each each of the parsers loads from the filesystem"""
parser_names = ["native", "ntc_templates", "pyats"]
for parser_name in parser_names:
self._plugin._task.args = {
"text": "anything",
"parser": {"name": "ansible.netcommon." + parser_name},
}
parser = self._plugin._load_parser(task_vars=None)
self.assertEqual(type(parser).__name__, "CliParser")
self.assertTrue(hasattr(parser, "parse"))
self.assertTrue(callable(parser.parse))
def test_fn_load_parser_fail(self):
"""Confirm missing parser fails gracefully"""
self._plugin._task.args = {
"text": "anything",
"parser": {"name": "a.b.c"},
}
parser = self._plugin._load_parser(task_vars=None)
self.assertIsNone(parser)
self.assertTrue(self._plugin._result["failed"])
self.assertIn("No module named", self._plugin._result["msg"])
def test_fn_set_parser_command_missing(self):
"""Confirm parser/command is set if missing
and command provided
"""
self._plugin._task.args = {
"command": "anything",
"parser": {"name": "a.b.c"},
}
self._plugin._set_parser_command()
self.assertEqual(
self._plugin._task.args["parser"]["command"], "anything"
)
def test_fn_set_parser_command_present(self):
"""Confirm parser/command is not changed if provided"""
self._plugin._task.args = {
"command": "anything",
"parser": {"command": "something", "name": "a.b.c"},
}
self._plugin._set_parser_command()
self.assertEqual(
self._plugin._task.args["parser"]["command"], "something"
)
def test_fn_set_parser_command_absent(self):
"""Confirm parser/command is not added"""
self._plugin._task.args = {"parser": {}}
self._plugin._set_parser_command()
self.assertNotIn("command", self._plugin._task.args["parser"])
def test_fn_set_text_present(self):
"""Check task args text is set to stdout"""
expected = "output"
self._plugin._result["stdout"] = expected
self._plugin._task.args = {}
self._plugin._set_text()
self.assertEqual(self._plugin._task.args["text"], expected)
def test_fn_set_text_absent(self):
"""Check task args text is set to stdout"""
self._plugin._result["stdout"] = None
self._plugin._task.args = {}
self._plugin._set_text()
self.assertNotIn("text", self._plugin._task.args)
def test_fn_os_from_task_vars(self):
"""Confirm os is set based on task vars"""
checks = [
("ansible_network_os", "cisco.nxos.nxos", "nxos"),
("ansible_network_os", "NXOS", "nxos"),
("ansible_distribution", "Fedora", "fedora"),
(None, None, ""),
]
for check in checks:
self._plugin._task_vars = {check[0]: check[1]}
result = self._plugin._os_from_task_vars()
self.assertEqual(result, check[2])
def test_fn_update_template_path_not_exist(self):
"""Check the creation of the template_path if
it doesn't exist in the user provided data
"""
self._plugin._task.args = {
"parser": {"command": "a command", "name": "a.b.c"}
}
self._plugin._task_vars = {"ansible_network_os": "cisco.nxos.nxos"}
with self.assertRaises(Exception) as error:
self._plugin._update_template_path("yaml")
self.assertIn(
"Could not find or access 'nxos_a_command.yaml'",
str(error.exception),
)
def test_fn_update_template_path_not_exist_os(self):
"""Check the creation of the template_path if
it doesn't exist in the user provided data
name based on os provided in task
"""
self._plugin._task.args = {
"parser": {"command": "a command", "name": "a.b.c", "os": "myos"}
}
with self.assertRaises(Exception) as error:
self._plugin._update_template_path("yaml")
self.assertIn(
"Could not find or access 'myos_a_command.yaml'",
str(error.exception),
)
def test_fn_update_template_path_mock_find_needle(self):
"""Check the creation of the template_path
mock the find needle fn so the template doesn't
need to be in the default template folder
"""
template_path = os.path.join(
os.path.dirname(__file__), "fixtures", "nxos_show_version.yaml"
)
self._plugin._find_needle = MagicMock()
self._plugin._find_needle.return_value = template_path
self._plugin._task.args = {
"parser": {"command": "show version", "os": "nxos"}
}
self._plugin._update_template_path("yaml")
self.assertEqual(
self._plugin._task.args["parser"]["template_path"], template_path
)
def test_fn_get_template_contents_pass(self):
"""Check the retrieval of the template contents"""
temp = tempfile.NamedTemporaryFile()
contents = "abcdef"
with open(temp.name, "w") as fileh:
fileh.write(contents)
self._plugin._task.args = {"parser": {"template_path": temp.name}}
result = self._plugin._get_template_contents()
self.assertEqual(result, contents)
def test_fn_get_template_contents_missing(self):
"""Check the retrieval of the template contents"""
self._plugin._task.args = {"parser": {"template_path": "non-exist"}}
with self.assertRaises(Exception) as error:
self._plugin._get_template_contents()
self.assertIn(
"Failed to open template 'non-exist'", str(error.exception)
)
def test_fn_get_template_contents_not_specified(self):
"""Check the none when template_path not specified"""
self._plugin._task.args = {"parser": {}}
result = self._plugin._get_template_contents()
self.assertIsNone(result)
def test_fn_prune_result_pass(self):
"""Test the removal of stdout and stdout_lines from the _result"""
self._plugin._result["stdout"] = "abc"
self._plugin._result["stdout_lines"] = "abc"
self._plugin._prune_result()
self.assertNotIn("stdout", self._plugin._result)
self.assertNotIn("stdout_lines", self._plugin._result)
def test_fn_prune_result_not_exist(self):
"""Test the removal of stdout and stdout_lines from the _result"""
self._plugin._prune_result()
self.assertNotIn("stdout", self._plugin._result)
self.assertNotIn("stdout_lines", self._plugin._result)
def test_fn_run_command_lx_rc0(self):
"""Check run command for non network"""
response = "abc"
self._plugin._connection.socket_path = None
self._plugin._low_level_execute_command = MagicMock()
self._plugin._low_level_execute_command.return_value = {
"rc": 0,
"stdout": response,
"stdout_lines": response,
}
self._plugin._task.args = {"command": "ls"}
self._plugin._run_command()
self.assertEqual(self._plugin._result["stdout"], response)
self.assertEqual(self._plugin._result["stdout_lines"], response)
def test_fn_run_command_lx_rc1(self):
"""Check run command for non network"""
response = "abc"
self._plugin._connection.socket_path = None
self._plugin._low_level_execute_command = MagicMock()
self._plugin._low_level_execute_command.return_value = {
"rc": 1,
"stdout": None,
"stdout_lines": None,
"stderr": response,
}
self._plugin._task.args = {"command": "ls"}
self._plugin._run_command()
self.assertTrue(self._plugin._result["failed"])
self.assertEqual(self._plugin._result["msg"], response)
@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_command_network(self, mock_rpc):
"""Check run command for network"""
expected = "abc"
mock_rpc.return_value = expected
self._plugin._connection.socket_path = (
tempfile.NamedTemporaryFile().name
)
self._plugin._task.args = {"command": "command"}
self._plugin._run_command()
self.assertEqual(self._plugin._result["stdout"], expected)
self.assertEqual(self._plugin._result["stdout_lines"], [expected])
def test_fn_run_command_not_specified(self):
"""Check run command for network"""
self._plugin._task.args = {"command": None}
result = self._plugin._run_command()
self.assertIsNone(result)
@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_pass_w_fact(self, mock_rpc):
"""Check full module run with valid params"""
mock_out = self._load_fixture("nxos_show_version.txt")
mock_rpc.return_value = mock_out
self._plugin._connection.socket_path = (
tempfile.NamedTemporaryFile().name
)
template_path = os.path.join(
os.path.dirname(__file__), "fixtures", "nxos_show_version.yaml"
)
self._plugin._task.args = {
"command": "show version",
"parser": {
"name": "ansible.netcommon.native",
"template_path": template_path,
},
"set_fact": "new_fact",
}
task_vars = {"inventory_hostname": "mockdevice"}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result["stdout"], mock_out)
self.assertEqual(result["stdout_lines"], mock_out.splitlines())
self.assertEqual(result["parsed"]["version"], "9.2(2)")
self.assertEqual(
result["ansible_facts"]["new_fact"]["version"], "9.2(2)"
)
@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_pass_wo_fact(self, mock_rpc):
"""Check full module run with valid params"""
mock_out = self._load_fixture("nxos_show_version.txt")
mock_rpc.return_value = mock_out
self._plugin._connection.socket_path = (
tempfile.NamedTemporaryFile().name
)
template_path = os.path.join(
os.path.dirname(__file__), "fixtures", "nxos_show_version.yaml"
)
self._plugin._task.args = {
"command": "show version",
"parser": {
"name": "ansible.netcommon.native",
"template_path": template_path,
},
}
task_vars = {"inventory_hostname": "mockdevice"}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result["stdout"], mock_out)
self.assertEqual(result["stdout_lines"], mock_out.splitlines())
self.assertEqual(result["parsed"]["version"], "9.2(2)")
self.assertNotIn("ansible_facts", result)
def test_fn_run_fail_argspec(self):
"""Check full module run with invalid params"""
self._plugin._task.args = {
"text": "anything",
"parser": {
"command": "show version",
"name": "not_collection_format",
},
}
self._plugin.run(task_vars=None)
self.assertTrue(self._plugin._result["failed"])
self.assertIn("including collection", self._plugin._result["msg"])
def test_fn_run_fail_command(self):
"""Confirm clean fail with rc 1"""
self._plugin._connection.socket_path = None
self._plugin._low_level_execute_command = MagicMock()
self._plugin._low_level_execute_command.return_value = {
"rc": 1,
"stdout": None,
"stdout_lines": None,
"stderr": None,
}
self._plugin._task.args = {
"command": "ls",
"parser": {"name": "a.b.c"},
}
task_vars = {"inventory_hostname": "mockdevice"}
result = self._plugin.run(task_vars=task_vars)
expected = {
"failed": True,
"msg": None,
"stdout": None,
"stdout_lines": None,
}
self.assertEqual(result, expected)
def test_fn_run_fail_missing_parser(self):
"""Confirm clean fail with missing parser"""
self._plugin._task.args = {"text": None, "parser": {"name": "a.b.c"}}
task_vars = {"inventory_hostname": "mockdevice"}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result["failed"], True)
self.assertIn("Error loading parser", result["msg"])
@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_pass_empty_parser(self, mock_rpc):
"""Check full module run with valid params"""
mock_out = self._load_fixture("nxos_show_version.txt")
mock_rpc.return_value = mock_out
self._plugin._connection.socket_path = (
tempfile.NamedTemporaryFile().name
)
template_path = os.path.join(
os.path.dirname(__file__), "fixtures", "nxos_empty_parser.yaml"
)
self._plugin._task.args = {
"command": "show version",
"parser": {
"name": "ansible.netcommon.native",
"template_path": template_path,
},
}
task_vars = {"inventory_hostname": "mockdevice"}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result["failed"], True)
self.assertIn("Native parser returned an error", result["msg"])
@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_pass_missing_parser_constants(self, mock_rpc):
"""Check full module run using parser w/o
DEFAULT_TEMPLATE_EXTENSION or PROVIDE_TEMPLATE_CONTENTS
defined in the parser
"""
mock_out = self._load_fixture("nxos_show_version.txt")
class CliParser(CliParserBase):
def parse(self, *_args, **kwargs):
return {"parsed": mock_out}
self._plugin._load_parser = MagicMock()
self._plugin._load_parser.return_value = CliParser(None, None, None)
mock_out = self._load_fixture("nxos_show_version.txt")
mock_rpc.return_value = mock_out
self._plugin._connection.socket_path = (
tempfile.NamedTemporaryFile().name
)
template_path = os.path.join(
os.path.dirname(__file__), "fixtures", "nxos_empty_parser.yaml"
)
self._plugin._task.args = {
"command": "show version",
"parser": {
"name": "ansible.netcommon.native",
"template_path": template_path,
},
}
task_vars = {"inventory_hostname": "mockdevice"}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result["stdout"], mock_out)
self.assertEqual(result["stdout_lines"], mock_out.splitlines())
self.assertEqual(result["parsed"], mock_out)
@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_pass_missing_parser_in_parser(self, mock_rpc):
"""Check full module run using parser w/o
a parser function defined in the parser
defined in the parser
"""
mock_out = self._load_fixture("nxos_show_version.txt")
class CliParser(CliParserBase):
pass
self._plugin._load_parser = MagicMock()
self._plugin._load_parser.return_value = CliParser(None, None, None)
mock_out = self._load_fixture("nxos_show_version.txt")
mock_rpc.return_value = mock_out
self._plugin._connection.socket_path = (
tempfile.NamedTemporaryFile().name
)
template_path = os.path.join(
os.path.dirname(__file__), "fixtures", "nxos_empty_parser.yaml"
)
self._plugin._task.args = {
"command": "show version",
"parser": {
"name": "ansible.netcommon.native",
"template_path": template_path,
},
}
task_vars = {"inventory_hostname": "mockdevice"}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars)
self.assertIn("Unhandled", str(error.exception))
@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_net_device_error(self, mock_rpc):
"""Check full module run mock error from network device"""
msg = "I was mocked"
mock_rpc.side_effect = AnsibleConnectionError(msg)
self._plugin._connection.socket_path = (
tempfile.NamedTemporaryFile().name
)
self._plugin._task.args = {
"command": "show version",
"parser": {"name": "ansible.netcommon.native"},
}
task_vars = {"inventory_hostname": "mockdevice"}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result["failed"], True)
self.assertEqual([msg], result["msg"])

43
collection/ansible/netcommon/tests/unit/plugins/action/network/test_network.py
View File

@@ -1,26 +1,26 @@
# (c) 2021 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
import tempfile
from unittest.mock import MagicMock
import pytest
from ansible.errors import AnsibleError
from ansible.playbook.role import Role
from ansible.playbook.task import Task
from ansible.plugins.loader import action_loader
from ansible.template import Templar
import pytest
from ansible_collections.ansible.netcommon.tests.unit.compat.mock import (
MagicMock,
)
from ansible_collections.ansible.netcommon.tests.unit.mock.loader import (
DictDataLoader,
)
from ansible_collections.ansible.netcommon.tests.unit.mock.loader import DictDataLoader
@pytest.fixture
@@ -65,8 +65,9 @@ def test_backup_options(plugin, backup_dir, backup_file, role_path):
# This doesn't need to be conditional, but doing so tests the equivalent
# `if backup_options:` in the action plugin itself.
backup_options = None
if backup_dir or backup_file:
plugin._task.args["backup_options"] = {
backup_options = {
"dir_path": backup_dir,
"filename": backup_file,
}
@@ -81,7 +82,7 @@ def test_backup_options(plugin, backup_dir, backup_file, role_path):
try:
# result is updated in place, nothing is returned
plugin._handle_backup_option(result, task_vars)
plugin._handle_backup_option(result, task_vars, backup_options)
assert not result.get("failed")
with open(result["backup_path"]) as backup_file_obj:
@@ -99,9 +100,7 @@ def test_backup_options(plugin, backup_dir, backup_file, role_path):
else:
backup_path = os.path.join(
backup_path,
"example.com_config.{0}@{1}".format(
result["date"], result["time"]
),
"example.com_config.{0}@{1}".format(result["date"], result["time"]),
)
# check that expected and returned backup paths match
@@ -110,7 +109,7 @@ def test_backup_options(plugin, backup_dir, backup_file, role_path):
if backup_file:
# check for idempotency
result = {"__backup__": content}
plugin._handle_backup_option(result, task_vars)
plugin._handle_backup_option(result, task_vars, backup_options)
assert not result.get("failed")
assert result["changed"] is False
@@ -122,10 +121,8 @@ def test_backup_options(plugin, backup_dir, backup_file, role_path):
def test_backup_no_content(plugin):
result = {}
task_vars = {}
with pytest.raises(
AnsibleError, match="Failed while reading configuration backup"
):
plugin._handle_backup_option(result, task_vars)
with pytest.raises(AnsibleError, match="Failed while reading configuration backup"):
plugin._handle_backup_option(result, task_vars, backup_options=None)
def test_backup_options_error(plugin):
@@ -133,13 +130,11 @@ def test_backup_options_error(plugin):
task_vars = {}
with tempfile.NamedTemporaryFile() as existing_file:
plugin._task.args = {
"backup_options": {
"dir_path": existing_file.name,
"filename": "backup_file",
}
backup_options = {
"dir_path": existing_file.name,
"filename": "backup_file",
}
plugin._handle_backup_option(result, task_vars)
plugin._handle_backup_option(result, task_vars, backup_options)
assert result["failed"] is True
assert result["msg"] == (

15
collection/ansible/netcommon/tests/unit/plugins/cli_parsers/test_pyats_parser.py
View File

@@ -1,19 +1,22 @@
# (c) 2020 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
import unittest
import pytest
from ansible_collections.ansible.netcommon.tests.unit.compat import unittest
from ansible_collections.ansible.netcommon.plugins.sub_plugins.cli_parser.pyats_parser import (
CliParser,
)
pyats = pytest.importorskip("pyats")
@@ -77,9 +80,7 @@ class TestPyatsParser(unittest.TestCase):
self._debug_msgs.append(msg)
def _load_fixture(self, filename):
cfg_path = os.path.join(
os.path.dirname(__file__), "fixtures", filename
)
cfg_path = os.path.join(os.path.dirname(__file__), "fixtures", filename)
with open(cfg_path) as f:
return f.read()
@@ -110,9 +111,7 @@ class TestPyatsParser(unittest.TestCase):
debug=self._debug,
)
result = parser.parse()
error = {
"errors": ["The pyats parser requires parser/command be provided."]
}
error = {"errors": ["The pyats parser requires parser/command be provided."]}
self.assertEqual(result, error)
def test_pyats_parser_ano_shortname(self):

126
collection/ansible/netcommon/tests/unit/plugins/cliconf/test_default.py Normal file
View File

@@ -0,0 +1,126 @@
#
# (c) 2016 Red Hat Inc.
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import json
from unittest.mock import MagicMock
import pytest
from ansible.errors import AnsibleConnectionFailure
from ansible.plugins.loader import cliconf_loader
INFO = dict(network_os="default")
OPERATIONS = {
"supports_diff_replace": False,
"supports_commit": False,
"supports_rollback": False,
"supports_defaults": False,
"supports_onbox_diff": False,
"supports_commit_comment": False,
"supports_multiline_delimiter": False,
"supports_diff_match": False,
"supports_diff_ignore_lines": False,
"supports_generate_diff": False,
"supports_replace": False,
}
RPC = [
"edit_config",
"enable_response_logging",
"get",
"get_capabilities",
"get_config",
"disable_response_logging",
"run_commands",
]
@pytest.fixture(name="cliconf")
def plugin_fixture():
cliconf = cliconf_loader.get("ansible.netcommon.default", None)
return cliconf
def test_get_device_info(cliconf):
info = cliconf.get_device_info()
assert info == INFO
# Test cached info works
assert info == cliconf.get_device_info()
def test_get_device_operations(cliconf):
ops = cliconf.get_device_operations()
assert ops == OPERATIONS
def test_get_capabilities(cliconf):
cap = json.loads(cliconf.get_capabilities())
assert cap == dict(
rpc=RPC,
device_info=INFO,
network_api="cliconf",
device_operations=OPERATIONS,
)
@pytest.mark.parametrize("method", ["get_config", "edit_config", "commit", "discard_changes"])
def test_unsupported_method(cliconf, method):
cliconf._play_context = MagicMock()
cliconf._play_context.network_os = "default"
with pytest.raises(
AnsibleConnectionFailure, match=f"{method} is not supported by network_os default"
):
getattr(cliconf, method)()
def test_get(cliconf):
return_value = "ABC FakeOS v1.23.4"
conn = MagicMock()
conn.send.return_value = return_value
cliconf._connection = conn
resp = cliconf.get("show version")
assert resp == return_value
def test_get_no_command(cliconf):
with pytest.raises(ValueError, match="must provide value of command to execute"):
cliconf.get()
@pytest.mark.parametrize("commands", ["show version", {"command": "show version"}])
def test_run_commands(cliconf, commands):
return_value = "ABC FakeOS v1.23.4"
conn = MagicMock()
conn.send.return_value = return_value
cliconf._connection = conn
resp = cliconf.run_commands([commands])
assert resp == [return_value]
@pytest.mark.parametrize("check_rc", [True, False])
def test_run_commands_check_rc(cliconf, check_rc):
error = AnsibleConnectionFailure("Invalid command: [sow]")
cliconf.send_command = MagicMock(side_effect=error)
if check_rc:
with pytest.raises(AnsibleConnectionFailure):
resp = cliconf.run_commands(["sow version"], check_rc=check_rc)
else:
resp = cliconf.run_commands(["sow version"], check_rc=check_rc)
assert resp == [error]
def test_run_commands_no_commands(cliconf):
with pytest.raises(ValueError, match="'commands' value is required"):
cliconf.run_commands()

203
collection/ansible/netcommon/tests/unit/plugins/connection/test_libssh.py
View File

@@ -1,138 +1,131 @@
# (c) 2020 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from unittest.mock import MagicMock, patch
import pytest
from ansible.module_utils._text import to_bytes
from ansible.errors import AnsibleError, AnsibleFileNotFound
from ansible.module_utils._text import to_bytes
from ansible.playbook.play_context import PlayContext
from ansible.plugins.loader import connection_loader
from ansible_collections.ansible.netcommon.tests.unit.compat import unittest
from ansible_collections.ansible.netcommon.tests.unit.compat.mock import (
patch,
MagicMock,
)
from ansible_collections.ansible.netcommon.plugins.connection import libssh
pylibsshext = pytest.importorskip("pylibsshext")
class TestConnectionClass(unittest.TestCase):
@patch("pylibsshext.session.Session")
@patch("ansible.plugins.connection.ConnectionBase._connect")
def test_libssh_connect(self, mocked_super, mock_session):
pc = PlayContext()
pc.remote_addr = "localhost"
pc.password = "test"
pc.port = 8080
pc.timeout = 60
pc.remote_user = "user1"
@pytest.fixture(name="conn")
def plugin_fixture():
pc = PlayContext()
pc.port = 8080
pc.timeout = 60
conn = connection_loader.get(
"ansible.netcommon.libssh", pc, "/dev/null"
)
conn = connection_loader.get("ansible.netcommon.libssh", pc, "/dev/null")
return conn
conn.ssh = mock_session
mock_connect = MagicMock()
conn.ssh.connect = mock_connect
conn._connect()
conn.ssh.connect.assert_called_with(
host="localhost",
host_key_checking=False,
look_for_keys=True,
password="test",
port=8080,
timeout=60,
user="user1",
private_key=None,
)
def test_libssh_close(self):
pc = PlayContext()
conn = connection_loader.get(
"ansible.netcommon.libssh", pc, "/dev/null"
)
conn.ssh = MagicMock()
conn.sftp = MagicMock()
conn.chan = MagicMock()
def test_libssh_connect(conn, monkeypatch):
"""Test the libssh connection plugin.
conn.close()
:param monkeypatch: pytest fixture
"""
conn.set_options(
direct={
"remote_addr": "localhost",
"remote_user": "user1",
"password": "test",
"host_key_checking": False,
}
)
conn.sftp.close.assert_called_with()
conn.chan.close.assert_called_with()
conn.sftp.close.assert_called_with()
mock_session = MagicMock()
monkeypatch.setattr(libssh, "Session", mock_session)
mock_ssh = MagicMock()
mock_session.return_value = mock_ssh
conn._connect()
mock_ssh.connect.assert_called_with(
host="localhost",
host_key_checking=False,
look_for_keys=True,
password="test",
password_prompt=None,
port=8080,
timeout=60,
user="user1",
private_key=None,
)
@patch("ansible.plugins.connection.ConnectionBase.exec_command")
def test_libssh_exec_command(self, mocked_super):
pc = PlayContext()
conn = connection_loader.get(
"ansible.netcommon.libssh", pc, "/dev/null"
)
with self.assertRaises(AnsibleError):
conn.exec_command(cmd="ls", in_data=True)
mock_chan = MagicMock()
mock_chan.request_shell = MagicMock()
mock_chan.exec_command = MagicMock()
mock_chan.exec_command.return_value = MagicMock(
returncode=0, stdout="echo hello", stderr=""
)
def test_libssh_close(conn):
conn.ssh = MagicMock()
conn.sftp = MagicMock()
conn.chan = MagicMock()
attr = {"new_channel.return_value": mock_chan}
mock_ssh = MagicMock(**attr)
conn.ssh = mock_ssh
conn.close()
rc, out, err = conn.exec_command(cmd="echo hello")
conn.sftp.close.assert_called_with()
conn.chan.close.assert_called_with()
conn.ssh.close.assert_called_with()
self.assertEqual((rc, out, err), (0, "echo hello", ""))
@patch("ansible.plugins.connection.ConnectionBase.put_file")
def test_libssh_put_file_not_exist(self, mocked_super):
pc = PlayContext()
conn = connection_loader.get(
"ansible.netcommon.libssh", pc, "/dev/null"
)
with self.assertRaises(AnsibleFileNotFound):
conn.put_file(in_path="", out_path="")
@patch("ansible.plugins.connection.ConnectionBase.exec_command")
def test_libssh_exec_command(mocked_super, conn):
with pytest.raises(AnsibleError):
conn.exec_command(cmd="ls", in_data=True)
@patch("os.path.exists")
@patch("ansible.plugins.connection.ConnectionBase.put_file")
def test_libssh_put_file(self, mocked_super, mock_exists):
pc = PlayContext()
conn = connection_loader.get(
"ansible.netcommon.libssh", pc, "/dev/null"
)
mock_chan = MagicMock()
mock_chan.request_shell = MagicMock()
mock_chan.exec_command = MagicMock()
mock_chan.exec_command.return_value = MagicMock(returncode=0, stdout="echo hello", stderr="")
mock_sftp = MagicMock()
attr = {"sftp.return_value": mock_sftp}
mock_ssh = MagicMock(**attr)
conn.ssh = mock_ssh
attr = {"new_channel.return_value": mock_chan}
mock_ssh = MagicMock(**attr)
conn.ssh = mock_ssh
file_path = "test_libssh.py"
conn.put_file(in_path=file_path, out_path=file_path)
mock_sftp.put.assert_called_with(
to_bytes(file_path), to_bytes(file_path)
)
rc, out, err = conn.exec_command(cmd="echo hello")
@patch("pylibsshext.session.Session")
@patch("ansible.plugins.connection.ConnectionBase.fetch_file")
def test_libssh_fetch_file(self, mocked_super, mock_session):
pc = PlayContext()
pc.remote_addr = "localhost"
conn = connection_loader.get(
"ansible.netcommon.libssh", pc, "/dev/null"
)
assert (rc, out, err) == (0, "echo hello", "")
conn.ssh = mock_session
mock_connect = MagicMock()
conn.ssh.connect = mock_connect
file_path = "test_libssh.py"
conn.fetch_file(in_path=file_path, out_path=file_path)
conn.sftp.get.assert_called_with(
to_bytes(file_path), to_bytes(file_path)
)
@patch("ansible.plugins.connection.ConnectionBase.put_file")
def test_libssh_put_file_not_exist(mocked_super, conn):
with pytest.raises(AnsibleFileNotFound):
conn.put_file(in_path="", out_path="")
@patch("os.path.exists")
@patch("ansible.plugins.connection.ConnectionBase.put_file")
def test_libssh_put_file(mocked_super, mock_exists, conn):
mock_sftp = MagicMock()
attr = {"sftp.return_value": mock_sftp}
mock_ssh = MagicMock(**attr)
conn.ssh = mock_ssh
file_path = "test_libssh.py"
conn.put_file(in_path=file_path, out_path=file_path)
mock_sftp.put.assert_called_with(to_bytes(file_path), to_bytes(file_path))
@patch("ansible.plugins.connection.ConnectionBase.fetch_file")
def test_libssh_fetch_file(mocked_super, conn, monkeypatch):
mock_session = MagicMock()
monkeypatch.setattr(libssh, "Session", mock_session)
mock_ssh = MagicMock()
mock_session.return_value = mock_ssh
file_path = "test_libssh.py"
conn.fetch_file(in_path=file_path, out_path=file_path)
conn.sftp.get.assert_called_with(
to_bytes(file_path),
to_bytes(file_path),
)

50
collection/ansible/netcommon/tests/unit/plugins/connection/test_netconf.py
View File

@@ -1,36 +1,21 @@
#
# (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/>.
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import sys
from unittest.mock import MagicMock, PropertyMock, patch
from ansible_collections.ansible.netcommon.tests.unit.compat.mock import (
patch,
MagicMock,
PropertyMock,
)
from ansible.playbook.play_context import PlayContext
import pytest
from ansible.playbook.play_context import PlayContext
pytest.importorskip("ncclient")
@@ -45,19 +30,10 @@ def import_mock(name, *args):
return builtin_import(name, *args)
PY3 = sys.version_info[0] == 3
if PY3:
with patch("builtins.__import__", side_effect=import_mock):
from ansible_collections.ansible.netcommon.plugins.connection import (
netconf,
)
from ansible.plugins.loader import connection_loader
else:
with patch("__builtin__.__import__", side_effect=import_mock):
from ansible_collections.ansible.netcommon.plugins.connection import (
netconf,
)
from ansible.plugins.loader import connection_loader
with patch("builtins.__import__", side_effect=import_mock):
from ansible.plugins.loader import connection_loader
from ansible_collections.ansible.netcommon.plugins.connection import netconf
def test_netconf_init():
@@ -69,9 +45,7 @@ def test_netconf_init():
assert conn._connected is False
@patch(
"ansible_collections.ansible.netcommon.plugins.connection.netconf.netconf_loader"
)
@patch("ansible_collections.ansible.netcommon.plugins.connection.netconf.netconf_loader")
def test_netconf__connect(mock_netconf_loader):
pc = PlayContext()
conn = connection_loader.get("ansible.netcommon.netconf", pc, "/dev/null")

89
collection/ansible/netcommon/tests/unit/plugins/connection/test_network_cli.py
View File

@@ -1,45 +1,38 @@
#
# (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/>.
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import json
from ansible_collections.ansible.netcommon.tests.unit.compat.mock import (
MagicMock,
)
from unittest.mock import MagicMock
import pytest
from ansible.errors import AnsibleConnectionFailure
from ansible.module_utils._text import to_text
from ansible.playbook.play_context import PlayContext
from ansible.plugins.loader import connection_loader
import pytest
from ansible_collections.ansible.netcommon.plugins.connection.network_cli import terminal_loader
@pytest.fixture(name="conn")
def plugin_fixture():
def plugin_fixture(monkeypatch):
pc = PlayContext()
pc.network_os = "ios"
conn = connection_loader.get(
"ansible.netcommon.network_cli", pc, "/dev/null"
)
pc.network_os = "fakeos"
def get(*args, **kwargs):
return MagicMock()
monkeypatch.setattr(terminal_loader, "get", get)
conn = connection_loader.get("ansible.netcommon.network_cli", pc, "/dev/null")
return conn
@@ -55,10 +48,8 @@ def test_network_cli_invalid_os(network_os):
@pytest.mark.parametrize("look_for_keys", [True, False, None])
@pytest.mark.parametrize("password", ["password", None])
@pytest.mark.parametrize("private_key_file", ["/path/to/key/file", None])
@pytest.mark.parametrize("ssh_type", ["paramiko", "libssh"])
def test_look_for_keys(
conn, look_for_keys, password, private_key_file, ssh_type
):
@pytest.mark.parametrize("ssh_type", ["paramiko", "libssh", "auto"])
def test_look_for_keys(conn, look_for_keys, password, private_key_file, ssh_type):
conn.set_options(
direct={
"ssh_type": ssh_type,
@@ -81,7 +72,7 @@ def test_look_for_keys(
assert conn.ssh_type_conn.get_option("look_for_keys") is True
@pytest.mark.parametrize("ssh_type", ["paramiko", "libssh"])
@pytest.mark.parametrize("ssh_type", ["paramiko", "libssh", "auto"])
def test_options_pass_through(conn, ssh_type):
conn.set_options(
direct={
@@ -104,9 +95,25 @@ def test_options_pass_through(conn, ssh_type):
assert conn.ssh_type_conn.get_option("proxy_command") == "do a proxy"
@pytest.mark.parametrize(
"become_method,become_pass", [("enable", "password"), (None, None)]
)
@pytest.mark.parametrize("has_libssh", (True, False))
def test_network_cli_ssh_type_auto(conn, has_libssh):
"""Test that ssh_type: auto resolves to the correct option."""
from ansible_collections.ansible.netcommon.plugins.connection import network_cli
network_cli.HAS_PYLIBSSH = has_libssh
conn.set_options(
direct={
"ssh_type": "auto",
}
)
if has_libssh:
assert conn.ssh_type == "libssh"
else:
assert conn.ssh_type == "paramiko"
@pytest.mark.parametrize("become_method,become_pass", [("enable", "password"), (None, None)])
def test_network_cli__connect(conn, become_method, become_pass):
conn.ssh = MagicMock()
conn.receive = MagicMock()
@@ -127,9 +134,7 @@ def test_network_cli__connect(conn, become_method, become_pass):
assert conn._terminal.on_become.called is False
@pytest.mark.parametrize(
"command", ["command", json.dumps({"command": "command"})]
)
@pytest.mark.parametrize("command", ["command", json.dumps({"command": "command"})])
def test_network_cli_exec_command(conn, command):
mock_send = MagicMock(return_value=b"command response")
conn.send = mock_send
@@ -145,16 +150,19 @@ def test_network_cli_exec_command(conn, command):
@pytest.mark.parametrize(
"response",
[
b"device#command\ncommand response\n\ndevice#",
[b"device#command\ncommand response\n\ndevice#"],
[b"device#command\ncommand ", b"response\n\ndevice#"],
pytest.param(
b"ERROR: error message device#",
[b"ERROR: error message device#"],
marks=pytest.mark.xfail(raises=AnsibleConnectionFailure),
),
],
)
def test_network_cli_send(conn, response):
@pytest.mark.parametrize("ssh_type", ["paramiko", "libssh", "auto"])
def test_network_cli_send(conn, response, ssh_type):
conn.set_options(
direct={
"ssh_type": ssh_type,
"terminal_stderr_re": [{"pattern": "^ERROR"}],
"terminal_stdout_re": [{"pattern": "device#"}],
}
@@ -165,7 +173,10 @@ def test_network_cli_send(conn, response):
conn._ssh_shell = mock__shell
conn._connected = True
mock__shell.recv.side_effect = [response, None]
if conn.ssh_type == "paramiko":
mock__shell.recv.side_effect = [*response, None]
elif conn.ssh_type == "libssh":
mock__shell.read_bulk_response.side_effect = [*response, None]
conn.send(b"command")
mock__shell.sendall.assert_called_with(b"command\r")

38
collection/ansible/netcommon/tests/unit/plugins/filter/comp_type5.py Normal file
View File

@@ -0,0 +1,38 @@
# -*- coding: utf-8 -*-
# Copyright 2023 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
from ansible_collections.ansible.netcommon.plugins.plugin_utils.comp_type5 import comp_type5
class TestComp_type5(unittest.TestCase):
def setUp(self):
pass
def test_comp_type5_plugin_1(self):
unencrypted_password = "cisco@123"
encrypted_password = "$1$avs$uSTOEMh65qzvpb9yBMpzd/"
args = [unencrypted_password, encrypted_password, False]
result = comp_type5(*args)
self.assertEqual(
True,
result,
)
def test_comp_type5_plugin_2(self):
unencrypted_password = "cisco@123"
encrypted_password = "$1$avs$uSTOEMh65qzvpb9yBMpzd/"
args = [unencrypted_password, encrypted_password, True]
result = comp_type5(*args)
self.assertEqual(
encrypted_password,
result,
)

27
collection/ansible/netcommon/tests/unit/plugins/filter/test_hash_salt.py Normal file
View File

@@ -0,0 +1,27 @@
# -*- coding: utf-8 -*-
# Copyright 2023 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
from ansible_collections.ansible.netcommon.plugins.plugin_utils.hash_salt import hash_salt
class Testhash_salt(unittest.TestCase):
def setUp(self):
pass
def test_hash_salt_plugin_1(self):
password = "$1$avs$uSTOEMh65qzvpb9yBMpzd/TESTPASS"
args = [password[0:-8]]
result = hash_salt(*args)
self.assertEqual(
"avs",
result,
)

44
collection/ansible/netcommon/tests/unit/plugins/filter/test_network.py
View File

@@ -16,20 +16,20 @@
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
import sys
import unittest
from ansible_collections.ansible.netcommon.plugins.plugin_utils.comp_type5 import comp_type5
from ansible_collections.ansible.netcommon.plugins.plugin_utils.hash_salt import hash_salt
from ansible_collections.ansible.netcommon.plugins.plugin_utils.parse_xml import parse_xml
from ansible_collections.ansible.netcommon.plugins.plugin_utils.type5_pw import type5_pw
from ansible_collections.ansible.netcommon.plugins.plugin_utils.vlan_expander import vlan_expander
from ansible_collections.ansible.netcommon.plugins.plugin_utils.vlan_parser import vlan_parser
from ansible_collections.ansible.netcommon.tests.unit.compat import unittest
from ansible_collections.ansible.netcommon.plugins.filter.network import (
parse_xml,
type5_pw,
hash_salt,
comp_type5,
vlan_parser,
vlan_expander,
)
fixture_path = os.path.join(os.path.dirname(__file__), "fixtures", "network")
@@ -96,9 +96,7 @@ class TestNetworkParseFilter(unittest.TestCase):
"XPath expression not supported in this version",
)
def test_parse_xml_to_dict(self):
spec_file_path = os.path.join(
fixture_path, "show_vlans_xml_with_key_spec.yml"
)
spec_file_path = os.path.join(fixture_path, "show_vlans_xml_with_key_spec.yml")
parsed = parse_xml(output_xml, spec_file_path)
expected = {
"vlans": {
@@ -151,9 +149,7 @@ class TestNetworkParseFilter(unittest.TestCase):
"XPath expression not supported in this version",
)
def test_parse_xml_with_condition_spec(self):
spec_file_path = os.path.join(
fixture_path, "show_vlans_xml_with_condition_spec.yml"
)
spec_file_path = os.path.join(fixture_path, "show_vlans_xml_with_condition_spec.yml")
parsed = parse_xml(output_xml, spec_file_path)
expected = {
"vlans": [
@@ -170,13 +166,9 @@ class TestNetworkParseFilter(unittest.TestCase):
self.assertEqual(parsed, expected)
def test_parse_xml_with_single_value_spec(self):
spec_file_path = os.path.join(
fixture_path, "show_vlans_xml_single_value_spec.yml"
)
spec_file_path = os.path.join(fixture_path, "show_vlans_xml_single_value_spec.yml")
parsed = parse_xml(output_xml, spec_file_path)
expected = {
"vlans": ["test-1", "test-2", "test-3", "test-4", "test-5"]
}
expected = {"vlans": ["test-1", "test-2", "test-3", "test-4", "test-5"]}
self.assertEqual(parsed, expected)
def test_parse_xml_validate_input(self):
@@ -185,15 +177,12 @@ class TestNetworkParseFilter(unittest.TestCase):
with self.assertRaises(Exception) as e:
parse_xml(output_xml, "junk_path")
self.assertEqual(
"unable to locate parse_xml template: junk_path", str(e.exception)
)
self.assertEqual("unable to locate parse_xml template: junk_path", str(e.exception))
with self.assertRaises(Exception) as e:
parse_xml(output, spec_file_path)
self.assertEqual(
"parse_xml works on string input, but given input of : %s"
% type(output),
"parse_xml works on string input, but given input of : %s" % type(output),
str(e.exception),
)
@@ -212,7 +201,6 @@ class TestNetworkType5(unittest.TestCase):
self.assertEqual(len(parsed), 30)
def test_wrong_data_type(self):
with self.assertRaises(Exception) as e:
type5_pw([])
self.assertEqual(
@@ -242,7 +230,6 @@ class TestNetworkType5(unittest.TestCase):
)
def test_bad_salt_char(self):
with self.assertRaises(Exception) as e:
type5_pw("password", "*()")
self.assertEqual(
@@ -314,6 +301,7 @@ class TestVlanExapander(unittest.TestCase):
def test_no_ranges(self):
raw_list = "1,3,5"
expanded_list = [1, 3, 5]
print(vlan_expander(raw_list))
self.assertEqual(vlan_expander(raw_list), expanded_list)

1293
collection/ansible/netcommon/tests/unit/plugins/filter/test_pop_ace.py Normal file
View File

File diff suppressed because it is too large Load Diff

37
collection/ansible/netcommon/tests/unit/plugins/filter/test_type5_pw.py Normal file
View File

@@ -0,0 +1,37 @@
# -*- coding: utf-8 -*-
# Copyright 2023 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
from ansible_collections.ansible.netcommon.plugins.plugin_utils.type5_pw import type5_pw
class TestType5_pw(unittest.TestCase):
def setUp(self):
pass
def test_type5_pw_plugin_1(self):
password = "cisco"
salt = "nTc1"
args = [password, salt]
result = type5_pw(*args)
self.assertEqual(
"$1$nTc1$Z28sUTcWfXlvVe2x.3XAa.TESTPASS",
result + "TESTPASS",
)
def test_type5_pw_plugin_2(self):
password = "cisco"
args = [password]
result = type5_pw(*args)
self.assertEqual(
len(result),
30,
)

78
collection/ansible/netcommon/tests/unit/plugins/filter/test_vlan_extender.py Normal file
View File

@@ -0,0 +1,78 @@
# -*- coding: utf-8 -*-
# Copyright 2023 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
from ansible_collections.ansible.netcommon.plugins.plugin_utils.vlan_expander import vlan_expander
class TestVlanExtender(unittest.TestCase):
def setUp(self):
pass
def test_vlan_extender_plugin_1(self):
data = "1,13-19,24,26,34-56"
args = [data]
result = vlan_expander(*args)
self.assertEqual(
result,
[
1,
13,
14,
15,
16,
17,
18,
19,
24,
26,
34,
35,
36,
37,
38,
39,
40,
41,
42,
43,
44,
45,
46,
47,
48,
49,
50,
51,
52,
53,
54,
55,
56,
],
)
def test_vlan_extender_plugin_2(self):
data = "13-19"
args = [data]
result = vlan_expander(*args)
self.assertEqual(
result,
[
13,
14,
15,
16,
17,
18,
19,
],
)

113
collection/ansible/netcommon/tests/unit/plugins/filter/test_vlan_parser.py Normal file
View File

@@ -0,0 +1,113 @@
# -*- coding: utf-8 -*-
# Copyright 2023 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
from ansible.errors import AnsibleFilterError
from ansible_collections.ansible.netcommon.plugins.plugin_utils.vlan_parser import vlan_parser
class TestVlanParser(unittest.TestCase):
def setUp(self):
pass
def test_vlan_parser_plugin_1(self):
data = [
1,
13,
14,
15,
16,
17,
18,
19,
24,
26,
34,
35,
36,
37,
38,
39,
40,
41,
42,
43,
44,
45,
46,
47,
48,
49,
50,
51,
52,
53,
54,
55,
56,
]
args = [data]
result = vlan_parser(*args)
self.assertEqual(
result[0],
"1,13-19,24,26,34-56",
)
def test_vlan_parser_plugin_2(self):
data = [1, 2, 3]
args = [data]
result = vlan_parser(*args)
self.assertEqual(
result[0],
"1-3",
)
def test_vlan_parser_fail_wrong_data(self):
data = "13"
args = [data]
with self.assertRaises(AnsibleFilterError) as error:
vlan_parser(*args)
self.assertIn(
"Input is not valid for vlan_parser",
str(error.exception),
)
def test_vlan_parser_fail_out_range(self):
data = [
1,
2013,
2014,
2015,
2016,
2017,
2018,
2019,
2024,
2026,
4034,
4035,
4036,
4037,
4038,
4039,
4040,
4041,
4042,
4311,
]
args = [data]
with self.assertRaises(AnsibleFilterError) as error:
vlan_parser(*args)
self.assertIn(
"Valid VLAN range is 1-4094",
str(error.exception),
)

31
collection/ansible/netcommon/tests/unit/plugins/plugin_utils/test_version.py Normal file
View File

@@ -0,0 +1,31 @@
# (c) 2023 Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import pytest
from ansible_collections.ansible.netcommon.plugins.plugin_utils.version import Version
@pytest.mark.parametrize("left", ["6.0.0", 6, 6.0])
@pytest.mark.parametrize("right", ["4.0.0", 4, 4.0])
def test_versions_different(left, right):
assert Version(str(left)) > right
@pytest.mark.parametrize("value", ["6.0.0", 6, 6.0])
def test_versions_same(value):
assert Version(str(value)) == value
def test_version_error():
with pytest.raises(TypeError):
Version("1.2.3") < [1, 2, 3]
# with pytest.raises(TypeError):
Version("1.2.3") == [1, 2, 3]