From 419f6ca67d093b1cf4bd665ee213e918315b0321 Mon Sep 17 00:00:00 2001 From: Guillaume Mulocher Date: Fri, 17 Jan 2025 18:07:52 +0100 Subject: [PATCH 1/5] doc: Sometimes decorators are not functions (#1012) --- anta/reporter/__init__.py | 2 +- docs/templates/python/material/class.html.jinja | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/anta/reporter/__init__.py b/anta/reporter/__init__.py index 696a499a6..5156ea7e8 100644 --- a/anta/reporter/__init__.py +++ b/anta/reporter/__init__.py @@ -28,7 +28,7 @@ class ReportTable: """TableReport Generate a Table based on TestResult.""" - @dataclass() + @dataclass class Headers: # pylint: disable=too-many-instance-attributes """Headers for the table report.""" diff --git a/docs/templates/python/material/class.html.jinja b/docs/templates/python/material/class.html.jinja index cbf9fac22..e8e3de06c 100644 --- a/docs/templates/python/material/class.html.jinja +++ b/docs/templates/python/material/class.html.jinja @@ -59,7 +59,7 @@ {{ super() }} {% for dec in class.decorators %} -{% if dec.value.function.name == "deprecated_test_class" %} +{% if dec.value.function is defined and dec.value.function.name == "deprecated_test_class" %} Static Badge {% for arg in dec.value.arguments | selectattr("name", "equalto", "removal_in_version") | list %} Static Badge From ab57f89c527babecac2382e9d7a7bb1d31328893 Mon Sep 17 00:00:00 2001 From: geetanjalimanegslab <96573243+geetanjalimanegslab@users.noreply.github.com> Date: Mon, 20 Jan 2025 20:28:54 +0530 Subject: [PATCH 2/5] fix(anta.tests): Updated Logging Module with Adding a user defined field for send cmd severity level (#1008) * Issue_922: Added user defined field for send cmd severity level * Apply suggestions from code review --------- Co-authored-by: Carl Baillargeon --- anta/custom_types.py | 1 + anta/tests/logging.py | 101 +++++++++++++++---------- examples/tests.yaml | 3 + tests/units/anta_tests/test_logging.py | 22 +++--- 4 files changed, 77 insertions(+), 50 deletions(-) diff --git a/anta/custom_types.py b/anta/custom_types.py index 3af27db97..5810620e2 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -263,3 +263,4 @@ def validate_regex(value: str) -> str: SnmpHashingAlgorithm = Literal["MD5", "SHA", "SHA-224", "SHA-256", "SHA-384", "SHA-512"] SnmpEncryptionAlgorithm = Literal["AES-128", "AES-192", "AES-256", "DES"] DynamicVlanSource = Literal["dmf", "dot1x", "dynvtep", "evpn", "mlag", "mlagsync", "mvpn", "swfwd", "vccbfd"] +LogSeverityLevel = Literal["alerts", "critical", "debugging", "emergencies", "errors", "informational", "notifications", "warnings"] diff --git a/anta/tests/logging.py b/anta/tests/logging.py index 4a1c594e8..15ec3dd4a 100644 --- a/anta/tests/logging.py +++ b/anta/tests/logging.py @@ -14,13 +14,12 @@ from ipaddress import IPv4Address from typing import TYPE_CHECKING, ClassVar -from anta.models import AntaCommand, AntaTest +from anta.custom_types import LogSeverityLevel +from anta.models import AntaCommand, AntaTemplate, AntaTest if TYPE_CHECKING: import logging - from anta.models import AntaTemplate - def _get_logging_states(logger: logging.Logger, command_output: str) -> str: """Parse `show logging` output and gets operational logging states used in the tests in this module. @@ -201,35 +200,43 @@ class VerifyLoggingLogsGeneration(AntaTest): This test performs the following checks: - 1. Sends a test log message at the **informational** level - 2. Retrieves the most recent logs (last 30 seconds) - 3. Verifies that the test message was successfully logged - - !!! warning - EOS logging buffer should be set to severity level `informational` or higher for this test to work. + 1. Sends a test log message at the specified severity log level. + 2. Retrieves the most recent logs (last 30 seconds). + 3. Verifies that the test message was successfully logged. Expected Results ---------------- * Success: If logs are being generated and the test message is found in recent logs. * Failure: If any of the following occur: - - The test message is not found in recent logs - - The logging system is not capturing new messages - - No logs are being generated + - The test message is not found in recent logs. + - The logging system is not capturing new messages. + - No logs are being generated. Examples -------- ```yaml anta.tests.logging: - VerifyLoggingLogsGeneration: + severity_level: informational ``` """ categories: ClassVar[list[str]] = ["logging"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ - AntaCommand(command="send log level informational message ANTA VerifyLoggingLogsGeneration validation", ofmt="text"), - AntaCommand(command="show logging informational last 30 seconds | grep ANTA", ofmt="text", use_cache=False), + AntaTemplate(template="send log level {severity_level} message ANTA VerifyLoggingLogsGeneration validation", ofmt="text"), + AntaTemplate(template="show logging {severity_level} last 30 seconds | grep ANTA", ofmt="text", use_cache=False), ] + class Input(AntaTest.Input): + """Input model for the VerifyLoggingLogsGeneration test.""" + + severity_level: LogSeverityLevel = "informational" + """Log severity level. Defaults to informational.""" + + def render(self, template: AntaTemplate) -> list[AntaCommand]: + """Render the template for log severity level in the input.""" + return [template.render(severity_level=self.inputs.severity_level)] + @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyLoggingLogsGeneration.""" @@ -248,37 +255,45 @@ class VerifyLoggingHostname(AntaTest): This test performs the following checks: - 1. Retrieves the device's configured FQDN - 2. Sends a test log message at the **informational** level - 3. Retrieves the most recent logs (last 30 seconds) - 4. Verifies that the test message includes the complete FQDN of the device - - !!! warning - EOS logging buffer should be set to severity level `informational` or higher for this test to work. + 1. Retrieves the device's configured FQDN. + 2. Sends a test log message at the specified severity log level. + 3. Retrieves the most recent logs (last 30 seconds). + 4. Verifies that the test message includes the complete FQDN of the device. Expected Results ---------------- * Success: If logs are generated with the device's complete FQDN. * Failure: If any of the following occur: - - The test message is not found in recent logs - - The log message does not include the device's FQDN - - The FQDN in the log message doesn't match the configured FQDN + - The test message is not found in recent logs. + - The log message does not include the device's FQDN. + - The FQDN in the log message doesn't match the configured FQDN. Examples -------- ```yaml anta.tests.logging: - VerifyLoggingHostname: + severity_level: informational ``` """ categories: ClassVar[list[str]] = ["logging"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ AntaCommand(command="show hostname", revision=1), - AntaCommand(command="send log level informational message ANTA VerifyLoggingHostname validation", ofmt="text"), - AntaCommand(command="show logging informational last 30 seconds | grep ANTA", ofmt="text", use_cache=False), + AntaTemplate(template="send log level {severity_level} message ANTA VerifyLoggingHostname validation", ofmt="text"), + AntaTemplate(template="show logging {severity_level} last 30 seconds | grep ANTA", ofmt="text", use_cache=False), ] + class Input(AntaTest.Input): + """Input model for the VerifyLoggingHostname test.""" + + severity_level: LogSeverityLevel = "informational" + """Log severity level. Defaults to informational.""" + + def render(self, template: AntaTemplate) -> list[AntaCommand]: + """Render the template for log severity level in the input.""" + return [template.render(severity_level=self.inputs.severity_level)] + @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyLoggingHostname.""" @@ -303,37 +318,45 @@ class VerifyLoggingTimestamp(AntaTest): This test performs the following checks: - 1. Sends a test log message at the **informational** level - 2. Retrieves the most recent logs (last 30 seconds) - 3. Verifies that the test message is present with a high-resolution RFC3339 timestamp format - - Example format: `2024-01-25T15:30:45.123456+00:00` - - Includes microsecond precision - - Contains timezone offset - - !!! warning - EOS logging buffer should be set to severity level `informational` or higher for this test to work. + 1. Sends a test log message at the specified severity log level. + 2. Retrieves the most recent logs (last 30 seconds). + 3. Verifies that the test message is present with a high-resolution RFC3339 timestamp format. + - Example format: `2024-01-25T15:30:45.123456+00:00`. + - Includes microsecond precision. + - Contains timezone offset. Expected Results ---------------- * Success: If logs are generated with the correct high-resolution RFC3339 timestamp format. * Failure: If any of the following occur: - - The test message is not found in recent logs - - The timestamp format does not match the expected RFC3339 format + - The test message is not found in recent logs. + - The timestamp format does not match the expected RFC3339 format. Examples -------- ```yaml anta.tests.logging: - VerifyLoggingTimestamp: + severity_level: informational ``` """ categories: ClassVar[list[str]] = ["logging"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ - AntaCommand(command="send log level informational message ANTA VerifyLoggingTimestamp validation", ofmt="text"), - AntaCommand(command="show logging informational last 30 seconds | grep ANTA", ofmt="text", use_cache=False), + AntaTemplate(template="send log level {severity_level} message ANTA VerifyLoggingTimestamp validation", ofmt="text"), + AntaTemplate(template="show logging {severity_level} last 30 seconds | grep ANTA", ofmt="text", use_cache=False), ] + class Input(AntaTest.Input): + """Input model for the VerifyLoggingTimestamp test.""" + + severity_level: LogSeverityLevel = "informational" + """Log severity level. Defaults to informational.""" + + def render(self, template: AntaTemplate) -> list[AntaCommand]: + """Render the template for log severity level in the input.""" + return [template.render(severity_level=self.inputs.severity_level)] + @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyLoggingTimestamp.""" diff --git a/examples/tests.yaml b/examples/tests.yaml index 6a3ff456e..f47285a13 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -291,6 +291,7 @@ anta.tests.logging: # Verifies there are no syslog messages with a severity of ERRORS or higher. - VerifyLoggingHostname: # Verifies if logs are generated with the device FQDN. + severity_level: informational - VerifyLoggingHosts: # Verifies logging hosts (syslog servers) for a specified VRF. hosts: @@ -299,6 +300,7 @@ anta.tests.logging: vrf: default - VerifyLoggingLogsGeneration: # Verifies if logs are generated. + severity_level: informational - VerifyLoggingPersistent: # Verifies if logging persistent is enabled and logs are saved in flash. - VerifyLoggingSourceIntf: @@ -307,6 +309,7 @@ anta.tests.logging: vrf: default - VerifyLoggingTimestamp: # Verifies if logs are generated with the appropriate timestamp. + severity_level: informational - VerifySyslogLogging: # Verifies if syslog logging is enabled. anta.tests.mlag: diff --git a/tests/units/anta_tests/test_logging.py b/tests/units/anta_tests/test_logging.py index 0c1408817..3022d4cd8 100644 --- a/tests/units/anta_tests/test_logging.py +++ b/tests/units/anta_tests/test_logging.py @@ -167,14 +167,14 @@ "2023-05-10T13:54:21.463497-05:00 NW-CORE.example.org ConfigAgent: %SYS-6-LOGMSG_INFO: " "Message from arista on command-api (10.22.1.107): ANTA VerifyLoggingLogsGeneration validation\n", ], - "inputs": None, + "inputs": {"severity_level": "informational"}, "expected": {"result": "success"}, }, { "name": "failure", "test": VerifyLoggingLogsGeneration, "eos_data": ["", "Log Buffer:\n"], - "inputs": None, + "inputs": {"severity_level": "notifications"}, "expected": {"result": "failure", "messages": ["Logs are not generated"]}, }, { @@ -186,7 +186,7 @@ "2023-05-10T15:41:44.701810-05:00 NW-CORE.example.org ConfigAgent: %SYS-6-LOGMSG_INFO: " "Message from arista on command-api (10.22.1.107): ANTA VerifyLoggingHostname validation\n", ], - "inputs": None, + "inputs": {"severity_level": "informational"}, "expected": {"result": "success"}, }, { @@ -195,10 +195,10 @@ "eos_data": [ {"hostname": "NW-CORE", "fqdn": "NW-CORE.example.org"}, "", - "2023-05-10T13:54:21.463497-05:00 NW-CORE ConfigAgent: %SYS-6-LOGMSG_INFO: " + "2023-05-10T13:54:21.463497-05:00 NW-CORE ConfigAgent: %SYS-6-LOGMSG_NOTICE: " "Message from arista on command-api (10.22.1.107): ANTA VerifyLoggingLogsHostname validation\n", ], - "inputs": None, + "inputs": {"severity_level": "notifications"}, "expected": {"result": "failure", "messages": ["Logs are not generated with the device FQDN"]}, }, { @@ -211,7 +211,7 @@ "2023-05-10T15:42:44.680813-05:00 NW-CORE.example.org ConfigAgent: %SYS-6-LOGMSG_INFO: " "Other log\n", ], - "inputs": None, + "inputs": {"severity_level": "informational"}, "expected": {"result": "success"}, }, { @@ -224,7 +224,7 @@ "2023-05-10T15:42:44.680813+05:00 NW-CORE.example.org ConfigAgent: %SYS-6-LOGMSG_INFO: " "Other log\n", ], - "inputs": None, + "inputs": {"severity_level": "informational"}, "expected": {"result": "success"}, }, { @@ -232,10 +232,10 @@ "test": VerifyLoggingTimestamp, "eos_data": [ "", - "May 10 13:54:22 NE-CORE.example.org ConfigAgent: %SYS-6-LOGMSG_INFO: " + "May 10 13:54:22 NE-CORE.example.org ConfigAgent: %SYS-6-LOGMSG_ALERT: " "Message from arista on command-api (10.22.1.107): ANTA VerifyLoggingTimestamp validation\n", ], - "inputs": None, + "inputs": {"severity_level": "alerts"}, "expected": {"result": "failure", "messages": ["Logs are not generated with the appropriate timestamp format"]}, }, { @@ -243,9 +243,9 @@ "test": VerifyLoggingTimestamp, "eos_data": [ "", - "May 10 13:54:22 NE-CORE.example.org ConfigAgent: %SYS-6-LOGMSG_INFO: Message from arista on command-api (10.22.1.107): BLAH\n", + "May 10 13:54:22 NE-CORE.example.org ConfigAgent: %SYS-6-LOGMSG_NOTICE: Message from arista on command-api (10.22.1.107): BLAH\n", ], - "inputs": None, + "inputs": {"severity_level": "notifications"}, "expected": {"result": "failure", "messages": ["Logs are not generated with the appropriate timestamp format"]}, }, { From e0545a8451eb71dcc2853489e713b749140241a2 Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Tue, 21 Jan 2025 00:43:28 +0530 Subject: [PATCH 3/5] fix(anta.tests): AntaTest.Input subclasses using common input models should have validators for their required fields. (#1013) * Added validators for required fields * added unit tests for input validators * Fix docstrings --------- Co-authored-by: Carl Baillargeon --- anta/input_models/routing/bgp.py | 4 +- anta/input_models/routing/generic.py | 6 +-- anta/tests/interfaces.py | 27 +++++++++- anta/tests/routing/generic.py | 20 +++++-- .../input_models/routing/test_generic.py | 41 +++++++++++++++ tests/units/input_models/test_interfaces.py | 52 +++++++++++++++++++ 6 files changed, 139 insertions(+), 11 deletions(-) create mode 100644 tests/units/input_models/routing/test_generic.py diff --git a/anta/input_models/routing/bgp.py b/anta/input_models/routing/bgp.py index a34227a1b..8b1256e18 100644 --- a/anta/input_models/routing/bgp.py +++ b/anta/input_models/routing/bgp.py @@ -224,8 +224,8 @@ class BgpRoute(BaseModel): """The IPv4 network address.""" vrf: str = "default" """Optional VRF for the BGP peer. Defaults to `default`.""" - paths: list[BgpRoutePath] | None = None - """A list of paths for the BGP route. Required field in the `VerifyBGPRouteOrigin` test.""" + paths: list[BgpRoutePath] + """A list of paths for the BGP route.""" def __str__(self) -> str: """Return a human-readable string representation of the BgpRoute for reporting. diff --git a/anta/input_models/routing/generic.py b/anta/input_models/routing/generic.py index b683a4582..277611237 100644 --- a/anta/input_models/routing/generic.py +++ b/anta/input_models/routing/generic.py @@ -17,11 +17,11 @@ class IPv4Routes(BaseModel): model_config = ConfigDict(extra="forbid") prefix: IPv4Network - """The IPV4 network to validate the route type.""" + """IPv4 prefix in CIDR notation.""" vrf: str = "default" """VRF context. Defaults to `default` VRF.""" - route_type: IPv4RouteType - """List of IPV4 Route type to validate the valid rout type.""" + route_type: IPv4RouteType | None = None + """Expected route type. Required field in the `VerifyIPv4RouteType` test.""" def __str__(self) -> str: """Return a human-readable string representation of the IPv4RouteType for reporting.""" diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py index 3191053fc..551c416b6 100644 --- a/anta/tests/interfaces.py +++ b/anta/tests/interfaces.py @@ -9,9 +9,9 @@ import re from ipaddress import IPv4Interface -from typing import Any, ClassVar +from typing import Any, ClassVar, TypeVar -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator from pydantic_extra_types.mac_address import MacAddress from anta import GITHUB_SUGGESTION @@ -23,6 +23,9 @@ BPS_GBPS_CONVERSIONS = 1000000000 +# Using a TypeVar for the InterfaceState model since mypy thinks it's a ClassVar and not a valid type when used in field validators +T = TypeVar("T", bound=InterfaceState) + class VerifyInterfaceUtilization(AntaTest): """Verifies that the utilization of interfaces is below a certain threshold. @@ -226,6 +229,16 @@ class Input(AntaTest.Input): """List of interfaces with their expected state.""" InterfaceState: ClassVar[type[InterfaceState]] = InterfaceState + @field_validator("interfaces") + @classmethod + def validate_interfaces(cls, interfaces: list[T]) -> list[T]: + """Validate that 'status' field is provided in each interface.""" + for interface in interfaces: + if interface.status is None: + msg = f"{interface} 'status' field missing in the input" + raise ValueError(msg) + return interfaces + @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyInterfacesStatus.""" @@ -891,6 +904,16 @@ class Input(AntaTest.Input): """List of interfaces with their expected state.""" InterfaceState: ClassVar[type[InterfaceState]] = InterfaceState + @field_validator("interfaces") + @classmethod + def validate_interfaces(cls, interfaces: list[T]) -> list[T]: + """Validate that 'portchannel' field is provided in each interface.""" + for interface in interfaces: + if interface.portchannel is None: + msg = f"{interface} 'portchannel' field missing in the input" + raise ValueError(msg) + return interfaces + @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyLACPInterfacesStatus.""" diff --git a/anta/tests/routing/generic.py b/anta/tests/routing/generic.py index 97709aa40..c3cd76008 100644 --- a/anta/tests/routing/generic.py +++ b/anta/tests/routing/generic.py @@ -11,7 +11,7 @@ from ipaddress import IPv4Address, IPv4Interface from typing import TYPE_CHECKING, ClassVar, Literal -from pydantic import model_validator +from pydantic import field_validator, model_validator from anta.custom_types import PositiveInteger from anta.input_models.routing.generic import IPv4Routes @@ -189,9 +189,10 @@ class VerifyIPv4RouteType(AntaTest): """Verifies the route-type of the IPv4 prefixes. This test performs the following checks for each IPv4 route: - 1. Verifies that the specified VRF is configured. - 2. Verifies that the specified IPv4 route is exists in the configuration. - 3. Verifies that the the specified IPv4 route is of the expected type. + + 1. Verifies that the specified VRF is configured. + 2. Verifies that the specified IPv4 route is exists in the configuration. + 3. Verifies that the the specified IPv4 route is of the expected type. Expected Results ---------------- @@ -230,6 +231,17 @@ class Input(AntaTest.Input): """Input model for the VerifyIPv4RouteType test.""" routes_entries: list[IPv4Routes] + """List of IPv4 route(s).""" + + @field_validator("routes_entries") + @classmethod + def validate_routes_entries(cls, routes_entries: list[IPv4Routes]) -> list[IPv4Routes]: + """Validate that 'route_type' field is provided in each BGP route entry.""" + for entry in routes_entries: + if entry.route_type is None: + msg = f"{entry} 'route_type' field missing in the input" + raise ValueError(msg) + return routes_entries @AntaTest.anta_test def test(self) -> None: diff --git a/tests/units/input_models/routing/test_generic.py b/tests/units/input_models/routing/test_generic.py new file mode 100644 index 000000000..59d069e08 --- /dev/null +++ b/tests/units/input_models/routing/test_generic.py @@ -0,0 +1,41 @@ +# 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. +"""Tests for anta.input_models.routing.generic.py.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from pydantic import ValidationError + +from anta.tests.routing.generic import VerifyIPv4RouteType + +if TYPE_CHECKING: + from anta.input_models.routing.generic import IPv4Routes + + +class TestVerifyIPv4RouteTypeInput: + """Test anta.tests.routing.bgp.VerifyIPv4RouteType.Input.""" + + @pytest.mark.parametrize( + ("routes_entries"), + [ + pytest.param([{"prefix": "192.168.0.0/24", "vrf": "default", "route_type": "eBGP"}], id="valid"), + ], + ) + def test_valid(self, routes_entries: list[IPv4Routes]) -> None: + """Test VerifyIPv4RouteType.Input valid inputs.""" + VerifyIPv4RouteType.Input(routes_entries=routes_entries) + + @pytest.mark.parametrize( + ("routes_entries"), + [ + pytest.param([{"prefix": "192.168.0.0/24", "vrf": "default"}], id="invalid"), + ], + ) + def test_invalid(self, routes_entries: list[IPv4Routes]) -> None: + """Test VerifyIPv4RouteType.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyIPv4RouteType.Input(routes_entries=routes_entries) diff --git a/tests/units/input_models/test_interfaces.py b/tests/units/input_models/test_interfaces.py index ee850ee7a..aefa31941 100644 --- a/tests/units/input_models/test_interfaces.py +++ b/tests/units/input_models/test_interfaces.py @@ -9,8 +9,10 @@ from typing import TYPE_CHECKING import pytest +from pydantic import ValidationError from anta.input_models.interfaces import InterfaceState +from anta.tests.interfaces import VerifyInterfacesStatus, VerifyLACPInterfacesStatus if TYPE_CHECKING: from anta.custom_types import Interface, PortChannelInterface @@ -31,3 +33,53 @@ class TestInterfaceState: def test_valid__str__(self, name: Interface, portchannel: PortChannelInterface | None, expected: str) -> None: """Test InterfaceState __str__.""" assert str(InterfaceState(name=name, portchannel=portchannel)) == expected + + +class TestVerifyInterfacesStatusInput: + """Test anta.tests.interfaces.VerifyInterfacesStatus.Input.""" + + @pytest.mark.parametrize( + ("interfaces"), + [ + pytest.param([{"name": "Ethernet1", "status": "up"}], id="valid"), + ], + ) + def test_valid(self, interfaces: list[InterfaceState]) -> None: + """Test VerifyInterfacesStatus.Input valid inputs.""" + VerifyInterfacesStatus.Input(interfaces=interfaces) + + @pytest.mark.parametrize( + ("interfaces"), + [ + pytest.param([{"name": "Ethernet1"}], id="invalid"), + ], + ) + def test_invalid(self, interfaces: list[InterfaceState]) -> None: + """Test VerifyInterfacesStatus.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyInterfacesStatus.Input(interfaces=interfaces) + + +class TestVerifyLACPInterfacesStatusInput: + """Test anta.tests.interfaces.VerifyLACPInterfacesStatus.Input.""" + + @pytest.mark.parametrize( + ("interfaces"), + [ + pytest.param([{"name": "Ethernet1", "portchannel": "Port-Channel100"}], id="valid"), + ], + ) + def test_valid(self, interfaces: list[InterfaceState]) -> None: + """Test VerifyLACPInterfacesStatus.Input valid inputs.""" + VerifyLACPInterfacesStatus.Input(interfaces=interfaces) + + @pytest.mark.parametrize( + ("interfaces"), + [ + pytest.param([{"name": "Ethernet1"}], id="invalid"), + ], + ) + def test_invalid(self, interfaces: list[InterfaceState]) -> None: + """Test VerifyLACPInterfacesStatus.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyLACPInterfacesStatus.Input(interfaces=interfaces) From 3f761532d79717ad92f9704eb3e738192ce4ab92 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 20 Jan 2025 20:29:06 +0100 Subject: [PATCH 4/5] ci: pre-commit autoupdate (#1014) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit updates: - [github.com/astral-sh/ruff-pre-commit: v0.9.1 → v0.9.2](https://github.com/astral-sh/ruff-pre-commit/compare/v0.9.1...v0.9.2) Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Guillaume Mulocher --- .pre-commit-config.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 42b0ce1cb..b0bf3c72d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -46,7 +46,7 @@ repos: - '' - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.9.1 + rev: v0.9.2 hooks: - id: ruff name: Run Ruff linter From 995943a6b047b93da263bc2cf8c2c1c41d0ea63a Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Tue, 21 Jan 2025 06:30:25 +0530 Subject: [PATCH 5/5] feat(anta): Added the test case to verify the SNMP source interface for specified VRF (#870) * Added TC for SNMP source interface * issue_854 handling review comments: updated failure msgs * Issue_854: Fixed pre-commit issues * Issue_854: Refactored snmp source interface verification test with latest input model changes * Issue_854: Fixed docstrings * Minor fixes --------- Co-authored-by: VitthalMagadum Co-authored-by: Geetanjali.mane Co-authored-by: Carl Baillargeon --- anta/input_models/snmp.py | 20 ++++++++++- anta/tests/snmp.py | 56 ++++++++++++++++++++++++++++- examples/tests.yaml | 7 ++++ tests/units/anta_tests/test_snmp.py | 44 +++++++++++++++++++++++ 4 files changed, 125 insertions(+), 2 deletions(-) diff --git a/anta/input_models/snmp.py b/anta/input_models/snmp.py index bd7350e45..d408d9311 100644 --- a/anta/input_models/snmp.py +++ b/anta/input_models/snmp.py @@ -9,7 +9,7 @@ from pydantic import BaseModel, ConfigDict -from anta.custom_types import Hostname, SnmpEncryptionAlgorithm, SnmpHashingAlgorithm, SnmpVersion +from anta.custom_types import Hostname, Interface, SnmpEncryptionAlgorithm, SnmpHashingAlgorithm, SnmpVersion class SnmpHost(BaseModel): @@ -54,3 +54,21 @@ def __str__(self) -> str: - User: Test Group: Test_Group Version: v2c """ return f"User: {self.username} Group: {self.group_name} Version: {self.version}" + + +class SnmpSourceInterface(BaseModel): + """Model for a SNMP source-interface.""" + + interface: Interface + """Interface to use as the source IP address of SNMP messages.""" + vrf: str = "default" + """VRF of the source interface.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the SnmpSourceInterface for reporting. + + Examples + -------- + - Source Interface: Ethernet1 VRF: default + """ + return f"Source Interface: {self.interface} VRF: {self.vrf}" diff --git a/anta/tests/snmp.py b/anta/tests/snmp.py index 910d5929c..84c5470e6 100644 --- a/anta/tests/snmp.py +++ b/anta/tests/snmp.py @@ -12,7 +12,7 @@ from pydantic import field_validator from anta.custom_types import PositiveInteger, SnmpErrorCounter, SnmpPdu -from anta.input_models.snmp import SnmpHost, SnmpUser +from anta.input_models.snmp import SnmpHost, SnmpSourceInterface, SnmpUser from anta.models import AntaCommand, AntaTest from anta.tools import get_value @@ -489,3 +489,57 @@ def test(self) -> None: if user.priv_type and (act_encryption := get_value(user_details, "v3Params.privType", "Not Found")) != user.priv_type: self.result.is_failure(f"{user} - Incorrect privacy type - Expected: {user.priv_type} Actual: {act_encryption}") + + +class VerifySnmpSourceInterface(AntaTest): + """Verifies SNMP source interfaces. + + This test performs the following checks: + + 1. Verifies that source interface(s) are configured for SNMP. + 2. For each specified source interface: + - Interface is configured in the specified VRF. + + Expected Results + ---------------- + * Success: The test will pass if the provided SNMP source interface(s) are configured in their specified VRF. + * Failure: The test will fail if any of the provided SNMP source interface(s) are NOT configured in their specified VRF. + + Examples + -------- + ```yaml + anta.tests.snmp: + - VerifySnmpSourceInterface: + interfaces: + - interface: Ethernet1 + vrf: default + - interface: Management0 + vrf: MGMT + ``` + """ + + categories: ClassVar[list[str]] = ["snmp"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show snmp", revision=1)] + + class Input(AntaTest.Input): + """Input model for the VerifySnmpSourceInterface test.""" + + interfaces: list[SnmpSourceInterface] + """List of source interfaces.""" + + @AntaTest.anta_test + def test(self) -> None: + """Main test function for VerifySnmpSourceInterface.""" + self.result.is_success() + command_output = self.instance_commands[0].json_output.get("srcIntf", {}) + + if not (interface_output := command_output.get("sourceInterfaces")): + self.result.is_failure("SNMP source interface(s) not configured") + return + + for interface_details in self.inputs.interfaces: + # If the source interface is not configured, or if it does not match the expected value, the test fails. + if not (actual_interface := interface_output.get(interface_details.vrf)): + self.result.is_failure(f"{interface_details} - Not configured") + elif actual_interface != interface_details.interface: + self.result.is_failure(f"{interface_details} - Incorrect source interface - Actual: {actual_interface}") diff --git a/examples/tests.yaml b/examples/tests.yaml index f47285a13..6c64f5d9f 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -793,6 +793,13 @@ anta.tests.snmp: pdus: - outTrapPdus - inGetNextPdus + - VerifySnmpSourceInterface: + # Verifies SNMP source interfaces. + interfaces: + - interface: Ethernet1 + vrf: default + - interface: Management0 + vrf: MGMT - VerifySnmpStatus: # Verifies if the SNMP agent is enabled. vrf: default diff --git a/tests/units/anta_tests/test_snmp.py b/tests/units/anta_tests/test_snmp.py index ebbf94897..fc30ad6ce 100644 --- a/tests/units/anta_tests/test_snmp.py +++ b/tests/units/anta_tests/test_snmp.py @@ -15,6 +15,7 @@ VerifySnmpIPv6Acl, VerifySnmpLocation, VerifySnmpPDUCounters, + VerifySnmpSourceInterface, VerifySnmpStatus, VerifySnmpUser, ) @@ -536,4 +537,47 @@ ], }, }, + { + "name": "success", + "test": VerifySnmpSourceInterface, + "eos_data": [ + { + "srcIntf": {"sourceInterfaces": {"default": "Ethernet1", "MGMT": "Management0"}}, + } + ], + "inputs": {"interfaces": [{"interface": "Ethernet1", "vrf": "default"}, {"interface": "Management0", "vrf": "MGMT"}]}, + "expected": {"result": "success"}, + }, + { + "name": "failure-not-configured", + "test": VerifySnmpSourceInterface, + "eos_data": [ + { + "srcIntf": {}, + } + ], + "inputs": {"interfaces": [{"interface": "Ethernet1", "vrf": "default"}, {"interface": "Management0", "vrf": "MGMT"}]}, + "expected": {"result": "failure", "messages": ["SNMP source interface(s) not configured"]}, + }, + { + "name": "failure-incorrect-interfaces", + "test": VerifySnmpSourceInterface, + "eos_data": [ + { + "srcIntf": { + "sourceInterfaces": { + "default": "Management0", + } + }, + } + ], + "inputs": {"interfaces": [{"interface": "Ethernet1", "vrf": "default"}, {"interface": "Management0", "vrf": "MGMT"}]}, + "expected": { + "result": "failure", + "messages": [ + "Source Interface: Ethernet1 VRF: default - Incorrect source interface - Actual: Management0", + "Source Interface: Management0 VRF: MGMT - Not configured", + ], + }, + }, ]