diff --git a/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/avdtestbase.py b/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/avdtestbase.py index 3b0cc094b25..07e8b0c796a 100644 --- a/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/avdtestbase.py +++ b/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/avdtestbase.py @@ -4,9 +4,9 @@ from __future__ import annotations import logging -from collections.abc import Mapping from functools import cached_property from ipaddress import ip_interface +from typing import Mapping from ansible_collections.arista.avd.plugins.plugin_utils.errors import AristaAvdError, AristaAvdMissingVariableError from ansible_collections.arista.avd.plugins.plugin_utils.utils import default, get, get_item @@ -20,18 +20,18 @@ class AvdTestBase: Base class for all AVD eos_validate_state tests. """ - def __init__(self, device_name: str, hostvars: dict | object): + def __init__(self, device_name: str, hostvars: Mapping): """ Initialize the AvdTestBase class. Args: device_name (str): The current device name for which the plugin is being run. - hostvars (dict | object): A dictionary that contains a key for each device with a value of the structured_config. + hostvars (Mapping): A mapping that contains a key for each device with a value of the structured_config. When using Ansible, this is the `task_vars['hostvars']` object. """ self.hostvars = hostvars self.device_name = device_name - self.structured_config = self.get_host_struct_cfg(host=device_name) + self.structured_config = self.get_host_structured_config(host=device_name) def render(self) -> dict: """ @@ -46,7 +46,7 @@ def render(self) -> dict: """ return getattr(self, "test_definition", None) or {} - def get_host_struct_cfg(self, host: str) -> dict: + def get_host_structured_config(self, host: str) -> dict: """ Retrieves and returns the structured configuration for a specified host. @@ -58,16 +58,13 @@ def get_host_struct_cfg(self, host: str) -> dict: Returns: dict: Structured configuration for the host. - - Raises: - AristaAvdError: If host is not in hostvars or if its structured_config is not a dictionary or is empty. """ if host not in self.hostvars: raise AristaAvdError(f"Host '{host}' is missing from the hostvars.") struct_cfg = get(self.hostvars, host, separator="..") - # Check if struct_cfg is a dict or behaves like a dict (e.g. Ansible 'hostvars' object) - if not isinstance(struct_cfg, (dict, Mapping)): + # Check if struct_cfg is a mapping object (e.g. Ansible 'hostvars' object or regular dict) + if not isinstance(struct_cfg, Mapping): raise AristaAvdError(f"Host '{host}' structured_config is not a dictionary or dictionary-like object.") if not isinstance(struct_cfg, dict): @@ -79,7 +76,7 @@ def log_skip_message( self, message=None, key: str | None = None, value=None, key_path: str | None = None, is_missing: bool = True, logging_level: str = "INFO" ) -> None: """ - Logging function that logs the test being skipped appended to a formatted message based on the provided parameters. + Logging function that logs the test being skipped, appended to a formatted message based on the provided parameters. Args: message (Any | None): The message to be logged. If provided, it will be logged as is, ignoring other parameters. @@ -124,7 +121,7 @@ def update_interface_shutdown(self, interface: dict, host: str | None = None) -> interface (dict): The interface to update. host (str): Host to verify. Defaults to the host running the test. """ - host_struct_cfg = self.get_host_struct_cfg(host=host) if host else self.structured_config + host_struct_cfg = self.get_host_structured_config(host=host) if host else self.structured_config if "Ethernet" in get(interface, "name", ""): interface["shutdown"] = default(get(interface, "shutdown"), get(host_struct_cfg, "interface_defaults.ethernet.shutdown"), False) else: @@ -160,7 +157,7 @@ def get_interface_ip(self, interface_model: str, interface_name: str, host: str Returns: str | None: IP address of the host interface or None if unavailable. """ - host_struct_cfg = self.get_host_struct_cfg(host=host) if host else self.structured_config + host_struct_cfg = self.get_host_structured_config(host=host) if host else self.structured_config try: peer_interfaces = get(host_struct_cfg, interface_model, required=True) peer_interface = get_item(peer_interfaces, "name", interface_name, required=True) @@ -178,7 +175,7 @@ def logged_get(self, key: str, host: str | None = None, logging_level: str = "WA host (str | None): The host from which to retrieve the key. Defaults to the device running the test. logging_level (str): The logging level to use for the log message. """ - host_struct_cfg = self.get_host_struct_cfg(host=host) if host else self.structured_config + host_struct_cfg = self.get_host_structured_config(host=host) if host else self.structured_config try: return get(host_struct_cfg, key, required=True, separator="..") except AristaAvdMissingVariableError: @@ -215,7 +212,7 @@ def validate_data( In this case, the function will log a warning message because the key 'c' with value '3' is not found, and it will return False as the data doesn't meet all validation requirements. """ - data = data or (self.get_host_struct_cfg(host=host) if host else self.structured_config) + data = data or (self.get_host_structured_config(host=host) if host else self.structured_config) valid = True # Check the expected key/value pairs first diff --git a/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/catalog.py b/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/catalog.py index 193616758db..700fb249ad0 100644 --- a/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/catalog.py +++ b/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/catalog.py @@ -3,7 +3,7 @@ # that can be found in the LICENSE file. from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Mapping from yaml import Dumper, dump, safe_load @@ -34,9 +34,7 @@ class Catalog: Catalog class that handles management of ANTA catalogs from AVD. """ - def __init__( - self, device_name: str, hostvars: dict | object, skipped_tests: dict, custom_catalog: dict | None = None, custom_catalog_path: str | None = None - ): + def __init__(self, device_name: str, hostvars: Mapping, skipped_tests: dict, custom_catalog: dict | None = None, custom_catalog_path: str | None = None): """ Initialize the Catalog class. @@ -44,7 +42,7 @@ def __init__( Args: device_name (str): The device for which this catalog is built - hostvars (dict | object): A dictionary that contains a key for each device with a value of the structured_config. + hostvars (Mapping): A mapping that contains a key for each device with a value of the structured_config. When using Ansible, this is the `task_vars['hostvars']` object. skipped_tests (list[dict]): Dictionary of AVD test classes to be skipped. custom_catalog (dict): An optional dictionary representing an ANTA Catalog diff --git a/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/get_anta_results.py b/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/get_anta_results.py index c7b6543393b..dfcac537114 100644 --- a/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/get_anta_results.py +++ b/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/get_anta_results.py @@ -6,7 +6,7 @@ import logging from asyncio import run from json import loads -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Mapping from ansible_collections.arista.avd.plugins.plugin_utils.errors import AristaAvdError from ansible_collections.arista.avd.plugins.plugin_utils.utils import NoAliasDumper @@ -63,7 +63,7 @@ def _get_skipped_tests_from_tags(run_tags: tuple, skip_tags: tuple) -> list[dict def get_anta_results( anta_device: AntaDevice, - hostvars: dict | object, + hostvars: Mapping, logging_level: str, skipped_tests: dict, ansible_tags: dict | None = None, @@ -75,7 +75,7 @@ def get_anta_results( Args: anta_device (AntaDevice): An instantiated AntaDevice When running in ansible, the action plugin will pass an AnsibleEOSDevice - hostvars (dict | object): A dictionary that contains a key for each device with a value of the structured_config. + hostvars (Mapping): A mapping that contains a key for each device with a value of the structured_config. When using Ansible, this is the `task_vars['hostvars']` object. logging_level (str): The level at which ANTA should be logging skipped_tests (list[dict]): A list of dictionary