diff --git a/anta/input_models/__init__.py b/anta/input_models/__init__.py new file mode 100644 index 000000000..5b8974c76 --- /dev/null +++ b/anta/input_models/__init__.py @@ -0,0 +1,4 @@ +# Copyright (c) 2023-2024 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Package related to all ANTA tests input models.""" diff --git a/anta/input_models/connectivity.py b/anta/input_models/connectivity.py new file mode 100644 index 000000000..5e40f6205 --- /dev/null +++ b/anta/input_models/connectivity.py @@ -0,0 +1,41 @@ +# Copyright (c) 2023-2024 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Module containing input models for connectivity tests.""" + +from __future__ import annotations + +from ipaddress import IPv4Address + +from pydantic import BaseModel, ConfigDict + +from anta.custom_types import Interface + + +class Host(BaseModel): + """Model for a remote host to ping.""" + + model_config = ConfigDict(extra="forbid") + destination: IPv4Address + """IPv4 address to ping.""" + source: IPv4Address | Interface + """IPv4 address source IP or egress interface to use.""" + vrf: str = "default" + """VRF context. Defaults to `default`.""" + repeat: int = 2 + """Number of ping repetition. Defaults to 2.""" + size: int = 100 + """Specify datagram size. Defaults to 100.""" + df_bit: bool = False + """Enable do not fragment bit in IP header. Defaults to False.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the Host for reporting. + + Examples + -------- + Host 10.1.1.1 (src: 10.2.2.2, vrf: mgmt, size: 100B, repeat: 2) + + """ + df_status = ", df-bit: enabled" if self.df_bit else "" + return f"Host {self.destination} (src: {self.source}, vrf: {self.vrf}, size: {self.size}B, repeat: {self.repeat}{df_status})" diff --git a/anta/tests/connectivity.py b/anta/tests/connectivity.py index 4174c1a9a..374e51090 100644 --- a/anta/tests/connectivity.py +++ b/anta/tests/connectivity.py @@ -7,12 +7,12 @@ # mypy: disable-error-code=attr-defined from __future__ import annotations -from ipaddress import IPv4Address from typing import ClassVar from pydantic import BaseModel from anta.custom_types import Interface +from anta.input_models.connectivity import Host from anta.models import AntaCommand, AntaTemplate, AntaTest @@ -44,8 +44,7 @@ class VerifyReachability(AntaTest): """ categories: ClassVar[list[str]] = ["connectivity"] - # Removing the between '{size}' and '{df_bit}' to compensate the df-bit set default value - # i.e if df-bit kept disable then it will add redundant space in between the command + # Template uses '{size}{df_bit}' without space since df_bit includes leading space when enabled commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ AntaTemplate(template="ping vrf {vrf} {destination} source {source} size {size}{df_bit} repeat {repeat}", revision=1) ] @@ -55,29 +54,13 @@ class Input(AntaTest.Input): hosts: list[Host] """List of host to ping.""" - - class Host(BaseModel): - """Model for a remote host to ping.""" - - destination: IPv4Address - """IPv4 address to ping.""" - source: IPv4Address | Interface - """IPv4 address source IP or egress interface to use.""" - vrf: str = "default" - """VRF context. Defaults to `default`.""" - repeat: int = 2 - """Number of ping repetition. Defaults to 2.""" - size: int = 100 - """Specify datagram size. Defaults to 100.""" - df_bit: bool = False - """Enable do not fragment bit in IP header. Defaults to False.""" + Host: ClassVar[type[Host]] = Host def render(self, template: AntaTemplate) -> list[AntaCommand]: """Render the template for each host in the input list.""" commands = [] for host in self.inputs.hosts: - # Enables do not fragment bit in IP header if needed else keeping disable. - # Adding the at start to compensate change in AntaTemplate + # df_bit includes leading space when enabled, empty string when disabled df_bit = " df-bit" if host.df_bit else "" command = template.render(destination=host.destination, source=host.source, vrf=host.vrf, repeat=host.repeat, size=host.size, df_bit=df_bit) commands.append(command) @@ -86,20 +69,11 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]: @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyReachability.""" - failures = [] - - for command in self.instance_commands: - src = command.params.source - dst = command.params.destination - repeat = command.params.repeat - - if f"{repeat} received" not in command.json_output["messages"][0]: - failures.append((str(src), str(dst))) + self.result.is_success() - if not failures: - self.result.is_success() - else: - self.result.is_failure(f"Connectivity test failed for the following source-destination pairs: {failures}") + for command, host in zip(self.instance_commands, self.inputs.hosts): + if f"{host.repeat} received" not in command.json_output["messages"][0]: + self.result.is_failure(f"{host} - Unreachable") class VerifyLLDPNeighbors(AntaTest): diff --git a/docs/api/tests.connectivity.md b/docs/api/tests.connectivity.md index 0dd5d4476..cb0cf061e 100644 --- a/docs/api/tests.connectivity.md +++ b/docs/api/tests.connectivity.md @@ -7,7 +7,10 @@ anta_title: ANTA catalog for connectivity tests ~ that can be found in the LICENSE file. --> +# Tests + ::: anta.tests.connectivity + options: show_root_heading: false show_root_toc_entry: false @@ -18,3 +21,15 @@ anta_title: ANTA catalog for connectivity tests filters: - "!test" - "!render" + +# Input models + +::: anta.input_models.connectivity + + options: + show_root_heading: false + show_root_toc_entry: false + show_bases: false + anta_hide_test_module_description: true + show_labels: true + filters: ["!^__str__"] diff --git a/docs/templates/python/material/anta_test_input_model.html.jinja b/docs/templates/python/material/anta_test_input_model.html.jinja new file mode 100644 index 000000000..ccd4307dd --- /dev/null +++ b/docs/templates/python/material/anta_test_input_model.html.jinja @@ -0,0 +1,164 @@ +{% if obj.members %} + {{ log.debug("Rendering children of " + obj.path) }} + +
+ + {% if root_members %} + {% set members_list = config.members %} + {% else %} + {% set members_list = none %} + {% endif %} + + {% if config.group_by_category %} + + {% with %} + + {% if config.show_category_heading %} + {% set extra_level = 1 %} + {% else %} + {% set extra_level = 0 %} + {% endif %} + + {% with attributes = obj.attributes|filter_objects( + filters=config.filters, + members_list=members_list, + inherited_members=config.inherited_members, + keep_no_docstrings=config.show_if_no_docstring, + ) %} + {% if attributes %} + {% if config.show_category_heading %} + {% filter heading(heading_level, id=html_id ~ "-attributes") %}Attributes{% endfilter %} + {% endif %} + {% with heading_level = heading_level + extra_level %} + {% set root = False %} + {% set heading_level = heading_level + 1 %} + {% set old_obj = obj %} + {% set obj = class %} + {% include "attributes_table.html" with context %} + {% set obj = old_obj %} + {% endwith %} + {% endif %} + {% endwith %} + + {% with classes = obj.classes|filter_objects( + filters=config.filters, + members_list=members_list, + inherited_members=config.inherited_members, + keep_no_docstrings=config.show_if_no_docstring, + ) %} + {% if classes %} + {% if config.show_category_heading %} + {% filter heading(heading_level, id=html_id ~ "-classes") %}Classes{% endfilter %} + {% endif %} + {% with heading_level = heading_level + extra_level %} + {% for class in classes|order_members(config.members_order, members_list) %} + {% if class.name == "Input" %} + {% filter heading(heading_level, id=html_id ~ "-attributes") %}Inputs{% endfilter %} + {% set root = False %} + {% set heading_level = heading_level + 1 %} + {% set old_obj = obj %} + {% set obj = class %} + {% include "attributes_table.html" with context %} + {% set obj = old_obj %} + {% else %} + {% if members_list is not none or class.is_public %} + {% include class|get_template with context %} + {% endif %} + {% endif %} + {% endfor %} + {% endwith %} + {% endif %} + {% endwith %} + + {% with functions = obj.functions|filter_objects( + filters=config.filters, + members_list=members_list, + inherited_members=config.inherited_members, + keep_no_docstrings=config.show_if_no_docstring, + ) %} + {% if functions %} + {% if config.show_category_heading %} + {% filter heading(heading_level, id=html_id ~ "-functions") %}Functions{% endfilter %} + {% endif %} + {% with heading_level = heading_level + extra_level %} + {% for function in functions|order_members(config.members_order, members_list) %} + {% if not (obj.kind.value == "class" and function.name == "__init__" and config.merge_init_into_class) %} + {% if members_list is not none or function.is_public %} + {% include function|get_template with context %} + {% endif %} + {% endif %} + {% endfor %} + {% endwith %} + {% endif %} + {% endwith %} + + {% if config.show_submodules %} + {% with modules = obj.modules|filter_objects( + filters=config.filters, + members_list=members_list, + inherited_members=config.inherited_members, + keep_no_docstrings=config.show_if_no_docstring, + ) %} + {% if modules %} + {% if config.show_category_heading %} + {% filter heading(heading_level, id=html_id ~ "-modules") %}Modules{% endfilter %} + {% endif %} + {% with heading_level = heading_level + extra_level %} + {% for module in modules|order_members(config.members_order.alphabetical, members_list) %} + {% if members_list is not none or module.is_public %} + {% include module|get_template with context %} + {% endif %} + {% endfor %} + {% endwith %} + {% endif %} + {% endwith %} + {% endif %} + + {% endwith %} + + {% else %} + + {% for child in obj.all_members + |filter_objects( + filters=config.filters, + members_list=members_list, + inherited_members=config.inherited_members, + keep_no_docstrings=config.show_if_no_docstring, + ) + |order_members(config.members_order, members_list) + %} + + {% if not (obj.is_class and child.name == "__init__" and config.merge_init_into_class) %} + + {% if members_list is not none or child.is_public %} + {% if child.is_attribute %} + {% with attribute = child %} + {% include attribute|get_template with context %} + {% endwith %} + + {% elif child.is_class %} + {% with class = child %} + {% include class|get_template with context %} + {% endwith %} + + {% elif child.is_function %} + {% with function = child %} + {% include function|get_template with context %} + {% endwith %} + + {% elif child.is_module and config.show_submodules %} + {% with module = child %} + {% include module|get_template with context %} + {% endwith %} + + {% endif %} + {% endif %} + + {% endif %} + + {% endfor %} + + {% endif %} + +
+{% endif %} diff --git a/docs/templates/python/material/class.html.jinja b/docs/templates/python/material/class.html.jinja index 1c1173ce4..c9c672fd0 100644 --- a/docs/templates/python/material/class.html.jinja +++ b/docs/templates/python/material/class.html.jinja @@ -1,9 +1,12 @@ {% extends "_base/class.html.jinja" %} {% set anta_test = namespace(found=false) %} +{% set anta_test_input_model = namespace(found=false) %} {% for base in class.bases %} {% set basestr = base | string %} {% if "AntaTest" == basestr %} {% set anta_test.found = True %} +{% elif class.parent.parent.name == "input_models" %} +{% set anta_test_input_model.found = True %} {% endif %} {% endfor %} {% block children %} @@ -22,6 +25,21 @@ {{ class.source|highlight(language="python", linestart=class.lineno, linenums=True) }} +{% elif anta_test_input_model.found %} + {% set root = False %} + {% set heading_level = heading_level + 1 %} + {% include "anta_test_input_model.html.jinja" with context %} + {# render source after children - TODO make add flag to respect disabling it.. though do we want to disable?#} +
+ Source code in + {%- if class.relative_filepath.is_absolute() -%} + {{ class.relative_package_filepath }} + {%- else -%} + {{ class.relative_filepath }} + {%- endif -%} + + {{ class.source|highlight(language="python", linestart=class.lineno, linenums=True) }} +
{% else %} {{ super() }} {% endif %} diff --git a/tests/units/anta_tests/test_connectivity.py b/tests/units/anta_tests/test_connectivity.py index beeaae65c..1d13cbdb6 100644 --- a/tests/units/anta_tests/test_connectivity.py +++ b/tests/units/anta_tests/test_connectivity.py @@ -153,7 +153,7 @@ ], }, ], - "expected": {"result": "failure", "messages": ["Connectivity test failed for the following source-destination pairs: [('10.0.0.5', '10.0.0.11')]"]}, + "expected": {"result": "failure", "messages": ["Host 10.0.0.11 (src: 10.0.0.5, vrf: default, size: 100B, repeat: 2) - Unreachable"]}, }, { "name": "failure-interface", @@ -187,7 +187,7 @@ ], }, ], - "expected": {"result": "failure", "messages": ["Connectivity test failed for the following source-destination pairs: [('Management0', '10.0.0.11')]"]}, + "expected": {"result": "failure", "messages": ["Host 10.0.0.11 (src: Management0, vrf: default, size: 100B, repeat: 2) - Unreachable"]}, }, { "name": "failure-size", @@ -209,7 +209,7 @@ ], }, ], - "expected": {"result": "failure", "messages": ["Connectivity test failed for the following source-destination pairs: [('Management0', '10.0.0.1')]"]}, + "expected": {"result": "failure", "messages": ["Host 10.0.0.1 (src: Management0, vrf: default, size: 1501B, repeat: 5, df-bit: enabled) - Unreachable"]}, }, { "name": "success",