Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix(eos_validate_state): ANTA Fix AvdTestBase structured_config objects #3447

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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:
"""
Expand All @@ -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.
Expand All @@ -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)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
# Check if struct_cfg is a mapping object (e.g. Ansible 'hostvars' object or regular dict)
# 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(
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down