diff --git a/ansible_collections/arista/avd/plugins/action/eos_validate_state_runner.py b/ansible_collections/arista/avd/plugins/action/eos_validate_state_runner.py index 57dae10f1a0..43bf92ad169 100644 --- a/ansible_collections/arista/avd/plugins/action/eos_validate_state_runner.py +++ b/ansible_collections/arista/avd/plugins/action/eos_validate_state_runner.py @@ -36,7 +36,7 @@ def ignore_aliases(self, data): class ActionModule(ActionBase): - # @cprofile + # @cprofile() def run(self, tmp=None, task_vars=None): self._supports_check_mode = True 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 09c94934199..a6294474026 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,12 +46,9 @@ 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) -> Mapping: """ - Retrieves and returns the structured configuration for a specified host. - - The function fetches the structured_config from hostvars. It ensures the returned object is a non-empty - dictionary, converting Ansible dictionary-like objects to a standard dictionary if necessary. + Get a specified host's structured configuration from the hostvars. Args: host (str): Hostname to retrieve the structured_config. @@ -60,19 +57,16 @@ def get_host_struct_cfg(self, host: str) -> dict: 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. + AristaAvdError: If host is not in hostvars or if its structured_config is not a mapping object. """ if host not in self.hostvars: raise AristaAvdError(f"Host '{host}' is missing from the hostvars.") - struct_cfg = get(self.hostvars, host, separator="..") + struct_cfg = self.hostvars[host] - # 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): - struct_cfg = dict(struct_cfg) - return struct_cfg def log_skip_message( @@ -124,7 +118,7 @@ def update_interface_shutdown(self, interface: dict, host: str | None = None) -> bool: Returns the interface shutdown key's value, or `interface_defaults.ethernet` if not available. Returns False if both are absent, which is the default EOS behavior. """ - 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 +154,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 +172,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 +209,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