Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add unit test showing high CPU utilization when attempting to get /120 subnet #137

Closed
wants to merge 6 commits into from
Closed

Add unit test showing high CPU utilization when attempting to get /120 subnet #137

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ed98b1e
fix typo. Add example for IPV6
sebastien-rosset Jan 24, 2022
6bec94e
add unit tests for IPv6
sebastien-rosset Jan 24, 2022
8309327
move unit test with /120 case to a separate PR
sebastien-rosset Jan 31, 2022
acf52be
fix linter issues, add changelog
sebastien-rosset Feb 1, 2022
c6acd5a
add unit test with /120 subnet
sebastien-rosset Feb 1, 2022
f016252
reformat with black
sebastien-rosset Feb 6, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions changelogs/fragments/add_ipfilter_doc_ut.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
minor_changes:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
minor_changes:
doc_changes:

- Add example for IPv6 in ipsubnet filter documentation.
- Fix typo in ipsubnet documentation.
- Add unit tests for ipsubnet filter.
14 changes: 13 additions & 1 deletion plugins/filter/ipsubnet.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,8 @@ class mac_linux(netaddr.mac_unix):
vars:
address: '192.168.144.5'
subnet: '192.168.0.0/16'
ipv6_address: '2001:4860:4860::8888'
ipv6_subnet: '2600:1f1c:1b3:8f00::/56'
tasks:
# If the given string is an IP address, it will be converted into a subnet.
- name: convert IP address to subnet
Expand All @@ -101,6 +103,16 @@ class mac_linux(netaddr.mac_unix):
debug:
msg: "{{ subnet | ansible.utils.ipsubnet(20, -1) }}"

# Get a new subnet with the specified index.
- name: Get first IPv6 subnet that has prefix length /120
debug:
msg: "{{ ipv6_subnet | ansible.utils.ipsubnet(120, 0) }}"

# Get a new subnet with the specified index.
- name: Get last subnet that has prefix length /120
debug:
msg: "{{ ipv6_subnet | ansible.utils.ipsubnet(120, -1) }}"

# If you specify an IP address instead of a subnet, and give a subnet size as the first argument, the ipsubnet() |
# filter will instead return the biggest subnet that contains that given IP address.
- name: Get biggest subnet that contains that given IP address.
Expand Down Expand Up @@ -137,7 +149,7 @@ class mac_linux(netaddr.mac_unix):

# By specifying another subnet as a second argument, if the second subnet includes the first, you can determine |
# the rank of the first subnet in the second.
- name: he fifth subnet /30 in a /24
- name: The fifth subnet /30 in a /24
debug:
msg: "{{ '192.168.144.16/30' | ansible.utils.ipsubnet('192.168.144.0/24') }}"

Expand Down
4 changes: 2 additions & 2 deletions plugins/plugin_utils/base/ipaddr_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,7 +368,7 @@ def _wrap_query(v, vtype, value):


def ipaddr(value, query="", version=False, alias="ipaddr"):
""" Check if string is an IP address or network and filter it """
"""Check if string is an IP address or network and filter it"""

query_func_extra_args = {
"": ("vtype",),
Expand Down Expand Up @@ -689,7 +689,7 @@ def _win_query(v):

# ---- HWaddr / MAC address filters ----
def hwaddr(value, query="", alias="hwaddr"):
""" Check if string is a HW/MAC address and filter it """
"""Check if string is a HW/MAC address and filter it"""

query_func_extra_args = {"": ("value",)}

Expand Down
100 changes: 36 additions & 64 deletions tests/unit/plugins/action/test_cli_parse.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ def setUp(self):

@staticmethod
def _load_fixture(filename):
""" Load a fixture from the filesystem
"""Load a fixture from the filesystem

:param filename: The name of the file to load
:type filename: str
Expand All @@ -72,23 +72,20 @@ def _load_fixture(filename):
return fhand.read()

def test_fn_debug(self):
""" Confirm debug doesn't fail and return None
"""
"""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
"""
"""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
"""
"""Confirm a valid argspec passes"""
kwargs = {
"text": "text",
"parser": {
Expand All @@ -102,8 +99,7 @@ def test_fn_check_argspec_pass(self):
self.assertEqual(valid, True)

def test_fn_check_argspec_fail_no_test_or_command(self):
""" Confirm failed argpsec w/o text or command
"""
"""Confirm failed argpsec w/o text or command"""
kwargs = {
"parser": {
"name": "ansible.utils.textfsm",
Expand All @@ -122,8 +118,7 @@ def test_fn_check_argspec_fail_no_test_or_command(self):
)

def test_fn_check_argspec_fail_no_parser_name(self):
""" Confirm failed argspec no parser name
"""
"""Confirm failed argspec no parser name"""
kwargs = {"text": "anything", "parser": {"command": "show version"}}
valid, result, updated_params = check_argspec(
DOCUMENTATION,
Expand All @@ -137,8 +132,7 @@ def test_fn_check_argspec_fail_no_parser_name(self):
)

def test_fn_extended_check_argspec_parser_name_not_coll(self):
""" Confirm failed argpsec parser not collection format
"""
"""Confirm failed argpsec parser not collection format"""
self._plugin._task.args = {
"text": "anything",
"parser": {
Expand All @@ -151,7 +145,7 @@ def test_fn_extended_check_argspec_parser_name_not_coll(self):
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
"""Confirm failed argpsec missing template_path
or command when text provided
"""
self._plugin._task.args = {
Expand All @@ -165,8 +159,7 @@ def test_fn_extended_check_argspec_missing_tpath_or_command(self):
)

def test_fn_load_parser_pass(self):
""" Confirm each each of the parsers loads from the filesystem
"""
"""Confirm each each of the parsers loads from the filesystem"""
parser_names = ["json", "textfsm", "ttp", "xml"]
for parser_name in parser_names:
self._plugin._task.args = {
Expand All @@ -179,8 +172,7 @@ def test_fn_load_parser_pass(self):
self.assertTrue(callable(parser.parse))

def test_fn_load_parser_fail(self):
""" Confirm missing parser fails gracefully
"""
"""Confirm missing parser fails gracefully"""
self._plugin._task.args = {
"text": "anything",
"parser": {"name": "a.b.c"},
Expand All @@ -191,7 +183,7 @@ def test_fn_load_parser_fail(self):
self.assertIn("No module named", self._plugin._result["msg"])

def test_fn_set_parser_command_missing(self):
""" Confirm parser/command is set if missing
"""Confirm parser/command is set if missing
and command provided
"""
self._plugin._task.args = {
Expand All @@ -204,8 +196,7 @@ def test_fn_set_parser_command_missing(self):
)

def test_fn_set_parser_command_present(self):
""" Confirm parser/command is not changed if provided
"""
"""Confirm parser/command is not changed if provided"""
self._plugin._task.args = {
"command": "anything",
"parser": {"command": "something", "name": "a.b.c"},
Expand All @@ -216,32 +207,28 @@ def test_fn_set_parser_command_present(self):
)

def test_fn_set_parser_command_absent(self):
""" Confirm parser/command is not added
"""
"""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
"""
"""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
"""
"""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
"""
"""Confirm os is set based on task vars"""
checks = [
("ansible_network_os", "cisco.nxos.nxos", "nxos"),
("ansible_network_os", "NXOS", "nxos"),
Expand All @@ -254,7 +241,7 @@ def test_fn_os_from_task_vars(self):
self.assertEqual(result, check[2])

def test_fn_update_template_path_not_exist(self):
""" Check the creation of the template_path if
"""Check the creation of the template_path if
it doesn't exist in the user provided data
"""
self._plugin._task.args = {
Expand All @@ -269,7 +256,7 @@ def test_fn_update_template_path_not_exist(self):
)

def test_fn_update_template_path_not_exist_os(self):
""" Check the creation of the template_path if
"""Check the creation of the template_path if
it doesn't exist in the user provided data
name based on os provided in task
"""
Expand All @@ -284,7 +271,7 @@ def test_fn_update_template_path_not_exist_os(self):
)

def test_fn_update_template_path_mock_find_needle(self):
""" Check the creation of the template_path
"""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
"""
Expand All @@ -302,8 +289,7 @@ def test_fn_update_template_path_mock_find_needle(self):
)

def test_fn_get_template_contents_pass(self):
""" Check the retrieval of the template contents
"""
"""Check the retrieval of the template contents"""
temp = tempfile.NamedTemporaryFile()
contents = "abcdef"
with open(temp.name, "w") as fileh:
Expand All @@ -314,8 +300,7 @@ def test_fn_get_template_contents_pass(self):
self.assertEqual(result, contents)

def test_fn_get_template_contents_missing(self):
""" Check the retrieval of the template contents
"""
"""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()
Expand All @@ -324,31 +309,27 @@ def test_fn_get_template_contents_missing(self):
)

def test_fn_get_template_contents_not_specified(self):
""" Check the none when template_path not specified
"""
"""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
"""
"""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
"""
"""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
"""
"""Check run command for non network"""
response = "abc"
self._plugin._connection.socket_path = None
self._plugin._low_level_execute_command = MagicMock()
Expand All @@ -363,8 +344,7 @@ def test_fn_run_command_lx_rc0(self):
self.assertEqual(self._plugin._result["stdout_lines"], response)

def test_fn_run_command_lx_rc1(self):
""" Check run command for non network
"""
"""Check run command for non network"""
response = "abc"
self._plugin._connection.socket_path = None
self._plugin._low_level_execute_command = MagicMock()
Expand All @@ -381,8 +361,7 @@ def test_fn_run_command_lx_rc1(self):

@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_command_network(self, mock_rpc):
""" Check run command for network
"""
"""Check run command for network"""
expected = "abc"
mock_rpc.return_value = expected
self._plugin._connection.socket_path = (
Expand All @@ -394,16 +373,14 @@ def test_fn_run_command_network(self, mock_rpc):
self.assertEqual(self._plugin._result["stdout_lines"], [expected])

def test_fn_run_command_not_specified(self):
""" Check run command for network
"""
"""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
"""
"""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 = (
Expand Down Expand Up @@ -431,8 +408,7 @@ def test_fn_run_pass_w_fact(self, mock_rpc):

@patch("ansible.module_utils.connection.Connection.__rpc__")
def test_fn_run_pass_wo_fact(self, mock_rpc):
""" Check full module run with valid params
"""
"""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 = (
Expand All @@ -456,8 +432,7 @@ def test_fn_run_pass_wo_fact(self, mock_rpc):
self.assertNotIn("ansible_facts", result)

def test_fn_run_fail_argspec(self):
""" Check full module run with invalid params
"""
"""Check full module run with invalid params"""
self._plugin._task.args = {
"text": "anything",
"parser": {
Expand All @@ -470,8 +445,7 @@ def test_fn_run_fail_argspec(self):
self.assertIn("including collection", self._plugin._result["msg"])

def test_fn_run_fail_command(self):
""" Confirm clean fail with rc 1
"""
"""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 = {
Expand All @@ -495,8 +469,7 @@ def test_fn_run_fail_command(self):
self.assertEqual(result, expected)

def test_fn_run_fail_missing_parser(self):
"""Confirm clean fail with missing parser
"""
"""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)
Expand All @@ -505,7 +478,7 @@ def test_fn_run_fail_missing_parser(self):

@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
"""Check full module run using parser w/o
DEFAULT_TEMPLATE_EXTENSION or PROVIDE_TEMPLATE_CONTENTS
defined in the parser
"""
Expand Down Expand Up @@ -542,7 +515,7 @@ def parse(self, *_args, **kwargs):

@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
"""Check full module run using parser w/o
a parser function defined in the parser
defined in the parser
"""
Expand Down Expand Up @@ -577,8 +550,7 @@ class CliParser(CliParserBase):

@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
"""
"""Check full module run mock error from network device"""
msg = "I was mocked"
mock_rpc.side_effect = AnsibleConnectionError(msg)
self._plugin._connection.socket_path = (
Expand Down
Loading