arista/anta
1
0
Fork 0
Code Issues Activity

Adding upstream version 1.1.0.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-05 11:54:23 +01:00
parent f13b7abbd8
commit 77504588ab
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
196 changed files with 10121 additions and 3780 deletions
Download patch file Download diff file Expand all files Collapse all files

214
anta/catalog.py
View file

@ -10,21 +10,29 @@ import logging
import math
from collections import defaultdict
from inspect import isclass
from itertools import chain
from json import load as json_load
from pathlib import Path
from typing import TYPE_CHECKING, Any, Optional, Union
from typing import TYPE_CHECKING, Any, Literal, Optional, Union
from warnings import warn
import yaml
from pydantic import BaseModel, ConfigDict, RootModel, ValidationError, ValidationInfo, field_validator, model_serializer, model_validator
from pydantic.types import ImportString
from pydantic_core import PydanticCustomError
from yaml import YAMLError, safe_load
from yaml import YAMLError, safe_dump, safe_load
from anta.logger import anta_log_exception
from anta.models import AntaTest
if TYPE_CHECKING:
import sys
from types import ModuleType
if sys.version_info >= (3, 11):
from typing import Self
else:
from typing_extensions import Self
logger = logging.getLogger(__name__)
# { <module_name> : [ { <test_class_name>: <input_as_dict_or_None> }, ... ] }
@ -37,8 +45,12 @@ ListAntaTestTuples = list[tuple[type[AntaTest], Optional[Union[AntaTest.Input, d
class AntaTestDefinition(BaseModel):
"""Define a test with its associated inputs.
test: An AntaTest concrete subclass
inputs: The associated AntaTest.Input subclass instance
Attributes
----------
test
An AntaTest concrete subclass.
inputs
The associated AntaTest.Input subclass instance.
"""
model_config = ConfigDict(frozen=True)
@ -58,6 +70,7 @@ class AntaTestDefinition(BaseModel):
Returns
-------
dict
A dictionary representing the model.
"""
return {self.test.__name__: self.inputs}
@ -116,7 +129,7 @@ class AntaTestDefinition(BaseModel):
raise ValueError(msg)
@model_validator(mode="after")
def check_inputs(self) -> AntaTestDefinition:
def check_inputs(self) -> Self:
"""Check the `inputs` field typing.
The `inputs` class attribute needs to be an instance of the AntaTest.Input subclass defined in the class `test`.
@ -130,14 +143,14 @@ class AntaTestDefinition(BaseModel):
class AntaCatalogFile(RootModel[dict[ImportString[Any], list[AntaTestDefinition]]]): # pylint: disable=too-few-public-methods
"""Represents an ANTA Test Catalog File.
Example:
Example
-------
A valid test catalog file must have the following structure:
```
<Python module>:
- <AntaTest subclass>:
<AntaTest.Input compliant dictionary>
```
A valid test catalog file must have the following structure:
```
<Python module>:
- <AntaTest subclass>:
<AntaTest.Input compliant dictionary>
```
"""
@ -147,16 +160,16 @@ class AntaCatalogFile(RootModel[dict[ImportString[Any], list[AntaTestDefinition]
def flatten_modules(data: dict[str, Any], package: str | None = None) -> dict[ModuleType, list[Any]]:
"""Allow the user to provide a data structure with nested Python modules.
Example:
Example
-------
```
anta.tests.routing:
generic:
- <AntaTestDefinition>
bgp:
- <AntaTestDefinition>
```
`anta.tests.routing.generic` and `anta.tests.routing.bgp` are importable Python modules.
```
anta.tests.routing:
generic:
- <AntaTestDefinition>
bgp:
- <AntaTestDefinition>
```
`anta.tests.routing.generic` and `anta.tests.routing.bgp` are importable Python modules.
"""
modules: dict[ModuleType, list[Any]] = {}
@ -166,7 +179,7 @@ class AntaCatalogFile(RootModel[dict[ImportString[Any], list[AntaTestDefinition]
module_name = f".{module_name}" # noqa: PLW2901
try:
module: ModuleType = importlib.import_module(name=module_name, package=package)
except Exception as e: # pylint: disable=broad-exception-caught
except Exception as e:
# A test module is potentially user-defined code.
# We need to catch everything if we want to have meaningful logs
module_str = f"{module_name[1:] if module_name.startswith('.') else module_name}{f' from package {package}' if package else ''}"
@ -232,13 +245,24 @@ class AntaCatalogFile(RootModel[dict[ImportString[Any], list[AntaTestDefinition]
Returns
-------
str
The YAML representation string of this model.
"""
# TODO: Pydantic and YAML serialization/deserialization is not supported natively.
# This could be improved.
# https://github.com/pydantic/pydantic/issues/1043
# Explore if this worth using this: https://github.com/NowanIlfideme/pydantic-yaml
return yaml.safe_dump(yaml.safe_load(self.model_dump_json(serialize_as_any=True, exclude_unset=True)), indent=2, width=math.inf)
return safe_dump(safe_load(self.model_dump_json(serialize_as_any=True, exclude_unset=True)), indent=2, width=math.inf)
def to_json(self) -> str:
"""Return a JSON representation string of this model.
Returns
-------
str
The JSON representation string of this model.
"""
return self.model_dump_json(serialize_as_any=True, exclude_unset=True, indent=2)
class AntaCatalog:
@ -254,10 +278,12 @@ class AntaCatalog:
) -> None:
"""Instantiate an AntaCatalog instance.
Args:
----
tests: A list of AntaTestDefinition instances.
filename: The path from which the catalog is loaded.
Parameters
----------
tests
A list of AntaTestDefinition instances.
filename
The path from which the catalog is loaded.
"""
self._tests: list[AntaTestDefinition] = []
@ -270,11 +296,14 @@ class AntaCatalog:
else:
self._filename = Path(filename)
# Default indexes for faster access
self.tag_to_tests: defaultdict[str | None, set[AntaTestDefinition]] = defaultdict(set)
self.tests_without_tags: set[AntaTestDefinition] = set()
self.indexes_built: bool = False
self.final_tests_count: int = 0
self.indexes_built: bool
self.tag_to_tests: defaultdict[str | None, set[AntaTestDefinition]]
self._init_indexes()
def _init_indexes(self) -> None:
"""Init indexes related variables."""
self.tag_to_tests = defaultdict(set)
self.indexes_built = False
@property
def filename(self) -> Path | None:
@ -298,19 +327,30 @@ class AntaCatalog:
self._tests = value
@staticmethod
def parse(filename: str | Path) -> AntaCatalog:
def parse(filename: str | Path, file_format: Literal["yaml", "json"] = "yaml") -> AntaCatalog:
"""Create an AntaCatalog instance from a test catalog file.
Args:
----
filename: Path to test catalog YAML file
Parameters
----------
filename
Path to test catalog YAML or JSON file.
file_format
Format of the file, either 'yaml' or 'json'.
Returns
-------
AntaCatalog
An AntaCatalog populated with the file content.
"""
if file_format not in ["yaml", "json"]:
message = f"'{file_format}' is not a valid format for an AntaCatalog file. Only 'yaml' and 'json' are supported."
raise ValueError(message)
try:
file: Path = filename if isinstance(filename, Path) else Path(filename)
with file.open(encoding="UTF-8") as f:
data = safe_load(f)
except (TypeError, YAMLError, OSError) as e:
data = safe_load(f) if file_format == "yaml" else json_load(f)
except (TypeError, YAMLError, OSError, ValueError) as e:
message = f"Unable to parse ANTA Test Catalog file '{filename}'"
anta_log_exception(e, message, logger)
raise
@ -325,11 +365,17 @@ class AntaCatalog:
It is the data structure returned by `yaml.load()` function of a valid
YAML Test Catalog file.
Args:
----
data: Python dictionary used to instantiate the AntaCatalog instance
filename: value to be set as AntaCatalog instance attribute
Parameters
----------
data
Python dictionary used to instantiate the AntaCatalog instance.
filename
value to be set as AntaCatalog instance attribute
Returns
-------
AntaCatalog
An AntaCatalog populated with the 'data' dictionary content.
"""
tests: list[AntaTestDefinition] = []
if data is None:
@ -359,10 +405,15 @@ class AntaCatalog:
See ListAntaTestTuples type alias for details.
Args:
----
data: Python list used to instantiate the AntaCatalog instance
Parameters
----------
data
Python list used to instantiate the AntaCatalog instance.
Returns
-------
AntaCatalog
An AntaCatalog populated with the 'data' list content.
"""
tests: list[AntaTestDefinition] = []
try:
@ -372,24 +423,54 @@ class AntaCatalog:
raise
return AntaCatalog(tests)
def merge(self, catalog: AntaCatalog) -> AntaCatalog:
"""Merge two AntaCatalog instances.
@classmethod
def merge_catalogs(cls, catalogs: list[AntaCatalog]) -> AntaCatalog:
"""Merge multiple AntaCatalog instances.
Args:
----
catalog: AntaCatalog instance to merge to this instance.
Parameters
----------
catalogs
A list of AntaCatalog instances to merge.
Returns
-------
AntaCatalog
A new AntaCatalog instance containing the tests of all the input catalogs.
"""
combined_tests = list(chain(*(catalog.tests for catalog in catalogs)))
return cls(tests=combined_tests)
def merge(self, catalog: AntaCatalog) -> AntaCatalog:
"""Merge two AntaCatalog instances.
Warning
-------
This method is deprecated and will be removed in ANTA v2.0. Use `AntaCatalog.merge_catalogs()` instead.
Parameters
----------
catalog
AntaCatalog instance to merge to this instance.
Returns
-------
AntaCatalog
A new AntaCatalog instance containing the tests of the two instances.
"""
return AntaCatalog(tests=self.tests + catalog.tests)
# TODO: Use a decorator to deprecate this method instead. See https://github.com/aristanetworks/anta/issues/754
warn(
message="AntaCatalog.merge() is deprecated and will be removed in ANTA v2.0. Use AntaCatalog.merge_catalogs() instead.",
category=DeprecationWarning,
stacklevel=2,
)
return self.merge_catalogs([self, catalog])
def dump(self) -> AntaCatalogFile:
"""Return an AntaCatalogFile instance from this AntaCatalog instance.
Returns
-------
AntaCatalogFile
An AntaCatalogFile instance containing tests of this AntaCatalog instance.
"""
root: dict[ImportString[Any], list[AntaTestDefinition]] = {}
@ -403,9 +484,7 @@ class AntaCatalog:
If a `filtered_tests` set is provided, only the tests in this set will be indexed.
This method populates two attributes:
- tag_to_tests: A dictionary mapping each tag to a set of tests that contain it.
- tests_without_tags: A set of tests that do not have any tags.
This method populates the tag_to_tests attribute, which is a dictionary mapping tags to sets of tests.
Once the indexes are built, the `indexes_built` attribute is set to True.
"""
@ -419,27 +498,34 @@ class AntaCatalog:
for tag in test_tags:
self.tag_to_tests[tag].add(test)
else:
self.tests_without_tags.add(test)
self.tag_to_tests[None].add(test)
self.tag_to_tests[None] = self.tests_without_tags
self.indexes_built = True
def clear_indexes(self) -> None:
"""Clear this AntaCatalog instance indexes."""
self._init_indexes()
def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> set[AntaTestDefinition]:
"""Return all tests that match a given set of tags, according to the specified strictness.
Args:
----
tags: The tags to filter tests by. If empty, return all tests without tags.
strict: If True, returns only tests that contain all specified tags (intersection).
If False, returns tests that contain any of the specified tags (union).
Parameters
----------
tags
The tags to filter tests by. If empty, return all tests without tags.
strict
If True, returns only tests that contain all specified tags (intersection).
If False, returns tests that contain any of the specified tags (union).
Returns
-------
set[AntaTestDefinition]: A set of tests that match the given tags.
set[AntaTestDefinition]
A set of tests that match the given tags.
Raises
------
ValueError: If the indexes have not been built prior to method call.
ValueError
If the indexes have not been built prior to method call.
"""
if not self.indexes_built:
msg = "Indexes have not been built yet. Call build_indexes() first."