Skip to content

Commit

Permalink
Merge branch 'main' into issue_784
Browse files Browse the repository at this point in the history
  • Loading branch information
carl-baillargeon committed Sep 17, 2024
2 parents aab554e + 900ebb7 commit 998501b
Show file tree
Hide file tree
Showing 48 changed files with 1,825 additions and 461 deletions.
4 changes: 2 additions & 2 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ repos:
- '<!--| ~| -->'

- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.6.3
rev: v0.6.5
hooks:
- id: ruff
name: Run Ruff linter
Expand All @@ -52,7 +52,7 @@ repos:
name: Run Ruff formatter

- repo: https://github.com/pycqa/pylint
rev: "v3.2.6"
rev: "v3.2.7"
hooks:
- id: pylint
name: Check code style with pylint
Expand Down
90 changes: 59 additions & 31 deletions anta/catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,12 @@
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)
Expand All @@ -60,6 +64,7 @@ def serialize_model(self) -> dict[str, AntaTest.Input]:
Returns
-------
dict
A dictionary representing the model.
"""
return {self.test.__name__: self.inputs}
Expand Down Expand Up @@ -132,14 +137,14 @@ def check_inputs(self) -> AntaTestDefinition:
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>
```
"""

Expand All @@ -149,16 +154,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]] = {}
Expand Down Expand Up @@ -234,6 +239,7 @@ def yaml(self) -> str:
Returns
-------
str
The YAML representation string of this model.
"""
# TODO: Pydantic and YAML serialization/deserialization is not supported natively.
Expand All @@ -247,6 +253,7 @@ def to_json(self) -> str:
Returns
-------
str
The JSON representation string of this model.
"""
return self.model_dump_json(serialize_as_any=True, exclude_unset=True, indent=2)
Expand All @@ -267,8 +274,10 @@ def __init__(
Parameters
----------
tests: A list of AntaTestDefinition instances.
filename: The path from which the catalog is loaded.
tests
A list of AntaTestDefinition instances.
filename
The path from which the catalog is loaded.
"""
self._tests: list[AntaTestDefinition] = []
Expand Down Expand Up @@ -314,8 +323,10 @@ def parse(filename: str | Path, file_format: Literal["yaml", "json"] = "yaml") -
Parameters
----------
filename: Path to test catalog YAML or JSON fil
file_format: Format of the file, either 'yaml' or 'json'
filename
Path to test catalog YAML or JSON file.
file_format
Format of the file, either 'yaml' or 'json'.
"""
if file_format not in ["yaml", "json"]:
Expand Down Expand Up @@ -343,7 +354,8 @@ def from_dict(data: RawCatalogInput, filename: str | Path | None = None) -> Anta
Parameters
----------
data: Python dictionary used to instantiate the AntaCatalog instance
data
Python dictionary used to instantiate the AntaCatalog instance.
filename: value to be set as AntaCatalog instance attribute
"""
Expand Down Expand Up @@ -377,7 +389,8 @@ def from_list(data: ListAntaTestTuples) -> AntaCatalog:
Parameters
----------
data: Python list used to instantiate the AntaCatalog instance
data
Python list used to instantiate the AntaCatalog instance.
"""
tests: list[AntaTestDefinition] = []
Expand All @@ -394,10 +407,12 @@ def merge_catalogs(cls, catalogs: list[AntaCatalog]) -> AntaCatalog:
Parameters
----------
catalogs: A list of AntaCatalog instances to merge.
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)))
Expand All @@ -406,12 +421,18 @@ def merge_catalogs(cls, catalogs: list[AntaCatalog]) -> AntaCatalog:
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.
catalog
AntaCatalog instance to merge to this instance.
Returns
-------
AntaCatalog
A new AntaCatalog instance containing the tests of the two instances.
"""
# TODO: Use a decorator to deprecate this method instead. See https://github.com/aristanetworks/anta/issues/754
Expand All @@ -427,6 +448,7 @@ def dump(self) -> AntaCatalogFile:
Returns
-------
AntaCatalogFile
An AntaCatalogFile instance containing tests of this AntaCatalog instance.
"""
root: dict[ImportString[Any], list[AntaTestDefinition]] = {}
Expand All @@ -441,7 +463,9 @@ def build_indexes(self, filtered_tests: set[str] | None = None) -> None:
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.
Once the indexes are built, the `indexes_built` attribute is set to True.
Expand All @@ -466,17 +490,21 @@ def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> set[Anta
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).
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."
Expand Down
27 changes: 18 additions & 9 deletions anta/cli/get/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,18 +84,24 @@ def get_cv_token(cvp_ip: str, cvp_username: str, cvp_password: str, *, verify_ce
Parameters
----------
cvp_ip: IP address of CloudVision.
cvp_username: Username to connect to CloudVision.
cvp_password: Password to connect to CloudVision.
verify_cert: Enable or disable certificate verification when connecting to CloudVision.
cvp_ip
IP address of CloudVision.
cvp_username
Username to connect to CloudVision.
cvp_password
Password to connect to CloudVision.
verify_cert
Enable or disable certificate verification when connecting to CloudVision.
Returns
-------
token(str): The token to use in further API calls to CloudVision.
str
The token to use in further API calls to CloudVision.
Raises
------
requests.ssl.SSLError: If the certificate verification fails
requests.ssl.SSLError
If the certificate verification fails.
"""
# use CVP REST API to generate a token
Expand Down Expand Up @@ -163,9 +169,12 @@ def create_inventory_from_ansible(inventory: Path, output: Path, ansible_group:
Parameters
----------
inventory: Ansible Inventory file to read
output: ANTA inventory file to generate.
ansible_group: Ansible group from where to extract data.
inventory
Ansible Inventory file to read.
output
ANTA inventory file to generate.
ansible_group
Ansible group from where to extract data.
"""
try:
Expand Down
6 changes: 4 additions & 2 deletions anta/cli/nrfu/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,8 +147,10 @@ def save_markdown_report(ctx: click.Context, md_output: pathlib.Path) -> None:
Parameters
----------
ctx: Click context containing the result manager.
md_output: Path to save the markdown report.
ctx
Click context containing the result manager.
md_output
Path to save the markdown report.
"""
try:
MDReportGenerator.generate(results=_get_result_manager(ctx), md_filename=md_output)
Expand Down
3 changes: 2 additions & 1 deletion anta/cli/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,8 @@ def exit_with_code(ctx: click.Context) -> None:
Parameters
----------
ctx: Click Context
ctx
Click Context.
"""
if ctx.obj.get("ignore_status"):
Expand Down
22 changes: 15 additions & 7 deletions anta/custom_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
"""Match EOS interface types like Ethernet1/1, Vlan1, Loopback1, etc."""
REGEXP_TYPE_VXLAN_SRC_INTERFACE = r"^(Loopback)([0-9]|[1-9][0-9]{1,2}|[1-7][0-9]{3}|8[01][0-9]{2}|819[01])$"
"""Match Vxlan source interface like Loopback10."""
REGEX_TYPE_PORTCHANNEL = r"^Port-Channel[0-9]{1,6}$"
"""Match Port Channel interface like Port-Channel5."""
REGEXP_TYPE_HOSTNAME = r"^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])\.)*([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]*[A-Za-z0-9])$"
"""Match hostname like `my-hostname`, `my-hostname-1`, `my-hostname-1-2`."""

Expand Down Expand Up @@ -66,9 +68,9 @@ def interface_case_sensitivity(v: str) -> str:
Examples
--------
- ethernet -> Ethernet
- vlan -> Vlan
- loopback -> Loopback
- ethernet -> Ethernet
- vlan -> Vlan
- loopback -> Loopback
"""
if isinstance(v, str) and v != "" and not v[0].isupper():
Expand All @@ -81,10 +83,10 @@ def bgp_multiprotocol_capabilities_abbreviations(value: str) -> str:
Examples
--------
- IPv4 Unicast
- L2vpnEVPN
- ipv4 MPLS Labels
- ipv4Mplsvpn
- IPv4 Unicast
- L2vpnEVPN
- ipv4 MPLS Labels
- ipv4Mplsvpn
"""
patterns = {
Expand Down Expand Up @@ -135,6 +137,12 @@ def validate_regex(value: str) -> str:
BeforeValidator(interface_autocomplete),
BeforeValidator(interface_case_sensitivity),
]
PortChannelInterface = Annotated[
str,
Field(pattern=REGEX_TYPE_PORTCHANNEL),
BeforeValidator(interface_autocomplete),
BeforeValidator(interface_case_sensitivity),
]
Afi = Literal["ipv4", "ipv6", "vpn-ipv4", "vpn-ipv6", "evpn", "rt-membership", "path-selection", "link-state"]
Safi = Literal["unicast", "multicast", "labeled-unicast", "sr-te"]
EncryptionAlgorithm = Literal["RSA", "ECDSA"]
Expand Down
Loading

0 comments on commit 998501b

Please sign in to comment.