diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index bbbb977dd..69a40ad95 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -43,16 +43,16 @@ repos: - --allow-past-years - --fuzzy-match-generates-todo - --comment-style - - '' + - "" - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.9.6 + rev: v0.9.7 hooks: - - id: ruff - name: Run Ruff linter - args: [ --fix ] - - id: ruff-format - name: Run Ruff formatter + - id: ruff + name: Run Ruff linter + args: [--fix] + - id: ruff-format + name: Run Ruff formatter - repo: https://github.com/pycqa/pylint rev: "v3.3.4" @@ -62,9 +62,9 @@ repos: description: This hook runs pylint. types: [python] args: - - -rn # Only display messages - - -sn # Don't display the score - - --rcfile=pyproject.toml # Link to config file + - -rn # Only display messages + - -sn # Don't display the score + - --rcfile=pyproject.toml # Link to config file additional_dependencies: - anta[cli] - types-PyYAML @@ -123,5 +123,14 @@ repos: pass_filenames: false additional_dependencies: - anta[cli] - # TODO: next can go once we have it added to anta properly - - numpydoc + - id: doc-snippets + name: Generate doc snippets + entry: >- + sh -c "docs/scripts/generate_doc_snippets.py" + language: python + types: [python] + files: anta/cli/ + verbose: true + pass_filenames: false + additional_dependencies: + - anta[cli] diff --git a/anta/cli/nrfu/__init__.py b/anta/cli/nrfu/__init__.py index a6f76c4cd..6dc912dcc 100644 --- a/anta/cli/nrfu/__init__.py +++ b/anta/cli/nrfu/__init__.py @@ -42,9 +42,10 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]: if "--help" not in args: raise - # remove the required params so that help can display + # Fake presence of the required params so that help can display for param in self.params: - param.required = False + if param.required: + param.value_is_missing = lambda value: False # type: ignore[method-assign] # noqa: ARG005 return super().parse_args(ctx, args) diff --git a/anta/cli/nrfu/commands.py b/anta/cli/nrfu/commands.py index d1a72a01f..ed0f43244 100644 --- a/anta/cli/nrfu/commands.py +++ b/anta/cli/nrfu/commands.py @@ -45,7 +45,10 @@ def table(ctx: click.Context, group_by: Literal["device", "test"] | None) -> Non help="Path to save report as a JSON file", ) def json(ctx: click.Context, output: pathlib.Path | None) -> None: - """ANTA command to check network state with JSON results.""" + """ANTA command to check network state with JSON results. + + If no `--output` is specified, the output is printed to stdout. + """ run_tests(ctx) print_json(ctx, output=output) exit_with_code(ctx) @@ -72,11 +75,11 @@ def text(ctx: click.Context) -> None: path_type=pathlib.Path, ), show_envvar=True, - required=False, + required=True, help="Path to save report as a CSV file", ) def csv(ctx: click.Context, csv_output: pathlib.Path) -> None: - """ANTA command to check network states with CSV result.""" + """ANTA command to check network state with CSV report.""" run_tests(ctx) save_to_csv(ctx, csv_file=csv_output) exit_with_code(ctx) diff --git a/anta/device.py b/anta/device.py index 3624fdb2e..f685357c5 100644 --- a/anta/device.py +++ b/anta/device.py @@ -32,6 +32,10 @@ # https://github.com/pyca/cryptography/issues/7236#issuecomment-1131908472 CLIENT_KEYS = asyncssh.public_key.load_default_keypairs() +# Limit concurrency to 100 requests (HTTPX default) to avoid high-concurrency performance issues +# See: https://github.com/encode/httpx/issues/3215 +MAX_CONCURRENT_REQUESTS = 100 + class AntaCache: """Class to be used as cache. @@ -296,6 +300,7 @@ async def copy(self, sources: list[Path], destination: Path, direction: Literal[ raise NotImplementedError(msg) +# pylint: disable=too-many-instance-attributes class AsyncEOSDevice(AntaDevice): """Implementation of AntaDevice for EOS using aio-eapi. @@ -388,6 +393,10 @@ def __init__( # noqa: PLR0913 host=host, port=ssh_port, username=username, password=password, client_keys=CLIENT_KEYS, **ssh_params ) + # In Python 3.9, Semaphore must be created within a running event loop + # TODO: Once we drop Python 3.9 support, initialize the semaphore here + self._command_semaphore: asyncio.Semaphore | None = None + def __rich_repr__(self) -> Iterator[tuple[str, Any]]: """Implement Rich Repr Protocol. @@ -431,6 +440,15 @@ def _keys(self) -> tuple[Any, ...]: """ return (self._session.host, self._session.port) + async def _get_semaphore(self) -> asyncio.Semaphore: + """Return the semaphore, initializing it if needed. + + TODO: Remove this method once we drop Python 3.9 support. + """ + if self._command_semaphore is None: + self._command_semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS) + return self._command_semaphore + async def _collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None: """Collect device command output from EOS using aio-eapi. @@ -445,57 +463,63 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No collection_id An identifier used to build the eAPI request ID. """ - commands: list[dict[str, str | int]] = [] - if self.enable and self._enable_password is not None: - commands.append( - { - "cmd": "enable", - "input": str(self._enable_password), - }, - ) - elif self.enable: - # No password - commands.append({"cmd": "enable"}) - commands += [{"cmd": command.command, "revision": command.revision}] if command.revision else [{"cmd": command.command}] - try: - response: list[dict[str, Any] | str] = await self._session.cli( - commands=commands, - ofmt=command.ofmt, - version=command.version, - req_id=f"ANTA-{collection_id}-{id(command)}" if collection_id else f"ANTA-{id(command)}", - ) # type: ignore[assignment] # multiple commands returns a list - # Do not keep response of 'enable' command - command.output = response[-1] - except asynceapi.EapiCommandError as e: - # This block catches exceptions related to EOS issuing an error. - self._log_eapi_command_error(command, e) - except TimeoutException as e: - # This block catches Timeout exceptions. - command.errors = [exc_to_str(e)] - timeouts = self._session.timeout.as_dict() - logger.error( - "%s occurred while sending a command to %s. Consider increasing the timeout.\nCurrent timeouts: Connect: %s | Read: %s | Write: %s | Pool: %s", - exc_to_str(e), - self.name, - timeouts["connect"], - timeouts["read"], - timeouts["write"], - timeouts["pool"], - ) - except (ConnectError, OSError) as e: - # This block catches OSError and socket issues related exceptions. - command.errors = [exc_to_str(e)] - if (isinstance(exc := e.__cause__, httpcore.ConnectError) and isinstance(os_error := exc.__context__, OSError)) or isinstance(os_error := e, OSError): # pylint: disable=no-member - if isinstance(os_error.__cause__, OSError): - os_error = os_error.__cause__ - logger.error("A local OS error occurred while connecting to %s: %s.", self.name, os_error) - else: + semaphore = await self._get_semaphore() + + async with semaphore: + commands: list[dict[str, str | int]] = [] + if self.enable and self._enable_password is not None: + commands.append( + { + "cmd": "enable", + "input": str(self._enable_password), + }, + ) + elif self.enable: + # No password + commands.append({"cmd": "enable"}) + commands += [{"cmd": command.command, "revision": command.revision}] if command.revision else [{"cmd": command.command}] + try: + response: list[dict[str, Any] | str] = await self._session.cli( + commands=commands, + ofmt=command.ofmt, + version=command.version, + req_id=f"ANTA-{collection_id}-{id(command)}" if collection_id else f"ANTA-{id(command)}", + ) # type: ignore[assignment] # multiple commands returns a list + # Do not keep response of 'enable' command + command.output = response[-1] + except asynceapi.EapiCommandError as e: + # This block catches exceptions related to EOS issuing an error. + self._log_eapi_command_error(command, e) + except TimeoutException as e: + # This block catches Timeout exceptions. + command.errors = [exc_to_str(e)] + timeouts = self._session.timeout.as_dict() + logger.error( + "%s occurred while sending a command to %s. Consider increasing the timeout.\nCurrent timeouts: Connect: %s | Read: %s | Write: %s | Pool: %s", + exc_to_str(e), + self.name, + timeouts["connect"], + timeouts["read"], + timeouts["write"], + timeouts["pool"], + ) + except (ConnectError, OSError) as e: + # This block catches OSError and socket issues related exceptions. + command.errors = [exc_to_str(e)] + # pylint: disable=no-member + if (isinstance(exc := e.__cause__, httpcore.ConnectError) and isinstance(os_error := exc.__context__, OSError)) or isinstance( + os_error := e, OSError + ): + if isinstance(os_error.__cause__, OSError): + os_error = os_error.__cause__ + logger.error("A local OS error occurred while connecting to %s: %s.", self.name, os_error) + else: + anta_log_exception(e, f"An error occurred while issuing an eAPI request to {self.name}", logger) + except HTTPError as e: + # This block catches most of the httpx Exceptions and logs a general message. + command.errors = [exc_to_str(e)] anta_log_exception(e, f"An error occurred while issuing an eAPI request to {self.name}", logger) - except HTTPError as e: - # This block catches most of the httpx Exceptions and logs a general message. - command.errors = [exc_to_str(e)] - anta_log_exception(e, f"An error occurred while issuing an eAPI request to {self.name}", logger) - logger.debug("%s: %s", self.name, command) + logger.debug("%s: %s", self.name, command) def _log_eapi_command_error(self, command: AntaCommand, e: asynceapi.EapiCommandError) -> None: """Appropriately log the eapi command error.""" diff --git a/anta/input_models/connectivity.py b/anta/input_models/connectivity.py index f967e775b..464d22a51 100644 --- a/anta/input_models/connectivity.py +++ b/anta/input_models/connectivity.py @@ -23,13 +23,15 @@ class Host(BaseModel): source: IPv4Address | IPv6Address | Interface """Source address IP or egress interface to use.""" vrf: str = "default" - """VRF context. Defaults to `default`.""" + """VRF context.""" repeat: int = 2 - """Number of ping repetition. Defaults to 2.""" + """Number of ping repetition.""" size: int = 100 - """Specify datagram size. Defaults to 100.""" + """Specify datagram size.""" df_bit: bool = False - """Enable do not fragment bit in IP header. Defaults to False.""" + """Enable do not fragment bit in IP header.""" + reachable: bool = True + """Indicates whether the destination should be reachable.""" def __str__(self) -> str: """Return a human-readable string representation of the Host for reporting. diff --git a/anta/input_models/interfaces.py b/anta/input_models/interfaces.py index 02ad76947..04cd4b223 100644 --- a/anta/input_models/interfaces.py +++ b/anta/input_models/interfaces.py @@ -5,7 +5,9 @@ from __future__ import annotations -from typing import Literal +from ipaddress import IPv4Interface +from typing import Any, Literal +from warnings import warn from pydantic import BaseModel, ConfigDict @@ -13,7 +15,10 @@ class InterfaceState(BaseModel): - """Model for an interface state.""" + """Model for an interface state. + + TODO: Need to review this class name in ANTA v2.0.0. + """ model_config = ConfigDict(extra="forbid") name: Interface @@ -33,6 +38,10 @@ class InterfaceState(BaseModel): Can be enabled in the `VerifyLACPInterfacesStatus` tests. """ + primary_ip: IPv4Interface | None = None + """Primary IPv4 address in CIDR notation. Required field in the `VerifyInterfaceIPv4` test.""" + secondary_ips: list[IPv4Interface] | None = None + """List of secondary IPv4 addresses in CIDR notation. Can be provided in the `VerifyInterfaceIPv4` test.""" def __str__(self) -> str: """Return a human-readable string representation of the InterfaceState for reporting. @@ -46,3 +55,21 @@ def __str__(self) -> str: if self.portchannel is not None: base_string += f" Port-Channel: {self.portchannel}" return base_string + + +class InterfaceDetail(InterfaceState): # pragma: no cover + """Alias for the InterfaceState model to maintain backward compatibility. + + When initialized, it will emit a deprecation warning and call the InterfaceState model. + + TODO: Remove this class in ANTA v2.0.0. + """ + + def __init__(self, **data: Any) -> None: # noqa: ANN401 + """Initialize the InterfaceState class, emitting a depreciation warning.""" + warn( + message="InterfaceDetail model is deprecated and will be removed in ANTA v2.0.0. Use the InterfaceState model instead.", + category=DeprecationWarning, + stacklevel=2, + ) + super().__init__(**data) diff --git a/anta/input_models/system.py b/anta/input_models/system.py index fd4a27097..1771c1a63 100644 --- a/anta/input_models/system.py +++ b/anta/input_models/system.py @@ -28,4 +28,4 @@ class NTPServer(BaseModel): def __str__(self) -> str: """Representation of the NTPServer model.""" - return f"{self.server_address} (Preferred: {self.preferred}, Stratum: {self.stratum})" + return f"NTP Server: {self.server_address} Preferred: {self.preferred} Stratum: {self.stratum}" diff --git a/anta/tests/connectivity.py b/anta/tests/connectivity.py index 2fd58c1bb..6568d8424 100644 --- a/anta/tests/connectivity.py +++ b/anta/tests/connectivity.py @@ -37,6 +37,7 @@ class VerifyReachability(AntaTest): vrf: MGMT df_bit: True size: 100 + reachable: true - source: Management0 destination: 8.8.8.8 vrf: MGMT @@ -47,6 +48,7 @@ class VerifyReachability(AntaTest): vrf: default df_bit: True size: 100 + reachable: false ``` """ @@ -89,9 +91,14 @@ def test(self) -> None: self.result.is_success() for command, host in zip(self.instance_commands, self.inputs.hosts): - if f"{host.repeat} received" not in command.json_output["messages"][0]: + # Verifies the network is reachable + if host.reachable and f"{host.repeat} received" not in command.json_output["messages"][0]: self.result.is_failure(f"{host} - Unreachable") + # Verifies the network is unreachable. + if not host.reachable and f"{host.repeat} received" in command.json_output["messages"][0]: + self.result.is_failure(f"{host} - Destination is expected to be unreachable but found reachable.") + class VerifyLLDPNeighbors(AntaTest): """Verifies the connection status of the specified LLDP (Link Layer Discovery Protocol) neighbors. diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py index f1a32a31d..091238883 100644 --- a/anta/tests/interfaces.py +++ b/anta/tests/interfaces.py @@ -8,16 +8,14 @@ from __future__ import annotations import re -from ipaddress import IPv4Interface from typing import ClassVar, TypeVar from pydantic import BaseModel, Field, field_validator from pydantic_extra_types.mac_address import MacAddress -from anta import GITHUB_SUGGESTION from anta.custom_types import EthernetInterface, Interface, Percent, PositiveInteger from anta.decorators import skip_on_platforms -from anta.input_models.interfaces import InterfaceState +from anta.input_models.interfaces import InterfaceDetail, InterfaceState from anta.models import AntaCommand, AntaTemplate, AntaTest from anta.tools import custom_division, format_data, get_failed_logs, get_item, get_value @@ -513,7 +511,7 @@ def test(self) -> None: class VerifyIPProxyARP(AntaTest): - """Verifies if Proxy-ARP is enabled for the provided list of interface(s). + """Verifies if Proxy ARP is enabled. Expected Results ---------------- @@ -531,32 +529,28 @@ class VerifyIPProxyARP(AntaTest): ``` """ - description = "Verifies if Proxy ARP is enabled." categories: ClassVar[list[str]] = ["interfaces"] - commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show ip interface {intf}", revision=2)] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ip interface", revision=2)] class Input(AntaTest.Input): """Input model for the VerifyIPProxyARP test.""" - interfaces: list[str] + interfaces: list[Interface] """List of interfaces to be tested.""" - def render(self, template: AntaTemplate) -> list[AntaCommand]: - """Render the template for each interface in the input list.""" - return [template.render(intf=intf) for intf in self.inputs.interfaces] - @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyIPProxyARP.""" - disabled_intf = [] - for command in self.instance_commands: - intf = command.params.intf - if not command.json_output["interfaces"][intf]["proxyArp"]: - disabled_intf.append(intf) - if disabled_intf: - self.result.is_failure(f"The following interface(s) have Proxy-ARP disabled: {disabled_intf}") - else: - self.result.is_success() + self.result.is_success() + command_output = self.instance_commands[0].json_output + + for interface in self.inputs.interfaces: + if (interface_detail := get_value(command_output["interfaces"], f"{interface}", separator="..")) is None: + self.result.is_failure(f"Interface: {interface} - Not found") + continue + + if not interface_detail["proxyArp"]: + self.result.is_failure(f"Interface: {interface} - Proxy-ARP disabled") class VerifyL2MTU(AntaTest): @@ -624,7 +618,7 @@ def test(self) -> None: class VerifyInterfaceIPv4(AntaTest): - """Verifies if an interface is configured with a correct primary and list of optional secondary IPv4 addresses. + """Verifies the interface IPv4 addresses. Expected Results ---------------- @@ -645,83 +639,61 @@ class VerifyInterfaceIPv4(AntaTest): ``` """ - description = "Verifies the interface IPv4 addresses." categories: ClassVar[list[str]] = ["interfaces"] - commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show ip interface {interface}", revision=2)] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ip interface", revision=2)] class Input(AntaTest.Input): """Input model for the VerifyInterfaceIPv4 test.""" - interfaces: list[InterfaceDetail] + interfaces: list[InterfaceState] """List of interfaces with their details.""" + InterfaceDetail: ClassVar[type[InterfaceDetail]] = InterfaceDetail - class InterfaceDetail(BaseModel): - """Model for an interface detail.""" - - name: Interface - """Name of the interface.""" - primary_ip: IPv4Interface - """Primary IPv4 address in CIDR notation.""" - secondary_ips: list[IPv4Interface] | None = None - """Optional list of secondary IPv4 addresses in CIDR notation.""" - - def render(self, template: AntaTemplate) -> list[AntaCommand]: - """Render the template for each interface in the input list.""" - return [template.render(interface=interface.name) for interface in self.inputs.interfaces] + @field_validator("interfaces") + @classmethod + def validate_interfaces(cls, interfaces: list[T]) -> list[T]: + """Validate that 'primary_ip' field is provided in each interface.""" + for interface in interfaces: + if interface.primary_ip is None: + msg = f"{interface} 'primary_ip' field missing in the input" + raise ValueError(msg) + return interfaces @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyInterfaceIPv4.""" self.result.is_success() - for command in self.instance_commands: - intf = command.params.interface - for interface in self.inputs.interfaces: - if interface.name == intf: - input_interface_detail = interface - break - else: - self.result.is_failure(f"Could not find `{intf}` in the input interfaces. {GITHUB_SUGGESTION}") - continue - - input_primary_ip = str(input_interface_detail.primary_ip) - failed_messages = [] + command_output = self.instance_commands[0].json_output - # Check if the interface has an IP address configured - if not (interface_output := get_value(command.json_output, f"interfaces.{intf}.interfaceAddress")): - self.result.is_failure(f"For interface `{intf}`, IP address is not configured.") + for interface in self.inputs.interfaces: + if (interface_detail := get_value(command_output["interfaces"], f"{interface.name}", separator="..")) is None: + self.result.is_failure(f"{interface} - Not found") continue - primary_ip = get_value(interface_output, "primaryIp") + if (ip_address := get_value(interface_detail, "interfaceAddress.primaryIp")) is None: + self.result.is_failure(f"{interface} - IP address is not configured") + continue # Combine IP address and subnet for primary IP - actual_primary_ip = f"{primary_ip['address']}/{primary_ip['maskLen']}" + actual_primary_ip = f"{ip_address['address']}/{ip_address['maskLen']}" # Check if the primary IP address matches the input - if actual_primary_ip != input_primary_ip: - failed_messages.append(f"The expected primary IP address is `{input_primary_ip}`, but the actual primary IP address is `{actual_primary_ip}`.") + if actual_primary_ip != str(interface.primary_ip): + self.result.is_failure(f"{interface} - IP address mismatch - Expected: {interface.primary_ip} Actual: {actual_primary_ip}") - if (param_secondary_ips := input_interface_detail.secondary_ips) is not None: - input_secondary_ips = sorted([str(network) for network in param_secondary_ips]) - secondary_ips = get_value(interface_output, "secondaryIpsOrderedList") + if interface.secondary_ips: + if not (secondary_ips := get_value(interface_detail, "interfaceAddress.secondaryIpsOrderedList")): + self.result.is_failure(f"{interface} - Secondary IP address is not configured") + continue - # Combine IP address and subnet for secondary IPs actual_secondary_ips = sorted([f"{secondary_ip['address']}/{secondary_ip['maskLen']}" for secondary_ip in secondary_ips]) + input_secondary_ips = sorted([str(ip) for ip in interface.secondary_ips]) - # Check if the secondary IP address is configured - if not actual_secondary_ips: - failed_messages.append( - f"The expected secondary IP addresses are `{input_secondary_ips}`, but the actual secondary IP address is not configured." - ) - - # Check if the secondary IP addresses match the input - elif actual_secondary_ips != input_secondary_ips: - failed_messages.append( - f"The expected secondary IP addresses are `{input_secondary_ips}`, but the actual secondary IP addresses are `{actual_secondary_ips}`." + if actual_secondary_ips != input_secondary_ips: + self.result.is_failure( + f"{interface} - Secondary IP address mismatch - Expected: {', '.join(input_secondary_ips)} Actual: {', '.join(actual_secondary_ips)}" ) - if failed_messages: - self.result.is_failure(f"For interface `{intf}`, " + " ".join(failed_messages)) - class VerifyIpVirtualRouterMac(AntaTest): """Verifies the IP virtual router MAC address. diff --git a/anta/tests/mlag.py b/anta/tests/mlag.py index 215630c91..708271ca7 100644 --- a/anta/tests/mlag.py +++ b/anta/tests/mlag.py @@ -22,10 +22,8 @@ class VerifyMlagStatus(AntaTest): Expected Results ---------------- - * Success: The test will pass if the MLAG state is 'active', negotiation status is 'connected', - peer-link status and local interface status are 'up'. - * Failure: The test will fail if the MLAG state is not 'active', negotiation status is not 'connected', - peer-link status or local interface status are not 'up'. + * Success: The test will pass if the MLAG state is 'active', negotiation status is 'connected', peer-link status and local interface status are 'up'. + * Failure: The test will fail if the MLAG state is not 'active', negotiation status is not 'connected', peer-link status or local interface status are not 'up'. * Skipped: The test will be skipped if MLAG is 'disabled'. Examples @@ -42,21 +40,25 @@ class VerifyMlagStatus(AntaTest): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyMlagStatus.""" + self.result.is_success() command_output = self.instance_commands[0].json_output + + # Skipping the test if MLAG is disabled if command_output["state"] == "disabled": self.result.is_skipped("MLAG is disabled") return - keys_to_verify = ["state", "negStatus", "localIntfStatus", "peerLinkStatus"] - verified_output = {key: get_value(command_output, key) for key in keys_to_verify} - if ( - verified_output["state"] == "active" - and verified_output["negStatus"] == "connected" - and verified_output["localIntfStatus"] == "up" - and verified_output["peerLinkStatus"] == "up" - ): - self.result.is_success() - else: - self.result.is_failure(f"MLAG status is not OK: {verified_output}") + + # Verifies the negotiation status + if (neg_status := command_output["negStatus"]) != "connected": + self.result.is_failure(f"MLAG negotiation status mismatch - Expected: connected Actual: {neg_status}") + + # Verifies the local interface interface status + if (intf_state := command_output["localIntfStatus"]) != "up": + self.result.is_failure(f"Operational state of the MLAG local interface is not correct - Expected: up Actual: {intf_state}") + + # Verifies the peerLinkStatus + if (peer_link_state := command_output["peerLinkStatus"]) != "up": + self.result.is_failure(f"Operational state of the MLAG peer link is not correct - Expected: up Actual: {peer_link_state}") class VerifyMlagInterfaces(AntaTest): @@ -82,14 +84,19 @@ class VerifyMlagInterfaces(AntaTest): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyMlagInterfaces.""" + self.result.is_success() command_output = self.instance_commands[0].json_output + + # Skipping the test if MLAG is disabled if command_output["state"] == "disabled": self.result.is_skipped("MLAG is disabled") return - if command_output["mlagPorts"]["Inactive"] == 0 and command_output["mlagPorts"]["Active-partial"] == 0: - self.result.is_success() - else: - self.result.is_failure(f"MLAG status is not OK: {command_output['mlagPorts']}") + + # Verifies the Inactive and Active-partial ports + inactive_ports = command_output["mlagPorts"]["Inactive"] + partial_active_ports = command_output["mlagPorts"]["Active-partial"] + if inactive_ports != 0 or partial_active_ports != 0: + self.result.is_failure(f"MLAG status is not ok - Inactive Ports: {inactive_ports} Partial Active Ports: {partial_active_ports}") class VerifyMlagConfigSanity(AntaTest): @@ -116,16 +123,21 @@ class VerifyMlagConfigSanity(AntaTest): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyMlagConfigSanity.""" + self.result.is_success() command_output = self.instance_commands[0].json_output + + # Skipping the test if MLAG is disabled if command_output["mlagActive"] is False: self.result.is_skipped("MLAG is disabled") return - keys_to_verify = ["globalConfiguration", "interfaceConfiguration"] - verified_output = {key: get_value(command_output, key) for key in keys_to_verify} - if not any(verified_output.values()): - self.result.is_success() - else: - self.result.is_failure(f"MLAG config-sanity returned inconsistencies: {verified_output}") + + # Verifies the globalConfiguration config-sanity + if get_value(command_output, "globalConfiguration"): + self.result.is_failure("MLAG config-sanity found in global configuration") + + # Verifies the interfaceConfiguration config-sanity + if get_value(command_output, "interfaceConfiguration"): + self.result.is_failure("MLAG config-sanity found in interface configuration") class VerifyMlagReloadDelay(AntaTest): @@ -161,17 +173,21 @@ class Input(AntaTest.Input): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyMlagReloadDelay.""" + self.result.is_success() command_output = self.instance_commands[0].json_output + + # Skipping the test if MLAG is disabled if command_output["state"] == "disabled": self.result.is_skipped("MLAG is disabled") return - keys_to_verify = ["reloadDelay", "reloadDelayNonMlag"] - verified_output = {key: get_value(command_output, key) for key in keys_to_verify} - if verified_output["reloadDelay"] == self.inputs.reload_delay and verified_output["reloadDelayNonMlag"] == self.inputs.reload_delay_non_mlag: - self.result.is_success() - else: - self.result.is_failure(f"The reload-delay parameters are not configured properly: {verified_output}") + # Verifies the reloadDelay + if (reload_delay := get_value(command_output, "reloadDelay")) != self.inputs.reload_delay: + self.result.is_failure(f"MLAG reload-delay mismatch - Expected: {self.inputs.reload_delay}s Actual: {reload_delay}s") + + # Verifies the reloadDelayNonMlag + if (non_mlag_reload_delay := get_value(command_output, "reloadDelayNonMlag")) != self.inputs.reload_delay_non_mlag: + self.result.is_failure(f"Delay for non-MLAG ports mismatch - Expected: {self.inputs.reload_delay_non_mlag}s Actual: {non_mlag_reload_delay}s") class VerifyMlagDualPrimary(AntaTest): @@ -214,25 +230,37 @@ class Input(AntaTest.Input): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyMlagDualPrimary.""" + self.result.is_success() errdisabled_action = "errdisableAllInterfaces" if self.inputs.errdisabled else "none" command_output = self.instance_commands[0].json_output + + # Skipping the test if MLAG is disabled if command_output["state"] == "disabled": self.result.is_skipped("MLAG is disabled") return + + # Verifies the dualPrimaryDetectionState if command_output["dualPrimaryDetectionState"] == "disabled": self.result.is_failure("Dual-primary detection is disabled") return - keys_to_verify = ["detail.dualPrimaryDetectionDelay", "detail.dualPrimaryAction", "dualPrimaryMlagRecoveryDelay", "dualPrimaryNonMlagRecoveryDelay"] - verified_output = {key: get_value(command_output, key) for key in keys_to_verify} - if ( - verified_output["detail.dualPrimaryDetectionDelay"] == self.inputs.detection_delay - and verified_output["detail.dualPrimaryAction"] == errdisabled_action - and verified_output["dualPrimaryMlagRecoveryDelay"] == self.inputs.recovery_delay - and verified_output["dualPrimaryNonMlagRecoveryDelay"] == self.inputs.recovery_delay_non_mlag - ): - self.result.is_success() - else: - self.result.is_failure(f"The dual-primary parameters are not configured properly: {verified_output}") + + # Verifies the dualPrimaryAction + if (primary_action := get_value(command_output, "detail.dualPrimaryAction")) != errdisabled_action: + self.result.is_failure(f"Dual-primary action mismatch - Expected: {errdisabled_action} Actual: {primary_action}") + + # Verifies the dualPrimaryDetectionDelay + if (detection_delay := get_value(command_output, "detail.dualPrimaryDetectionDelay")) != self.inputs.detection_delay: + self.result.is_failure(f"Dual-primary detection delay mismatch - Expected: {self.inputs.detection_delay} Actual: {detection_delay}") + + # Verifies the dualPrimaryMlagRecoveryDelay + if (recovery_delay := get_value(command_output, "dualPrimaryMlagRecoveryDelay")) != self.inputs.recovery_delay: + self.result.is_failure(f"Dual-primary MLAG recovery delay mismatch - Expected: {self.inputs.recovery_delay} Actual: {recovery_delay}") + + # Verifies the dualPrimaryNonMlagRecoveryDelay + if (recovery_delay_non_mlag := get_value(command_output, "dualPrimaryNonMlagRecoveryDelay")) != self.inputs.recovery_delay_non_mlag: + self.result.is_failure( + f"Dual-primary non MLAG recovery delay mismatch - Expected: {self.inputs.recovery_delay_non_mlag} Actual: {recovery_delay_non_mlag}" + ) class VerifyMlagPrimaryPriority(AntaTest): @@ -282,6 +310,4 @@ def test(self) -> None: # Check primary priority if primary_priority != self.inputs.primary_priority: - self.result.is_failure( - f"The primary priority does not match expected. Expected `{self.inputs.primary_priority}`, but found `{primary_priority}` instead.", - ) + self.result.is_failure(f"MLAG primary priority mismatch - Expected: {self.inputs.primary_priority} Actual: {primary_priority}") diff --git a/anta/tests/stp.py b/anta/tests/stp.py index 87e3cd104..40f72c18e 100644 --- a/anta/tests/stp.py +++ b/anta/tests/stp.py @@ -7,7 +7,7 @@ # mypy: disable-error-code=attr-defined from __future__ import annotations -from typing import Any, ClassVar, Literal +from typing import ClassVar, Literal from pydantic import Field @@ -54,8 +54,7 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]: @AntaTest.anta_test def test(self) -> None: """Main test function for VerifySTPMode.""" - not_configured = [] - wrong_stp_mode = [] + self.result.is_success() for command in self.instance_commands: vlan_id = command.params.vlan if not ( @@ -64,15 +63,9 @@ def test(self) -> None: f"spanningTreeVlanInstances.{vlan_id}.spanningTreeVlanInstance.protocol", ) ): - not_configured.append(vlan_id) + self.result.is_failure(f"VLAN {vlan_id} STP mode: {self.inputs.mode} - Not configured") elif stp_mode != self.inputs.mode: - wrong_stp_mode.append(vlan_id) - if not_configured: - self.result.is_failure(f"STP mode '{self.inputs.mode}' not configured for the following VLAN(s): {not_configured}") - if wrong_stp_mode: - self.result.is_failure(f"Wrong STP mode configured for the following VLAN(s): {wrong_stp_mode}") - if not not_configured and not wrong_stp_mode: - self.result.is_success() + self.result.is_failure(f"VLAN {vlan_id} - Incorrect STP mode - Expected: {self.inputs.mode} Actual: {stp_mode}") class VerifySTPBlockedPorts(AntaTest): @@ -102,8 +95,8 @@ def test(self) -> None: self.result.is_success() else: for key, value in stp_instances.items(): - stp_instances[key] = value.pop("spanningTreeBlockedPorts") - self.result.is_failure(f"The following ports are blocked by STP: {stp_instances}") + stp_block_ports = value.get("spanningTreeBlockedPorts") + self.result.is_failure(f"STP Instance: {key} - Blocked ports - {', '.join(stp_block_ports)}") class VerifySTPCounters(AntaTest): @@ -128,14 +121,14 @@ class VerifySTPCounters(AntaTest): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifySTPCounters.""" + self.result.is_success() command_output = self.instance_commands[0].json_output - interfaces_with_errors = [ - interface for interface, counters in command_output["interfaces"].items() if counters["bpduTaggedError"] or counters["bpduOtherError"] != 0 - ] - if interfaces_with_errors: - self.result.is_failure(f"The following interfaces have STP BPDU packet errors: {interfaces_with_errors}") - else: - self.result.is_success() + + for interface, counters in command_output["interfaces"].items(): + if counters["bpduTaggedError"] != 0: + self.result.is_failure(f"Interface {interface} - STP BPDU packet tagged errors count mismatch - Expected: 0 Actual: {counters['bpduTaggedError']}") + if counters["bpduOtherError"] != 0: + self.result.is_failure(f"Interface {interface} - STP BPDU packet other errors count mismatch - Expected: 0 Actual: {counters['bpduOtherError']}") class VerifySTPForwardingPorts(AntaTest): @@ -174,25 +167,22 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]: @AntaTest.anta_test def test(self) -> None: """Main test function for VerifySTPForwardingPorts.""" - not_configured = [] - not_forwarding = [] + self.result.is_success() + interfaces_state = [] for command in self.instance_commands: vlan_id = command.params.vlan if not (topologies := get_value(command.json_output, "topologies")): - not_configured.append(vlan_id) - else: - interfaces_not_forwarding = [] - for value in topologies.values(): - if vlan_id and int(vlan_id) in value["vlans"]: - interfaces_not_forwarding = [interface for interface, state in value["interfaces"].items() if state["state"] != "forwarding"] - if interfaces_not_forwarding: - not_forwarding.append({f"VLAN {vlan_id}": interfaces_not_forwarding}) - if not_configured: - self.result.is_failure(f"STP instance is not configured for the following VLAN(s): {not_configured}") - if not_forwarding: - self.result.is_failure(f"The following VLAN(s) have interface(s) that are not in a forwarding state: {not_forwarding}") - if not not_configured and not interfaces_not_forwarding: - self.result.is_success() + self.result.is_failure(f"VLAN {vlan_id} - STP instance is not configured") + continue + for value in topologies.values(): + if vlan_id and int(vlan_id) in value["vlans"]: + interfaces_state = [ + (interface, actual_state) for interface, state in value["interfaces"].items() if (actual_state := state["state"]) != "forwarding" + ] + + if interfaces_state: + for interface, state in interfaces_state: + self.result.is_failure(f"VLAN {vlan_id} Interface: {interface} - Invalid state - Expected: forwarding Actual: {state}") class VerifySTPRootPriority(AntaTest): @@ -229,6 +219,7 @@ class Input(AntaTest.Input): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifySTPRootPriority.""" + self.result.is_success() command_output = self.instance_commands[0].json_output if not (stp_instances := command_output["instances"]): self.result.is_failure("No STP instances configured") @@ -240,16 +231,15 @@ def test(self) -> None: elif first_name.startswith("VL"): prefix = "VL" else: - self.result.is_failure(f"Unsupported STP instance type: {first_name}") + self.result.is_failure(f"STP Instance: {first_name} - Unsupported STP instance type") return check_instances = [f"{prefix}{instance_id}" for instance_id in self.inputs.instances] if self.inputs.instances else command_output["instances"].keys() - wrong_priority_instances = [ - instance for instance in check_instances if get_value(command_output, f"instances.{instance}.rootBridge.priority") != self.inputs.priority - ] - if wrong_priority_instances: - self.result.is_failure(f"The following instance(s) have the wrong STP root priority configured: {wrong_priority_instances}") - else: - self.result.is_success() + for instance in check_instances: + if not (instance_details := get_value(command_output, f"instances.{instance}")): + self.result.is_failure(f"Instance: {instance} - Not configured") + continue + if (priority := get_value(instance_details, "rootBridge.priority")) != self.inputs.priority: + self.result.is_failure(f"STP Instance: {instance} - Incorrect root priority - Expected: {self.inputs.priority} Actual: {priority}") class VerifyStpTopologyChanges(AntaTest): @@ -282,8 +272,7 @@ class Input(AntaTest.Input): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyStpTopologyChanges.""" - failures: dict[str, Any] = {"topologies": {}} - + self.result.is_success() command_output = self.instance_commands[0].json_output stp_topologies = command_output.get("topologies", {}) @@ -297,18 +286,12 @@ def test(self) -> None: # Verifies the number of changes across all interfaces for topology, topology_details in stp_topologies.items(): - interfaces = { - interface: {"Number of changes": num_of_changes} - for interface, details in topology_details.get("interfaces", {}).items() - if (num_of_changes := details.get("numChanges")) > self.inputs.threshold - } - if interfaces: - failures["topologies"][topology] = interfaces - - if failures["topologies"]: - self.result.is_failure(f"The following STP topologies are not configured or number of changes not within the threshold:\n{failures}") - else: - self.result.is_success() + for interface, details in topology_details.get("interfaces", {}).items(): + if (num_of_changes := details.get("numChanges")) > self.inputs.threshold: + self.result.is_failure( + f"Topology: {topology} Interface: {interface} - Number of changes not within the threshold - Expected: " + f"{self.inputs.threshold} Actual: {num_of_changes}" + ) class VerifySTPDisabledVlans(AntaTest): diff --git a/anta/tests/system.py b/anta/tests/system.py index 11cf8398c..9ec719180 100644 --- a/anta/tests/system.py +++ b/anta/tests/system.py @@ -24,7 +24,7 @@ class VerifyUptime(AntaTest): - """Verifies if the device uptime is higher than the provided minimum uptime value. + """Verifies the device uptime. Expected Results ---------------- @@ -40,7 +40,6 @@ class VerifyUptime(AntaTest): ``` """ - description = "Verifies the device uptime." categories: ClassVar[list[str]] = ["system"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show uptime", revision=1)] @@ -53,11 +52,10 @@ class Input(AntaTest.Input): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyUptime.""" + self.result.is_success() command_output = self.instance_commands[0].json_output - if command_output["upTime"] > self.inputs.minimum: - self.result.is_success() - else: - self.result.is_failure(f"Device uptime is {command_output['upTime']} seconds") + if command_output["upTime"] < self.inputs.minimum: + self.result.is_failure(f"Device uptime is incorrect - Expected: {self.inputs.minimum} Actual: {command_output['upTime']} seconds") class VerifyReloadCause(AntaTest): @@ -96,11 +94,11 @@ def test(self) -> None: ]: self.result.is_success() else: - self.result.is_failure(f"Reload cause is: '{command_output_data}'") + self.result.is_failure(f"Reload cause is: {command_output_data}") class VerifyCoredump(AntaTest): - """Verifies if there are core dump files in the /var/core directory. + """Verifies there are no core dump files. Expected Results ---------------- @@ -119,7 +117,6 @@ class VerifyCoredump(AntaTest): ``` """ - description = "Verifies there are no core dump files." categories: ClassVar[list[str]] = ["system"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show system coredump", revision=1)] @@ -133,7 +130,7 @@ def test(self) -> None: if not core_files: self.result.is_success() else: - self.result.is_failure(f"Core dump(s) have been found: {core_files}") + self.result.is_failure(f"Core dump(s) have been found: {', '.join(core_files)}") class VerifyAgentLogs(AntaTest): @@ -189,12 +186,11 @@ class VerifyCPUUtilization(AntaTest): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyCPUUtilization.""" + self.result.is_success() command_output = self.instance_commands[0].json_output command_output_data = command_output["cpuInfo"]["%Cpu(s)"]["idle"] - if command_output_data > CPU_IDLE_THRESHOLD: - self.result.is_success() - else: - self.result.is_failure(f"Device has reported a high CPU utilization: {100 - command_output_data}%") + if command_output_data < CPU_IDLE_THRESHOLD: + self.result.is_failure(f"Device has reported a high CPU utilization - Expected: < 75% Actual: {100 - command_output_data}%") class VerifyMemoryUtilization(AntaTest): @@ -219,12 +215,11 @@ class VerifyMemoryUtilization(AntaTest): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyMemoryUtilization.""" + self.result.is_success() command_output = self.instance_commands[0].json_output memory_usage = command_output["memFree"] / command_output["memTotal"] - if memory_usage > MEMORY_THRESHOLD: - self.result.is_success() - else: - self.result.is_failure(f"Device has reported a high memory usage: {(1 - memory_usage) * 100:.2f}%") + if memory_usage < MEMORY_THRESHOLD: + self.result.is_failure(f"Device has reported a high memory usage - Expected: < 75% Actual: {(1 - memory_usage) * 100:.2f}%") class VerifyFileSystemUtilization(AntaTest): @@ -253,7 +248,7 @@ def test(self) -> None: self.result.is_success() for line in command_output.split("\n")[1:]: if "loop" not in line and len(line) > 0 and (percentage := int(line.split()[4].replace("%", ""))) > DISK_SPACE_THRESHOLD: - self.result.is_failure(f"Mount point {line} is higher than 75%: reported {percentage}%") + self.result.is_failure(f"Mount point: {line} - Higher disk space utilization - Expected: {DISK_SPACE_THRESHOLD}% Actual: {percentage}%") class VerifyNTP(AntaTest): @@ -272,7 +267,6 @@ class VerifyNTP(AntaTest): ``` """ - description = "Verifies if NTP is synchronised." categories: ClassVar[list[str]] = ["system"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ntp status", ofmt="text")] @@ -284,7 +278,7 @@ def test(self) -> None: self.result.is_success() else: data = command_output.split("\n")[0] - self.result.is_failure(f"The device is not synchronized with the configured NTP server(s): '{data}'") + self.result.is_failure(f"NTP status mismatch - Expected: synchronised Actual: {data}") class VerifyNTPAssociations(AntaTest): diff --git a/docs/cli/nrfu.md b/docs/cli/nrfu.md index cb008a327..9e06f3645 100644 --- a/docs/cli/nrfu.md +++ b/docs/cli/nrfu.md @@ -48,12 +48,7 @@ The `text` subcommand provides a straightforward text report for each test execu ### Command overview ```bash -Usage: anta nrfu text [OPTIONS] - - ANTA command to check network states with text result. - -Options: - --help Show this message and exit. +--8<-- "anta_nrfu_text_help.txt" ``` ### Example @@ -71,13 +66,7 @@ The `table` command under the `anta nrfu` namespace offers a clear and organized ### Command overview ```bash -Usage: anta nrfu table [OPTIONS] - - ANTA command to check network states with table result. - -Options: - --group-by [device|test] Group result by test or device. - --help Show this message and exit. +--8<-- "anta_nrfu_table_help.txt" ``` The `--group-by` option show a summarized view of the test results per host or per test. @@ -125,15 +114,7 @@ The JSON rendering command in NRFU testing will generate an output of all test r ### Command overview ```bash -anta nrfu json --help -Usage: anta nrfu json [OPTIONS] - - ANTA command to check network state with JSON result. - -Options: - -o, --output FILE Path to save report as a JSON file [env var: - ANTA_NRFU_JSON_OUTPUT] - --help Show this message and exit. +--8<-- "anta_nrfu_json_help.txt" ``` The `--output` option allows you to save the JSON report as a file. If specified, no output will be displayed in the terminal. This is useful for further processing or integration with other tools. @@ -153,15 +134,7 @@ The `csv` command in NRFU testing is useful for generating a CSV file with all t ### Command overview ```bash -anta nrfu csv --help -Usage: anta nrfu csv [OPTIONS] - - ANTA command to check network states with CSV result. - -Options: - --csv-output FILE Path to save report as a CSV file [env var: - ANTA_NRFU_CSV_CSV_OUTPUT] - --help Show this message and exit. +--8<-- "anta_nrfu_csv_help.txt" ``` ### Example @@ -175,16 +148,7 @@ The `md-report` command in NRFU testing generates a comprehensive Markdown repor ### Command overview ```bash -anta nrfu md-report --help - -Usage: anta nrfu md-report [OPTIONS] - - ANTA command to check network state with Markdown report. - -Options: - --md-output FILE Path to save the report as a Markdown file [env var: - ANTA_NRFU_MD_REPORT_MD_OUTPUT; required] - --help Show this message and exit. +--8<-- "anta_nrfu_mdreport_help.txt" ``` ### Example @@ -198,17 +162,7 @@ ANTA offers a CLI option for creating custom reports. This leverages the Jinja2 ### Command overview ```bash -anta nrfu tpl-report --help -Usage: anta nrfu tpl-report [OPTIONS] - - ANTA command to check network state with templated report - -Options: - -tpl, --template FILE Path to the template to use for the report [env var: - ANTA_NRFU_TPL_REPORT_TEMPLATE; required] - -o, --output FILE Path to save report as a file [env var: - ANTA_NRFU_TPL_REPORT_OUTPUT] - --help Show this message and exit. +--8<-- "anta_nrfu_tplreport_help.txt" ``` The `--template` option is used to specify the Jinja2 template file for generating the custom report. diff --git a/docs/getting-started.md b/docs/getting-started.md index b36ea74c7..878e04be7 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -86,12 +86,14 @@ This entrypoint has multiple options to manage test coverage and reporting. --8<-- "anta_help.txt" ``` +To run the NRFU, you need to select an output format amongst [`csv`, `json`, `md-report`, `table`, `text`, `tpl-report`]. + +For a first usage, `table` is recommended. By default all test results for all devices are rendered but it can be changed to a report per test case or per host + ```bash --8<-- "anta_nrfu_help.txt" ``` -To run the NRFU, you need to select an output format amongst ["json", "table", "text", "tpl-report"]. For a first usage, `table` is recommended. By default all test results for all devices are rendered but it can be changed to a report per test case or per host - !!! Note The following examples shows how to pass all the CLI options. diff --git a/docs/imgs/anta_debug_help.svg b/docs/imgs/anta_debug_help.svg new file mode 100644 index 000000000..7c8f271ca --- /dev/null +++ b/docs/imgs/anta_debug_help.svg @@ -0,0 +1,100 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta debug --help + + + + + + + + + + Usage: anta debug [OPTIONS] COMMAND [ARGS]... + +  Commands to execute EOS commands on remote devices. + +Options: +  --help  Show this message and exit. + +Commands: +  run-cmd       Run arbitrary command to an ANTA device. +  run-template  Run arbitrary templated command to an ANTA device. + + + + + diff --git a/docs/imgs/anta_help.svg b/docs/imgs/anta_help.svg new file mode 100644 index 000000000..8a8f8f37b --- /dev/null +++ b/docs/imgs/anta_help.svg @@ -0,0 +1,139 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta --help + + + + + + + + + Usage: anta [OPTIONS] COMMAND [ARGS]... + +   Arista Network Test Automation (ANTA) CLI. + + Options: +   --help                          Show this message and exit. +   --version                       Show the version and exit. +   --log-file FILE                 Send the logs to a file. If logging level is +                                   DEBUG, only INFO or higher will be sent to +                                   stdout.  [env var: ANTA_LOG_FILE] +   -l, --log-level [CRITICAL|ERROR|WARNING|INFO|DEBUG] +                                   ANTA logging level  [env var: +                                   ANTA_LOG_LEVEL; default: INFO] + + Commands: +   check  Commands to validate configuration files. +   debug  Commands to execute EOS commands on remote devices. +   exec   Commands to execute various scripts on EOS devices. +   get    Commands to get information from or generate inventories. +   nrfu   Run ANTA tests on selected inventory devices. + + + + + diff --git a/docs/imgs/anta_nrfu_csv_help.svg b/docs/imgs/anta_nrfu_csv_help.svg new file mode 100644 index 000000000..8657d5538 --- /dev/null +++ b/docs/imgs/anta_nrfu_csv_help.svg @@ -0,0 +1,91 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta nrfu csv --help + + + + + + + + + + Usage: anta nrfu csv [OPTIONS] + +  ANTA command to check network states with CSV result. + +Options: +  --csv-output FILE  Path to save report as a CSV file  [env var: +                     ANTA_NRFU_CSV_CSV_OUTPUT] +  --help             Show this message and exit. + + + + + diff --git a/docs/imgs/anta_nrfu_help.svg b/docs/imgs/anta_nrfu_help.svg new file mode 100644 index 000000000..d687be309 --- /dev/null +++ b/docs/imgs/anta_nrfu_help.svg @@ -0,0 +1,303 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta nrfu --help + + + + + + + + + + Usage: anta nrfu [OPTIONS] COMMAND [ARGS]... + +  Run ANTA tests on selected inventory devices. + +Options: +  -u, --username TEXT             Username to connect to EOS  [env var: +                                  ANTA_USERNAME; required] +  -p, --password TEXT             Password to connect to EOS that must be +                                  provided. It can be prompted using '-- +                                  prompt' option.  [env var: ANTA_PASSWORD] +  --enable-password TEXT          Password to access EOS Privileged EXEC mode. +                                  It can be prompted using '--prompt' option. +                                  Requires '--enable' option.  [env var: +                                  ANTA_ENABLE_PASSWORD] +  --enable                        Some commands may require EOS Privileged +                                  EXEC mode. This option tries to access this +                                  mode before sending a command to the device. +[env var: ANTA_ENABLE] +  -P, --prompt                    Prompt for passwords if they are not +                                  provided.  [env var: ANTA_PROMPT] +  --timeout FLOAT                 Global API timeout. This value will be used +                                  for all devices.  [env var: ANTA_TIMEOUT; +                                  default: 30.0] +  --insecure                      Disable SSH Host Key validation.  [env var: +                                  ANTA_INSECURE] +  --disable-cache                 Disable cache globally.  [env var: +                                  ANTA_DISABLE_CACHE] +  -i, --inventory FILE            Path to the inventory YAML file.  [env var: +                                  ANTA_INVENTORY; required] +  --tags TEXT                     List of tags using comma as separator: +                                  tag1,tag2,tag3.  [env var: ANTA_TAGS] +  -c, --catalog FILE              Path to the test catalog file  [env var: +                                  ANTA_CATALOG; required] +  --catalog-format [yaml|json]    Format of the catalog file, either 'yaml' or +'json'[env var: ANTA_CATALOG_FORMAT] +  -d, --device TEXT               Run tests on a specific device. Can be +                                  provided multiple times. +  -t, --test TEXT                 Run a specific test. Can be provided +                                  multiple times. +  --ignore-status                 Exit code will always be 0.  [env var: +                                  ANTA_NRFU_IGNORE_STATUS] +  --ignore-error                  Exit code will be 0 if all tests succeeded +                                  or 1 if any test failed.  [env var: +                                  ANTA_NRFU_IGNORE_ERROR] +  --hide [success|failure|error|skipped] +                                  Hide results by type: success / failure / +                                  error / skipped'. +  --dry-run                       Run anta nrfu command but stop before +                                  starting to execute the tests. Considers all +                                  devices as connected.  [env var: +                                  ANTA_NRFU_DRY_RUN] +  --help                          Show this message and exit. + +Commands: +  csv         ANTA command to check network states with CSV result. +  json        ANTA command to check network state with JSON results. +  md-report   ANTA command to check network state with Markdown report. +  table       ANTA command to check network state with table results. +  text        ANTA command to check network state with text results. +  tpl-report  ANTA command to check network state with templated report. + + + + + diff --git a/docs/imgs/anta_nrfu_json_help.svg b/docs/imgs/anta_nrfu_json_help.svg new file mode 100644 index 000000000..f546ace15 --- /dev/null +++ b/docs/imgs/anta_nrfu_json_help.svg @@ -0,0 +1,91 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta nrfu json --help + + + + + + + + + + Usage: anta nrfu json [OPTIONS] + +  ANTA command to check network state with JSON results. + +Options: +  -o, --output FILE  Path to save report as a JSON file  [env var: +                     ANTA_NRFU_JSON_OUTPUT] +  --help             Show this message and exit. + + + + + diff --git a/docs/imgs/anta_nrfu_mdreport_help.svg b/docs/imgs/anta_nrfu_mdreport_help.svg new file mode 100644 index 000000000..b0c3964f2 --- /dev/null +++ b/docs/imgs/anta_nrfu_mdreport_help.svg @@ -0,0 +1,91 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta nrfu md-report --help + + + + + + + + + + Usage: anta nrfu md-report [OPTIONS] + +  ANTA command to check network state with Markdown report. + +Options: +  --md-output FILE  Path to save the report as a Markdown file  [env var: +                    ANTA_NRFU_MD_REPORT_MD_OUTPUT; required] +  --help            Show this message and exit. + + + + + diff --git a/docs/imgs/anta_nrfu_table_help.svg b/docs/imgs/anta_nrfu_table_help.svg new file mode 100644 index 000000000..55448db64 --- /dev/null +++ b/docs/imgs/anta_nrfu_table_help.svg @@ -0,0 +1,87 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta nrfu table --help + + + + + + + + + + Usage: anta nrfu table [OPTIONS] + +  ANTA command to check network state with table results. + +Options: +  --group-by [device|test]  Group result by test or device. +  --help                    Show this message and exit. + + + + + diff --git a/docs/imgs/anta_nrfu_text_help.svg b/docs/imgs/anta_nrfu_text_help.svg new file mode 100644 index 000000000..c5929b9c1 --- /dev/null +++ b/docs/imgs/anta_nrfu_text_help.svg @@ -0,0 +1,83 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta nrfu text --help + + + + + + + + + + Usage: anta nrfu text [OPTIONS] + +  ANTA command to check network state with text results. + +Options: +  --help  Show this message and exit. + + + + + diff --git a/docs/imgs/anta_nrfu_tplreport_help.svg b/docs/imgs/anta_nrfu_tplreport_help.svg new file mode 100644 index 000000000..77f30a06d --- /dev/null +++ b/docs/imgs/anta_nrfu_tplreport_help.svg @@ -0,0 +1,99 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + anta nrfu tpl-report --help + + + + + + + + + + Usage: anta nrfu tpl-report [OPTIONS] + +  ANTA command to check network state with templated report. + +Options: +  -tpl, --template FILE  Path to the template to use for the report  [env var: +                         ANTA_NRFU_TPL_REPORT_TEMPLATE; required] +  -o, --output FILE      Path to save report as a file  [env var: +                         ANTA_NRFU_TPL_REPORT_OUTPUT] +  --help                 Show this message and exit. + + + + + diff --git a/docs/scripts/__init__.py b/docs/scripts/__init__.py deleted file mode 100644 index e702a5103..000000000 --- a/docs/scripts/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -# Copyright (c) 2024-2025 Arista Networks, Inc. -# Use of this source code is governed by the Apache License 2.0 -# that can be found in the LICENSE file. -"""Scripts for ANTA documentation.""" diff --git a/docs/scripts/generate_doc_snippets.py b/docs/scripts/generate_doc_snippets.py new file mode 100755 index 000000000..ccaa02ee7 --- /dev/null +++ b/docs/scripts/generate_doc_snippets.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python +# Copyright (c) 2024-2025 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Generates SVG for documentation purposes.""" + +import sys +from pathlib import Path + +# TODO: svg in another PR +from generate_snippet import main as generate_snippet + +sys.path.insert(0, str(Path(__file__).parents[2])) + +COMMANDS = [ + "anta --help", + "anta nrfu --help", + "anta nrfu csv --help", + "anta nrfu json --help", + "anta nrfu table --help", + "anta nrfu text --help", + "anta nrfu tpl-report --help", + "anta nrfu md-report --help", +] + +for command in COMMANDS: + # TODO: svg in another PR + generate_snippet(command.split(" "), output="txt") diff --git a/docs/scripts/generate_svg.py b/docs/scripts/generate_snippet.py old mode 100644 new mode 100755 similarity index 62% rename from docs/scripts/generate_svg.py rename to docs/scripts/generate_snippet.py index 0add9f1b2..da3064553 --- a/docs/scripts/generate_svg.py +++ b/docs/scripts/generate_snippet.py @@ -1,11 +1,12 @@ +#!/usr/bin/env python # Copyright (c) 2023-2025 Arista Networks, Inc. # Use of this source code is governed by the Apache License 2.0 # that can be found in the LICENSE file. -"""A script to generate svg files from anta command. +"""A script to generate svg or txt files from anta command. usage: -python generate_svg.py anta ... +python generate_snippet.py anta ... """ # This script is not a package # ruff: noqa: INP001 @@ -20,12 +21,15 @@ from contextlib import redirect_stdout, suppress from importlib import import_module from importlib.metadata import entry_points +from typing import Literal from unittest.mock import patch -from rich.console import Console from rich.logging import RichHandler +from rich.markup import escape from rich.progress import Progress +sys.path.insert(0, str(pathlib.Path(__file__).parents[2])) + from anta.cli.console import console from anta.cli.nrfu.utils import anta_progress_bar @@ -35,9 +39,6 @@ root.addHandler(r) -OUTPUT_DIR = pathlib.Path(__file__).parent.parent / "imgs" - - def custom_progress_bar() -> Progress: """Set the console of progress_bar to main anta console. @@ -50,12 +51,14 @@ def custom_progress_bar() -> Progress: return progress -if __name__ == "__main__": +def main(args: list[str], output: Literal["svg", "txt"] = "svg") -> None: + """Execute the script.""" # Sane rich size os.environ["COLUMNS"] = "120" + output_dir = pathlib.Path(__file__).parent.parent / "snippets" if output == "txt" else pathlib.Path(__file__).parent.parent / "imgs" + # stolen from https://github.com/ewels/rich-click/blob/main/src/rich_click/cli.py - args = sys.argv[1:] script_name = args[0] console_scripts = entry_points(group="console_scripts") scripts = {script.name: script for script in console_scripts} @@ -80,27 +83,32 @@ def custom_progress_bar() -> Progress: module = import_module(module_path) function = getattr(module, function_name) - # Console to captur everything - new_console = Console(record=True) - pipe = io.StringIO() console.record = True console.file = pipe - with redirect_stdout(io.StringIO()) as f: - # tweaks to record and redirect to a dummy file - - console.print(f"ant@anthill$ {' '.join(sys.argv)}") - - # Redirect stdout of the program towards another StringIO to capture help - # that is not part or anta rich console - # redirect potential progress bar output to console by patching - with patch("anta.cli.nrfu.utils.anta_progress_bar", custom_progress_bar), suppress(SystemExit): - function() + # Redirect stdout of the program towards another StringIO to capture help + # that is not part or anta rich console + # redirect potential progress bar output to console by patching + with redirect_stdout(io.StringIO()) as f, patch("anta.cli.nrfu.utils.anta_progress_bar", custom_progress_bar), suppress(SystemExit): + if output == "txt": + console.print(f"$ {' '.join(sys.argv)}") + function() if "--help" in args: - console.print(f.getvalue()) + console.print(escape(f.getvalue())) + + filename = f"{'_'.join(x.replace('/', '_').replace('-', '').replace('.', '') for x in args)}.{output}" + filename = output_dir / filename + if output == "txt": + content = console.export_text()[:-1] + with filename.open("w") as fd: + fd.write(content) + # TODO: Not using this to avoid newline console.save_text(str(filename)) + elif output == "svg": + console.save_svg(str(filename), title=" ".join(args)) - filename = f"{'_'.join(x.replace('/', '_').replace('-', '_').replace('.', '_') for x in args)}.svg" - filename = f"{OUTPUT_DIR}/{filename}" print(f"File saved at {filename}") - console.save_svg(filename, title=" ".join(args)) + + +if __name__ == "__main__": + main(sys.argv[1:], "txt") diff --git a/docs/snippets/anta_debug_help.txt b/docs/snippets/anta_debug_help.txt new file mode 100644 index 000000000..0b74be25b --- /dev/null +++ b/docs/snippets/anta_debug_help.txt @@ -0,0 +1,11 @@ +$ anta debug --help +Usage: anta debug [OPTIONS] COMMAND [ARGS]... + + Commands to execute EOS commands on remote devices. + +Options: + --help Show this message and exit. + +Commands: + run-cmd Run arbitrary command to an ANTA device. + run-template Run arbitrary templated command to an ANTA device. diff --git a/docs/snippets/anta_help.txt b/docs/snippets/anta_help.txt index 7bc37adeb..dd552fd04 100644 --- a/docs/snippets/anta_help.txt +++ b/docs/snippets/anta_help.txt @@ -1,8 +1,10 @@ +$ anta --help Usage: anta [OPTIONS] COMMAND [ARGS]... Arista Network Test Automation (ANTA) CLI. Options: + --help Show this message and exit. --version Show the version and exit. --log-file FILE Send the logs to a file. If logging level is DEBUG, only INFO or higher will be sent to @@ -10,7 +12,6 @@ Options: -l, --log-level [CRITICAL|ERROR|WARNING|INFO|DEBUG] ANTA logging level [env var: ANTA_LOG_LEVEL; default: INFO] - --help Show this message and exit. Commands: check Commands to validate configuration files. diff --git a/docs/snippets/anta_nrfu_csv_help.txt b/docs/snippets/anta_nrfu_csv_help.txt new file mode 100644 index 000000000..483b1c7d4 --- /dev/null +++ b/docs/snippets/anta_nrfu_csv_help.txt @@ -0,0 +1,9 @@ +$ anta nrfu csv --help +Usage: anta nrfu csv [OPTIONS] + + ANTA command to check network state with CSV report. + +Options: + --csv-output FILE Path to save report as a CSV file [env var: + ANTA_NRFU_CSV_CSV_OUTPUT; required] + --help Show this message and exit. diff --git a/docs/snippets/anta_nrfu_help.txt b/docs/snippets/anta_nrfu_help.txt index cb23fa7ed..5801f4e3a 100644 --- a/docs/snippets/anta_nrfu_help.txt +++ b/docs/snippets/anta_nrfu_help.txt @@ -1,3 +1,4 @@ +$ anta nrfu --help Usage: anta nrfu [OPTIONS] COMMAND [ARGS]... Run ANTA tests on selected inventory devices. diff --git a/docs/snippets/anta_nrfu_json_help.txt b/docs/snippets/anta_nrfu_json_help.txt new file mode 100644 index 000000000..6aebec9c4 --- /dev/null +++ b/docs/snippets/anta_nrfu_json_help.txt @@ -0,0 +1,11 @@ +$ anta nrfu json --help +Usage: anta nrfu json [OPTIONS] + + ANTA command to check network state with JSON results. + + If no `--output` is specified, the output is printed to stdout. + +Options: + -o, --output FILE Path to save report as a JSON file [env var: + ANTA_NRFU_JSON_OUTPUT] + --help Show this message and exit. diff --git a/docs/snippets/anta_nrfu_mdreport_help.txt b/docs/snippets/anta_nrfu_mdreport_help.txt new file mode 100644 index 000000000..0d4581190 --- /dev/null +++ b/docs/snippets/anta_nrfu_mdreport_help.txt @@ -0,0 +1,9 @@ +$ anta nrfu md-report --help +Usage: anta nrfu md-report [OPTIONS] + + ANTA command to check network state with Markdown report. + +Options: + --md-output FILE Path to save the report as a Markdown file [env var: + ANTA_NRFU_MD_REPORT_MD_OUTPUT; required] + --help Show this message and exit. diff --git a/docs/snippets/anta_nrfu_table_help.txt b/docs/snippets/anta_nrfu_table_help.txt new file mode 100644 index 000000000..9d368ab96 --- /dev/null +++ b/docs/snippets/anta_nrfu_table_help.txt @@ -0,0 +1,8 @@ +$ anta nrfu table --help +Usage: anta nrfu table [OPTIONS] + + ANTA command to check network state with table results. + +Options: + --group-by [device|test] Group result by test or device. + --help Show this message and exit. diff --git a/docs/snippets/anta_nrfu_text_help.txt b/docs/snippets/anta_nrfu_text_help.txt new file mode 100644 index 000000000..3bc587a90 --- /dev/null +++ b/docs/snippets/anta_nrfu_text_help.txt @@ -0,0 +1,7 @@ +$ anta nrfu text --help +Usage: anta nrfu text [OPTIONS] + + ANTA command to check network state with text results. + +Options: + --help Show this message and exit. diff --git a/docs/snippets/anta_nrfu_tplreport_help.txt b/docs/snippets/anta_nrfu_tplreport_help.txt new file mode 100644 index 000000000..b19bc8c19 --- /dev/null +++ b/docs/snippets/anta_nrfu_tplreport_help.txt @@ -0,0 +1,11 @@ +$ anta nrfu tpl-report --help +Usage: anta nrfu tpl-report [OPTIONS] + + ANTA command to check network state with templated report. + +Options: + -tpl, --template FILE Path to the template to use for the report [env var: + ANTA_NRFU_TPL_REPORT_TEMPLATE; required] + -o, --output FILE Path to save report as a file [env var: + ANTA_NRFU_TPL_REPORT_OUTPUT] + --help Show this message and exit. diff --git a/examples/tests.yaml b/examples/tests.yaml index f5fd3ebd0..6e5d88c66 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -130,6 +130,7 @@ anta.tests.connectivity: vrf: MGMT df_bit: True size: 100 + reachable: true - source: Management0 destination: 8.8.8.8 vrf: MGMT @@ -140,6 +141,7 @@ anta.tests.connectivity: vrf: default df_bit: True size: 100 + reachable: false anta.tests.cvx: - VerifyActiveCVXConnections: # Verifies the number of active CVX Connections. @@ -952,7 +954,7 @@ anta.tests.system: - VerifyMemoryUtilization: # Verifies whether the memory utilization is below 75%. - VerifyNTP: - # Verifies if NTP is synchronised. + # Verifies that the Network Time Protocol (NTP) is synchronized. - VerifyNTPAssociations: # Verifies the Network Time Protocol (NTP) associations. ntp_servers: diff --git a/tests/units/anta_tests/test_connectivity.py b/tests/units/anta_tests/test_connectivity.py index 86ada5dea..1f5044b04 100644 --- a/tests/units/anta_tests/test_connectivity.py +++ b/tests/units/anta_tests/test_connectivity.py @@ -45,6 +45,23 @@ ], "expected": {"result": "success"}, }, + { + "name": "success-expected-unreachable", + "test": VerifyReachability, + "eos_data": [ + { + "messages": [ + """PING 10.0.0.1 (10.0.0.1) from 10.0.0.5 : 72(100) bytes of data. + + --- 10.0.0.1 ping statistics --- + 2 packets transmitted, 0 received, 100% packet loss, time 10ms + """, + ], + }, + ], + "inputs": {"hosts": [{"destination": "10.0.0.1", "source": "10.0.0.5", "reachable": False}]}, + "expected": {"result": "success"}, + }, { "name": "success-ipv6", "test": VerifyReachability, @@ -268,6 +285,30 @@ ], "expected": {"result": "failure", "messages": ["Host: 10.0.0.1 Source: Management0 VRF: default - Unreachable"]}, }, + { + "name": "failure-expected-unreachable", + "test": VerifyReachability, + "eos_data": [ + { + "messages": [ + """PING 10.0.0.1 (10.0.0.1) from 10.0.0.5 : 72(100) bytes of data. + 80 bytes from 10.0.0.1: icmp_seq=1 ttl=64 time=0.247 ms + 80 bytes from 10.0.0.1: icmp_seq=2 ttl=64 time=0.072 ms + + --- 10.0.0.1 ping statistics --- + 2 packets transmitted, 2 received, 0% packet loss, time 0ms + rtt min/avg/max/mdev = 0.072/0.159/0.247/0.088 ms, ipg/ewma 0.370/0.225 ms + + """, + ], + }, + ], + "inputs": {"hosts": [{"destination": "10.0.0.1", "source": "10.0.0.5", "reachable": False}]}, + "expected": { + "result": "failure", + "messages": ["Host: 10.0.0.1 Source: 10.0.0.5 VRF: default - Destination is expected to be unreachable but found reachable."], + }, + }, { "name": "success", "test": VerifyLLDPNeighbors, diff --git a/tests/units/anta_tests/test_interfaces.py b/tests/units/anta_tests/test_interfaces.py index 4f4e08f3c..cdb1f24a8 100644 --- a/tests/units/anta_tests/test_interfaces.py +++ b/tests/units/anta_tests/test_interfaces.py @@ -1860,45 +1860,13 @@ "name": "Ethernet1", "lineProtocolStatus": "up", "interfaceStatus": "connected", - "mtu": 1500, - "interfaceAddressBrief": {"ipAddr": {"address": "10.1.0.0", "maskLen": 31}}, - "ipv4Routable240": False, - "ipv4Routable0": False, - "enabled": True, - "description": "P2P_LINK_TO_NW-CORE_Ethernet1", "proxyArp": True, - "localProxyArp": False, - "gratuitousArp": False, - "vrf": "default", - "urpf": "disable", - "addresslessForwarding": "isInvalid", - "directedBroadcastEnabled": False, - "maxMssIngress": 0, - "maxMssEgress": 0, }, - }, - }, - { - "interfaces": { "Ethernet2": { "name": "Ethernet2", "lineProtocolStatus": "up", "interfaceStatus": "connected", - "mtu": 1500, - "interfaceAddressBrief": {"ipAddr": {"address": "10.1.0.2", "maskLen": 31}}, - "ipv4Routable240": False, - "ipv4Routable0": False, - "enabled": True, - "description": "P2P_LINK_TO_SW-CORE_Ethernet1", "proxyArp": True, - "localProxyArp": False, - "gratuitousArp": False, - "vrf": "default", - "urpf": "disable", - "addresslessForwarding": "isInvalid", - "directedBroadcastEnabled": False, - "maxMssIngress": 0, - "maxMssEgress": 0, }, }, }, @@ -1907,7 +1875,7 @@ "expected": {"result": "success"}, }, { - "name": "failure", + "name": "failure-interface-not-found", "test": VerifyIPProxyARP, "eos_data": [ { @@ -1916,51 +1884,37 @@ "name": "Ethernet1", "lineProtocolStatus": "up", "interfaceStatus": "connected", - "mtu": 1500, - "interfaceAddressBrief": {"ipAddr": {"address": "10.1.0.0", "maskLen": 31}}, - "ipv4Routable240": False, - "ipv4Routable0": False, - "enabled": True, - "description": "P2P_LINK_TO_NW-CORE_Ethernet1", "proxyArp": True, - "localProxyArp": False, - "gratuitousArp": False, - "vrf": "default", - "urpf": "disable", - "addresslessForwarding": "isInvalid", - "directedBroadcastEnabled": False, - "maxMssIngress": 0, - "maxMssEgress": 0, }, }, }, + ], + "inputs": {"interfaces": ["Ethernet1", "Ethernet2"]}, + "expected": {"result": "failure", "messages": ["Interface: Ethernet2 - Not found"]}, + }, + { + "name": "failure", + "test": VerifyIPProxyARP, + "eos_data": [ { "interfaces": { + "Ethernet1": { + "name": "Ethernet1", + "lineProtocolStatus": "up", + "interfaceStatus": "connected", + "proxyArp": True, + }, "Ethernet2": { "name": "Ethernet2", "lineProtocolStatus": "up", "interfaceStatus": "connected", - "mtu": 1500, - "interfaceAddressBrief": {"ipAddr": {"address": "10.1.0.2", "maskLen": 31}}, - "ipv4Routable240": False, - "ipv4Routable0": False, - "enabled": True, - "description": "P2P_LINK_TO_SW-CORE_Ethernet1", "proxyArp": False, - "localProxyArp": False, - "gratuitousArp": False, - "vrf": "default", - "urpf": "disable", - "addresslessForwarding": "isInvalid", - "directedBroadcastEnabled": False, - "maxMssIngress": 0, - "maxMssEgress": 0, }, }, }, ], "inputs": {"interfaces": ["Ethernet1", "Ethernet2"]}, - "expected": {"result": "failure", "messages": ["The following interface(s) have Proxy-ARP disabled: ['Ethernet2']"]}, + "expected": {"result": "failure", "messages": ["Interface: Ethernet2 - Proxy-ARP disabled"]}, }, { "name": "success", @@ -1973,17 +1927,13 @@ "primaryIp": {"address": "172.30.11.1", "maskLen": 31}, "secondaryIpsOrderedList": [{"address": "10.10.10.1", "maskLen": 31}, {"address": "10.10.10.10", "maskLen": 31}], } - } - } - }, - { - "interfaces": { + }, "Ethernet12": { "interfaceAddress": { "primaryIp": {"address": "172.30.11.10", "maskLen": 31}, "secondaryIpsOrderedList": [{"address": "10.10.10.10", "maskLen": 31}, {"address": "10.10.10.20", "maskLen": 31}], } - } + }, } }, ], @@ -2006,17 +1956,13 @@ "primaryIp": {"address": "172.30.11.0", "maskLen": 31}, "secondaryIpsOrderedList": [], } - } - } - }, - { - "interfaces": { + }, "Ethernet12": { "interfaceAddress": { "primaryIp": {"address": "172.30.11.10", "maskLen": 31}, "secondaryIpsOrderedList": [], } - } + }, } }, ], @@ -2028,10 +1974,36 @@ }, "expected": {"result": "success"}, }, + { + "name": "failure-interface-not-found", + "test": VerifyInterfaceIPv4, + "eos_data": [ + { + "interfaces": { + "Ethernet10": { + "interfaceAddress": { + "primaryIp": {"address": "172.30.11.0", "maskLen": 31}, + "secondaryIpsOrderedList": [], + } + } + } + } + ], + "inputs": { + "interfaces": [ + {"name": "Ethernet2", "primary_ip": "172.30.11.0/31", "secondary_ips": ["10.10.10.0/31", "10.10.10.10/31"]}, + {"name": "Ethernet12", "primary_ip": "172.30.11.20/31", "secondary_ips": ["10.10.11.0/31", "10.10.11.10/31"]}, + ] + }, + "expected": { + "result": "failure", + "messages": ["Interface: Ethernet2 - Not found", "Interface: Ethernet12 - Not found"], + }, + }, { "name": "failure-not-l3-interface", "test": VerifyInterfaceIPv4, - "eos_data": [{"interfaces": {"Ethernet2": {"interfaceAddress": {}}}}, {"interfaces": {"Ethernet12": {"interfaceAddress": {}}}}], + "eos_data": [{"interfaces": {"Ethernet2": {"interfaceAddress": {}}, "Ethernet12": {"interfaceAddress": {}}}}], "inputs": { "interfaces": [ {"name": "Ethernet2", "primary_ip": "172.30.11.0/31", "secondary_ips": ["10.10.10.0/31", "10.10.10.10/31"]}, @@ -2040,7 +2012,7 @@ }, "expected": { "result": "failure", - "messages": ["For interface `Ethernet2`, IP address is not configured.", "For interface `Ethernet12`, IP address is not configured."], + "messages": ["Interface: Ethernet2 - IP address is not configured", "Interface: Ethernet12 - IP address is not configured"], }, }, { @@ -2054,17 +2026,13 @@ "primaryIp": {"address": "0.0.0.0", "maskLen": 0}, "secondaryIpsOrderedList": [], } - } - } - }, - { - "interfaces": { + }, "Ethernet12": { "interfaceAddress": { "primaryIp": {"address": "0.0.0.0", "maskLen": 0}, "secondaryIpsOrderedList": [], } - } + }, } }, ], @@ -2077,10 +2045,10 @@ "expected": { "result": "failure", "messages": [ - "For interface `Ethernet2`, The expected primary IP address is `172.30.11.0/31`, but the actual primary IP address is `0.0.0.0/0`. " - "The expected secondary IP addresses are `['10.10.10.0/31', '10.10.10.10/31']`, but the actual secondary IP address is not configured.", - "For interface `Ethernet12`, The expected primary IP address is `172.30.11.10/31`, but the actual primary IP address is `0.0.0.0/0`. " - "The expected secondary IP addresses are `['10.10.11.0/31', '10.10.11.10/31']`, but the actual secondary IP address is not configured.", + "Interface: Ethernet2 - IP address mismatch - Expected: 172.30.11.0/31 Actual: 0.0.0.0/0", + "Interface: Ethernet2 - Secondary IP address is not configured", + "Interface: Ethernet12 - IP address mismatch - Expected: 172.30.11.10/31 Actual: 0.0.0.0/0", + "Interface: Ethernet12 - Secondary IP address is not configured", ], }, }, @@ -2095,17 +2063,13 @@ "primaryIp": {"address": "172.30.11.0", "maskLen": 31}, "secondaryIpsOrderedList": [{"address": "10.10.10.0", "maskLen": 31}, {"address": "10.10.10.10", "maskLen": 31}], } - } - } - }, - { - "interfaces": { + }, "Ethernet3": { "interfaceAddress": { "primaryIp": {"address": "172.30.10.10", "maskLen": 31}, "secondaryIpsOrderedList": [{"address": "10.10.11.0", "maskLen": 31}, {"address": "10.11.11.10", "maskLen": 31}], } - } + }, } }, ], @@ -2118,12 +2082,10 @@ "expected": { "result": "failure", "messages": [ - "For interface `Ethernet2`, The expected primary IP address is `172.30.11.2/31`, but the actual primary IP address is `172.30.11.0/31`. " - "The expected secondary IP addresses are `['10.10.10.20/31', '10.10.10.30/31']`, but the actual secondary IP addresses are " - "`['10.10.10.0/31', '10.10.10.10/31']`.", - "For interface `Ethernet3`, The expected primary IP address is `172.30.10.2/31`, but the actual primary IP address is `172.30.10.10/31`. " - "The expected secondary IP addresses are `['10.10.11.0/31', '10.10.11.10/31']`, but the actual secondary IP addresses are " - "`['10.10.11.0/31', '10.11.11.10/31']`.", + "Interface: Ethernet2 - IP address mismatch - Expected: 172.30.11.2/31 Actual: 172.30.11.0/31", + "Interface: Ethernet2 - Secondary IP address mismatch - Expected: 10.10.10.20/31, 10.10.10.30/31 Actual: 10.10.10.0/31, 10.10.10.10/31", + "Interface: Ethernet3 - IP address mismatch - Expected: 172.30.10.2/31 Actual: 172.30.10.10/31", + "Interface: Ethernet3 - Secondary IP address mismatch - Expected: 10.10.11.0/31, 10.10.11.10/31 Actual: 10.10.11.0/31, 10.11.11.10/31", ], }, }, @@ -2138,17 +2100,13 @@ "primaryIp": {"address": "172.30.11.0", "maskLen": 31}, "secondaryIpsOrderedList": [], } - } - } - }, - { - "interfaces": { + }, "Ethernet3": { "interfaceAddress": { "primaryIp": {"address": "172.30.10.10", "maskLen": 31}, "secondaryIpsOrderedList": [{"address": "10.10.11.0", "maskLen": 31}, {"address": "10.11.11.10", "maskLen": 31}], } - } + }, } }, ], @@ -2161,11 +2119,10 @@ "expected": { "result": "failure", "messages": [ - "For interface `Ethernet2`, The expected primary IP address is `172.30.11.2/31`, but the actual primary IP address is `172.30.11.0/31`. " - "The expected secondary IP addresses are `['10.10.10.20/31', '10.10.10.30/31']`, but the actual secondary IP address is not configured.", - "For interface `Ethernet3`, The expected primary IP address is `172.30.10.2/31`, but the actual primary IP address is `172.30.10.10/31`. " - "The expected secondary IP addresses are `['10.10.11.0/31', '10.10.11.10/31']`, but the actual secondary IP addresses are " - "`['10.10.11.0/31', '10.11.11.10/31']`.", + "Interface: Ethernet2 - IP address mismatch - Expected: 172.30.11.2/31 Actual: 172.30.11.0/31", + "Interface: Ethernet2 - Secondary IP address is not configured", + "Interface: Ethernet3 - IP address mismatch - Expected: 172.30.10.2/31 Actual: 172.30.10.10/31", + "Interface: Ethernet3 - Secondary IP address mismatch - Expected: 10.10.11.0/31, 10.10.11.10/31 Actual: 10.10.11.0/31, 10.11.11.10/31", ], }, }, diff --git a/tests/units/anta_tests/test_mlag.py b/tests/units/anta_tests/test_mlag.py index 387c88979..2a0cd25dc 100644 --- a/tests/units/anta_tests/test_mlag.py +++ b/tests/units/anta_tests/test_mlag.py @@ -30,13 +30,33 @@ "expected": {"result": "skipped", "messages": ["MLAG is disabled"]}, }, { - "name": "failure", + "name": "failure-negotiation-status", + "test": VerifyMlagStatus, + "eos_data": [{"state": "active", "negStatus": "connecting", "peerLinkStatus": "up", "localIntfStatus": "up"}], + "inputs": None, + "expected": { + "result": "failure", + "messages": ["MLAG negotiation status mismatch - Expected: connected Actual: connecting"], + }, + }, + { + "name": "failure-local-interface", + "test": VerifyMlagStatus, + "eos_data": [{"state": "active", "negStatus": "connected", "peerLinkStatus": "up", "localIntfStatus": "down"}], + "inputs": None, + "expected": { + "result": "failure", + "messages": ["Operational state of the MLAG local interface is not correct - Expected: up Actual: down"], + }, + }, + { + "name": "failure-peer-link", "test": VerifyMlagStatus, "eos_data": [{"state": "active", "negStatus": "connected", "peerLinkStatus": "down", "localIntfStatus": "up"}], "inputs": None, "expected": { "result": "failure", - "messages": ["MLAG status is not OK: {'state': 'active', 'negStatus': 'connected', 'localIntfStatus': 'up', 'peerLinkStatus': 'down'}"], + "messages": ["Operational state of the MLAG peer link is not correct - Expected: up Actual: down"], }, }, { @@ -74,7 +94,7 @@ "inputs": None, "expected": { "result": "failure", - "messages": ["MLAG status is not OK: {'Disabled': 0, 'Configured': 0, 'Inactive': 0, 'Active-partial': 1, 'Active-full': 1}"], + "messages": ["MLAG status is not ok - Inactive Ports: 0 Partial Active Ports: 1"], }, }, { @@ -89,7 +109,7 @@ "inputs": None, "expected": { "result": "failure", - "messages": ["MLAG status is not OK: {'Disabled': 0, 'Configured': 0, 'Inactive': 1, 'Active-partial': 1, 'Active-full': 1}"], + "messages": ["MLAG status is not ok - Inactive Ports: 1 Partial Active Ports: 1"], }, }, { @@ -124,12 +144,7 @@ "inputs": None, "expected": { "result": "failure", - "messages": [ - "MLAG config-sanity returned inconsistencies: " - "{'globalConfiguration': {'mlag': {'globalParameters': " - "{'dual-primary-detection-delay': {'localValue': '0', 'peerValue': '200'}}}}, " - "'interfaceConfiguration': {}}", - ], + "messages": ["MLAG config-sanity found in global configuration"], }, }, { @@ -146,12 +161,7 @@ "inputs": None, "expected": { "result": "failure", - "messages": [ - "MLAG config-sanity returned inconsistencies: " - "{'globalConfiguration': {}, " - "'interfaceConfiguration': {'trunk-native-vlan mlag30': " - "{'interface': {'Port-Channel30': {'localValue': '123', 'peerValue': '3700'}}}}}", - ], + "messages": ["MLAG config-sanity found in interface configuration"], }, }, { @@ -177,7 +187,10 @@ "test": VerifyMlagReloadDelay, "eos_data": [{"state": "active", "reloadDelay": 400, "reloadDelayNonMlag": 430}], "inputs": {"reload_delay": 300, "reload_delay_non_mlag": 330}, - "expected": {"result": "failure", "messages": ["The reload-delay parameters are not configured properly: {'reloadDelay': 400, 'reloadDelayNonMlag': 430}"]}, + "expected": { + "result": "failure", + "messages": ["MLAG reload-delay mismatch - Expected: 300s Actual: 400s", "Delay for non-MLAG ports mismatch - Expected: 330s Actual: 430s"], + }, }, { "name": "success", @@ -236,13 +249,8 @@ "expected": { "result": "failure", "messages": [ - ( - "The dual-primary parameters are not configured properly: " - "{'detail.dualPrimaryDetectionDelay': 300, " - "'detail.dualPrimaryAction': 'none', " - "'dualPrimaryMlagRecoveryDelay': 160, " - "'dualPrimaryNonMlagRecoveryDelay': 0}" - ), + "Dual-primary detection delay mismatch - Expected: 200 Actual: 300", + "Dual-primary MLAG recovery delay mismatch - Expected: 60 Actual: 160", ], }, }, @@ -262,15 +270,26 @@ "inputs": {"detection_delay": 200, "errdisabled": True, "recovery_delay": 60, "recovery_delay_non_mlag": 0}, "expected": { "result": "failure", - "messages": [ - ( - "The dual-primary parameters are not configured properly: " - "{'detail.dualPrimaryDetectionDelay': 200, " - "'detail.dualPrimaryAction': 'none', " - "'dualPrimaryMlagRecoveryDelay': 60, " - "'dualPrimaryNonMlagRecoveryDelay': 0}" - ), - ], + "messages": ["Dual-primary action mismatch - Expected: errdisableAllInterfaces Actual: none"], + }, + }, + { + "name": "failure-wrong-non-mlag-delay", + "test": VerifyMlagDualPrimary, + "eos_data": [ + { + "state": "active", + "dualPrimaryDetectionState": "configured", + "dualPrimaryPortsErrdisabled": False, + "dualPrimaryMlagRecoveryDelay": 60, + "dualPrimaryNonMlagRecoveryDelay": 120, + "detail": {"dualPrimaryDetectionDelay": 200, "dualPrimaryAction": "errdisableAllInterfaces"}, + }, + ], + "inputs": {"detection_delay": 200, "errdisabled": True, "recovery_delay": 60, "recovery_delay_non_mlag": 60}, + "expected": { + "result": "failure", + "messages": ["Dual-primary non MLAG recovery delay mismatch - Expected: 60 Actual: 120"], }, }, { @@ -325,7 +344,7 @@ "inputs": {"primary_priority": 1}, "expected": { "result": "failure", - "messages": ["The device is not set as MLAG primary.", "The primary priority does not match expected. Expected `1`, but found `32767` instead."], + "messages": ["The device is not set as MLAG primary.", "MLAG primary priority mismatch - Expected: 1 Actual: 32767"], }, }, ] diff --git a/tests/units/anta_tests/test_stp.py b/tests/units/anta_tests/test_stp.py index 5de5df468..3dbd8c5d0 100644 --- a/tests/units/anta_tests/test_stp.py +++ b/tests/units/anta_tests/test_stp.py @@ -37,7 +37,7 @@ {"spanningTreeVlanInstances": {}}, ], "inputs": {"mode": "rstp", "vlans": [10, 20]}, - "expected": {"result": "failure", "messages": ["STP mode 'rstp' not configured for the following VLAN(s): [10, 20]"]}, + "expected": {"result": "failure", "messages": ["VLAN 10 STP mode: rstp - Not configured", "VLAN 20 STP mode: rstp - Not configured"]}, }, { "name": "failure-wrong-mode", @@ -47,7 +47,10 @@ {"spanningTreeVlanInstances": {"20": {"spanningTreeVlanInstance": {"protocol": "mstp"}}}}, ], "inputs": {"mode": "rstp", "vlans": [10, 20]}, - "expected": {"result": "failure", "messages": ["Wrong STP mode configured for the following VLAN(s): [10, 20]"]}, + "expected": { + "result": "failure", + "messages": ["VLAN 10 - Incorrect STP mode - Expected: rstp Actual: mstp", "VLAN 20 - Incorrect STP mode - Expected: rstp Actual: mstp"], + }, }, { "name": "failure-both", @@ -59,7 +62,7 @@ "inputs": {"mode": "rstp", "vlans": [10, 20]}, "expected": { "result": "failure", - "messages": ["STP mode 'rstp' not configured for the following VLAN(s): [10]", "Wrong STP mode configured for the following VLAN(s): [20]"], + "messages": ["VLAN 10 STP mode: rstp - Not configured", "VLAN 20 - Incorrect STP mode - Expected: rstp Actual: mstp"], }, }, { @@ -74,7 +77,7 @@ "test": VerifySTPBlockedPorts, "eos_data": [{"spanningTreeInstances": {"MST0": {"spanningTreeBlockedPorts": ["Ethernet10"]}, "MST10": {"spanningTreeBlockedPorts": ["Ethernet10"]}}}], "inputs": None, - "expected": {"result": "failure", "messages": ["The following ports are blocked by STP: {'MST0': ['Ethernet10'], 'MST10': ['Ethernet10']}"]}, + "expected": {"result": "failure", "messages": ["STP Instance: MST0 - Blocked ports - Ethernet10", "STP Instance: MST10 - Blocked ports - Ethernet10"]}, }, { "name": "success", @@ -84,18 +87,44 @@ "expected": {"result": "success"}, }, { - "name": "failure", + "name": "failure-bpdu-tagged-error-mismatch", "test": VerifySTPCounters, "eos_data": [ { "interfaces": { "Ethernet10": {"bpduSent": 201, "bpduReceived": 0, "bpduTaggedError": 3, "bpduOtherError": 0, "bpduRateLimitCount": 0}, + "Ethernet11": {"bpduSent": 99, "bpduReceived": 0, "bpduTaggedError": 3, "bpduOtherError": 0, "bpduRateLimitCount": 0}, + }, + }, + ], + "inputs": None, + "expected": { + "result": "failure", + "messages": [ + "Interface Ethernet10 - STP BPDU packet tagged errors count mismatch - Expected: 0 Actual: 3", + "Interface Ethernet11 - STP BPDU packet tagged errors count mismatch - Expected: 0 Actual: 3", + ], + }, + }, + { + "name": "failure-bpdu-other-error-mismatch", + "test": VerifySTPCounters, + "eos_data": [ + { + "interfaces": { + "Ethernet10": {"bpduSent": 201, "bpduReceived": 0, "bpduTaggedError": 0, "bpduOtherError": 3, "bpduRateLimitCount": 0}, "Ethernet11": {"bpduSent": 99, "bpduReceived": 0, "bpduTaggedError": 0, "bpduOtherError": 6, "bpduRateLimitCount": 0}, }, }, ], "inputs": None, - "expected": {"result": "failure", "messages": ["The following interfaces have STP BPDU packet errors: ['Ethernet10', 'Ethernet11']"]}, + "expected": { + "result": "failure", + "messages": [ + "Interface Ethernet10 - STP BPDU packet other errors count mismatch - Expected: 0 Actual: 3", + "Interface Ethernet11 - STP BPDU packet other errors count mismatch - Expected: 0 Actual: 6", + ], + }, }, { "name": "success", @@ -134,7 +163,7 @@ "test": VerifySTPForwardingPorts, "eos_data": [{"unmappedVlans": [], "topologies": {}}, {"unmappedVlans": [], "topologies": {}}], "inputs": {"vlans": [10, 20]}, - "expected": {"result": "failure", "messages": ["STP instance is not configured for the following VLAN(s): [10, 20]"]}, + "expected": {"result": "failure", "messages": ["VLAN 10 - STP instance is not configured", "VLAN 20 - STP instance is not configured"]}, }, { "name": "failure", @@ -152,7 +181,10 @@ "inputs": {"vlans": [10, 20]}, "expected": { "result": "failure", - "messages": ["The following VLAN(s) have interface(s) that are not in a forwarding state: [{'VLAN 10': ['Ethernet10']}, {'VLAN 20': ['Ethernet10']}]"], + "messages": [ + "VLAN 10 Interface: Ethernet10 - Invalid state - Expected: forwarding Actual: discarding", + "VLAN 20 Interface: Ethernet10 - Invalid state - Expected: forwarding Actual: discarding", + ], }, }, { @@ -261,6 +293,28 @@ "inputs": {"priority": 16384, "instances": [0]}, "expected": {"result": "success"}, }, + { + "name": "success-input-instance-none", + "test": VerifySTPRootPriority, + "eos_data": [ + { + "instances": { + "MST0": { + "rootBridge": { + "priority": 16384, + "systemIdExtension": 0, + "macAddress": "02:1c:73:8b:93:ac", + "helloTime": 2.0, + "maxAge": 20, + "forwardDelay": 15, + }, + }, + }, + }, + ], + "inputs": {"priority": 16384}, + "expected": {"result": "success"}, + }, { "name": "failure-no-instances", "test": VerifySTPRootPriority, @@ -281,7 +335,7 @@ }, ], "inputs": {"priority": 32768, "instances": [0]}, - "expected": {"result": "failure", "messages": ["Unsupported STP instance type: WRONG0"]}, + "expected": {"result": "failure", "messages": ["STP Instance: WRONG0 - Unsupported STP instance type"]}, }, { "name": "failure-wrong-instance-type", @@ -290,6 +344,28 @@ "inputs": {"priority": 32768, "instances": [10, 20]}, "expected": {"result": "failure", "messages": ["No STP instances configured"]}, }, + { + "name": "failure-instance-not-found", + "test": VerifySTPRootPriority, + "eos_data": [ + { + "instances": { + "VL10": { + "rootBridge": { + "priority": 32768, + "systemIdExtension": 10, + "macAddress": "00:1c:73:27:95:a2", + "helloTime": 2.0, + "maxAge": 20, + "forwardDelay": 15, + }, + } + } + } + ], + "inputs": {"priority": 32768, "instances": [11, 20]}, + "expected": {"result": "failure", "messages": ["Instance: VL11 - Not configured", "Instance: VL20 - Not configured"]}, + }, { "name": "failure-wrong-priority", "test": VerifySTPRootPriority, @@ -330,7 +406,13 @@ }, ], "inputs": {"priority": 32768, "instances": [10, 20, 30]}, - "expected": {"result": "failure", "messages": ["The following instance(s) have the wrong STP root priority configured: ['VL20', 'VL30']"]}, + "expected": { + "result": "failure", + "messages": [ + "STP Instance: VL20 - Incorrect root priority - Expected: 32768 Actual: 8196", + "STP Instance: VL30 - Incorrect root priority - Expected: 32768 Actual: 8196", + ], + }, }, { "name": "success-mstp", @@ -470,8 +552,8 @@ "expected": { "result": "failure", "messages": [ - "The following STP topologies are not configured or number of changes not within the threshold:\n" - "{'topologies': {'Cist': {'Cpu': {'Number of changes': 15}, 'Port-Channel5': {'Number of changes': 15}}}}" + "Topology: Cist Interface: Cpu - Number of changes not within the threshold - Expected: 10 Actual: 15", + "Topology: Cist Interface: Port-Channel5 - Number of changes not within the threshold - Expected: 10 Actual: 15", ], }, }, diff --git a/tests/units/anta_tests/test_system.py b/tests/units/anta_tests/test_system.py index 858b793d1..5fe0fbad1 100644 --- a/tests/units/anta_tests/test_system.py +++ b/tests/units/anta_tests/test_system.py @@ -33,7 +33,7 @@ "test": VerifyUptime, "eos_data": [{"upTime": 665.15, "loadAvg": [0.13, 0.12, 0.09], "users": 1, "currentTime": 1683186659.139859}], "inputs": {"minimum": 666}, - "expected": {"result": "failure", "messages": ["Device uptime is 665.15 seconds"]}, + "expected": {"result": "failure", "messages": ["Device uptime is incorrect - Expected: 666 Actual: 665.15 seconds"]}, }, { "name": "success-no-reload", @@ -74,7 +74,7 @@ }, ], "inputs": None, - "expected": {"result": "failure", "messages": ["Reload cause is: 'Reload after crash.'"]}, + "expected": {"result": "failure", "messages": ["Reload cause is: Reload after crash."]}, }, { "name": "success-without-minidump", @@ -95,14 +95,14 @@ "test": VerifyCoredump, "eos_data": [{"mode": "compressedDeferred", "coreFiles": ["core.2344.1584483862.Mlag.gz", "core.23101.1584483867.Mlag.gz"]}], "inputs": None, - "expected": {"result": "failure", "messages": ["Core dump(s) have been found: ['core.2344.1584483862.Mlag.gz', 'core.23101.1584483867.Mlag.gz']"]}, + "expected": {"result": "failure", "messages": ["Core dump(s) have been found: core.2344.1584483862.Mlag.gz, core.23101.1584483867.Mlag.gz"]}, }, { "name": "failure-with-minidump", "test": VerifyCoredump, "eos_data": [{"mode": "compressedDeferred", "coreFiles": ["minidump", "core.2344.1584483862.Mlag.gz", "core.23101.1584483867.Mlag.gz"]}], "inputs": None, - "expected": {"result": "failure", "messages": ["Core dump(s) have been found: ['core.2344.1584483862.Mlag.gz', 'core.23101.1584483867.Mlag.gz']"]}, + "expected": {"result": "failure", "messages": ["Core dump(s) have been found: core.2344.1584483862.Mlag.gz, core.23101.1584483867.Mlag.gz"]}, }, { "name": "success", @@ -190,7 +190,7 @@ }, ], "inputs": None, - "expected": {"result": "failure", "messages": ["Device has reported a high CPU utilization: 75.2%"]}, + "expected": {"result": "failure", "messages": ["Device has reported a high CPU utilization - Expected: < 75% Actual: 75.2%"]}, }, { "name": "success", @@ -222,7 +222,7 @@ }, ], "inputs": None, - "expected": {"result": "failure", "messages": ["Device has reported a high memory usage: 95.56%"]}, + "expected": {"result": "failure", "messages": ["Device has reported a high memory usage - Expected: < 75% Actual: 95.56%"]}, }, { "name": "success", @@ -253,8 +253,8 @@ "expected": { "result": "failure", "messages": [ - "Mount point /dev/sda2 3.9G 988M 2.9G 84% /mnt/flash is higher than 75%: reported 84%", - "Mount point none 294M 78M 217M 84% /.overlay is higher than 75%: reported 84%", + "Mount point: /dev/sda2 3.9G 988M 2.9G 84% /mnt/flash - Higher disk space utilization - Expected: 75% Actual: 84%", + "Mount point: none 294M 78M 217M 84% /.overlay - Higher disk space utilization - Expected: 75% Actual: 84%", ], }, }, @@ -278,7 +278,7 @@ """, ], "inputs": None, - "expected": {"result": "failure", "messages": ["The device is not synchronized with the configured NTP server(s): 'unsynchronised'"]}, + "expected": {"result": "failure", "messages": ["NTP status mismatch - Expected: synchronised Actual: unsynchronised"]}, }, { "name": "success", @@ -413,9 +413,9 @@ "expected": { "result": "failure", "messages": [ - "1.1.1.1 (Preferred: True, Stratum: 1) - Bad association - Condition: candidate, Stratum: 2", - "2.2.2.2 (Preferred: False, Stratum: 2) - Bad association - Condition: sys.peer, Stratum: 2", - "3.3.3.3 (Preferred: False, Stratum: 2) - Bad association - Condition: sys.peer, Stratum: 3", + "NTP Server: 1.1.1.1 Preferred: True Stratum: 1 - Bad association - Condition: candidate, Stratum: 2", + "NTP Server: 2.2.2.2 Preferred: False Stratum: 2 - Bad association - Condition: sys.peer, Stratum: 2", + "NTP Server: 3.3.3.3 Preferred: False Stratum: 2 - Bad association - Condition: sys.peer, Stratum: 3", ], }, }, @@ -463,7 +463,7 @@ }, "expected": { "result": "failure", - "messages": ["3.3.3.3 (Preferred: False, Stratum: 1) - Not configured"], + "messages": ["NTP Server: 3.3.3.3 Preferred: False Stratum: 1 - Not configured"], }, }, { @@ -490,9 +490,9 @@ "expected": { "result": "failure", "messages": [ - "1.1.1.1 (Preferred: True, Stratum: 1) - Bad association - Condition: candidate, Stratum: 1", - "2.2.2.2 (Preferred: False, Stratum: 1) - Not configured", - "3.3.3.3 (Preferred: False, Stratum: 1) - Not configured", + "NTP Server: 1.1.1.1 Preferred: True Stratum: 1 - Bad association - Condition: candidate, Stratum: 1", + "NTP Server: 2.2.2.2 Preferred: False Stratum: 1 - Not configured", + "NTP Server: 3.3.3.3 Preferred: False Stratum: 1 - Not configured", ], }, }, diff --git a/tests/units/input_models/test_interfaces.py b/tests/units/input_models/test_interfaces.py index aefa31941..881e3bdbe 100644 --- a/tests/units/input_models/test_interfaces.py +++ b/tests/units/input_models/test_interfaces.py @@ -12,7 +12,7 @@ from pydantic import ValidationError from anta.input_models.interfaces import InterfaceState -from anta.tests.interfaces import VerifyInterfacesStatus, VerifyLACPInterfacesStatus +from anta.tests.interfaces import VerifyInterfaceIPv4, VerifyInterfacesStatus, VerifyLACPInterfacesStatus if TYPE_CHECKING: from anta.custom_types import Interface, PortChannelInterface @@ -83,3 +83,28 @@ def test_invalid(self, interfaces: list[InterfaceState]) -> None: """Test VerifyLACPInterfacesStatus.Input invalid inputs.""" with pytest.raises(ValidationError): VerifyLACPInterfacesStatus.Input(interfaces=interfaces) + + +class TestVerifyInterfaceIPv4Input: + """Test anta.tests.interfaces.VerifyInterfaceIPv4.Input.""" + + @pytest.mark.parametrize( + ("interfaces"), + [ + pytest.param([{"name": "Ethernet1", "primary_ip": "172.30.11.1/31"}], id="valid"), + ], + ) + def test_valid(self, interfaces: list[InterfaceState]) -> None: + """Test VerifyInterfaceIPv4.Input valid inputs.""" + VerifyInterfaceIPv4.Input(interfaces=interfaces) + + @pytest.mark.parametrize( + ("interfaces"), + [ + pytest.param([{"name": "Ethernet1"}], id="invalid-no-primary-ip"), + ], + ) + def test_invalid(self, interfaces: list[InterfaceState]) -> None: + """Test VerifyInterfaceIPv4.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyInterfaceIPv4.Input(interfaces=interfaces)