feat(anta): Add tag mapping between devices and tests (#377)

* feat(anta): Add tag mapping between devices and tests

* ci: fix code linting

* doc: Update test documentation

* feat(anta): Propagate default tags for all CLI commands

* fix: remove unecessary mypy ignore code

* doc: Update docs/usage-inventory-catalog.md

Co-authored-by: Guillaume Mulocher <[email protected]>

* fix(anta.cli): Update tags default value management

* fix(anta): code review

* fix(anta): code linting

* fix(anta): code linting & logic optimization

* fix(anta): Update for unit tests

* doc: Update with tag management

* doc: Document TAG management

* Refactor: Remove some more Optional

* Refactor: Add show_default=True for nrfu tags

* feat: Update --tags implementation as per #324

* test: remove last update in fixture

* cut(anta): Remove default tag from ANTA

* fix: remove breaking logs

* doc: update new tag management

* ci: lint code with pylint

* ci: code typing

* docs: remove tag "all" occurences

* use list type for typing

* Update docs/cli/tag-management.md

Co-authored-by: Guillaume Mulocher <[email protected]>

* Update docs/cli/tag-management.md

Co-authored-by: Guillaume Mulocher <[email protected]>

---------

Co-authored-by: Guillaume Mulocher <[email protected]>
Co-authored-by: Matthieu Tâche <[email protected]>

.gitignore

@@ -104,4 +104,7 @@ tech-support/*
 2*
 
 **/report.html
-.*report.html
+.*report.html
+
+clab-atd-anta/*
+clab-atd-anta/

anta/cli/exec/commands.py

@@ -12,7 +12,6 @@
 import sys
 from datetime import datetime
 from pathlib import Path
-from typing import Optional
 
 import click
 from yaml import safe_load
@@ -26,7 +25,7 @@
 @click.command()
 @click.pass_context
 @click.option("--tags", "-t", help="List of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags)
-def clear_counters(ctx: click.Context, tags: Optional[list[str]]) -> None:
+def clear_counters(ctx: click.Context, tags: list[str] | None) -> None:
     """Clear counter statistics on EOS devices"""
     asyncio.run(clear_counters_utils(ctx.obj["inventory"], tags=tags))
 
@@ -51,7 +50,7 @@ def clear_counters(ctx: click.Context, tags: Optional[list[str]]) -> None:
     default=f"anta_snapshot_{datetime.now().strftime('%Y-%m-%d_%H_%M_%S')}",
     show_default=True,
 )
-def snapshot(ctx: click.Context, tags: Optional[list[str]], commands_list: Path, output: Path) -> None:
+def snapshot(ctx: click.Context, tags: list[str] | None, commands_list: Path, output: Path) -> None:
     """Collect commands output from devices in inventory"""
     print(f"Collecting data for {commands_list}")
     print(f"Output directory is {output}")
@@ -77,6 +76,6 @@ def snapshot(ctx: click.Context, tags: Optional[list[str]], commands_list: Path,
     show_default=True,
 )
 @click.option("--tags", "-t", help="List of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags)
-def collect_tech_support(ctx: click.Context, tags: Optional[list[str]], output: Path, latest: Optional[int], configure: bool) -> None:
+def collect_tech_support(ctx: click.Context, tags: list[str] | None, output: Path, latest: int | None, configure: bool) -> None:
     """Collect scheduled tech-support from EOS devices"""
     asyncio.run(collect_scheduled_show_tech(ctx.obj["inventory"], output, configure, tags=tags, latest=latest))

anta/cli/exec/utils.py

@@ -12,7 +12,7 @@
 import json
 import logging
 from pathlib import Path
-from typing import Literal, Optional
+from typing import Literal
 
 from aioeapi import EapiCommandError
 
@@ -26,7 +26,7 @@
 logger = logging.getLogger(__name__)
 
 
-async def clear_counters_utils(anta_inventory: AntaInventory, tags: Optional[list[str]] = None) -> None:
+async def clear_counters_utils(anta_inventory: AntaInventory, tags: list[str] | None = None) -> None:
     """
     Clear counters
     """
@@ -52,7 +52,7 @@ async def collect_commands(
     inv: AntaInventory,
     commands: dict[str, str],
     root_dir: Path,
-    tags: Optional[list[str]] = None,
+    tags: list[str] | None = None,
 ) -> None:
     """
     Collect EOS commands
@@ -92,7 +92,7 @@ async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "tex
             anta_log_exception(r, message, logger)
 
 
-async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, configure: bool, tags: Optional[list[str]] = None, latest: Optional[int] = None) -> None:
+async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, configure: bool, tags: list[str] | None = None, latest: int | None = None) -> None:
     """
     Collect scheduled show-tech on devices
     """

anta/cli/get/commands.py

@@ -23,7 +23,6 @@
 
 from anta.cli.console import console
 from anta.cli.utils import parse_tags
-from anta.models import DEFAULT_TAG
 
 from .utils import create_inventory_from_ansible, create_inventory_from_cvp, get_cv_token
 
@@ -130,4 +129,3 @@ def tags(ctx: click.Context) -> None:
     tags_found = sorted(set(tags_found))
     console.print("Tags found:")
     console.print_json(json.dumps(tags_found, indent=2))
-    console.print(f"\n* note that tag [green]{DEFAULT_TAG}[/green] has been added by anta")

anta/cli/nrfu/commands.py

@@ -1,8 +1,6 @@
-#!/usr/bin/env python
 # Copyright (c) 2023 Arista Networks, Inc.
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the LICENSE file.
-# coding: utf-8 -*-
 """
 Commands for Anta CLI to run nrfu commands.
 """
@@ -11,7 +9,6 @@
 import asyncio
 import logging
 import pathlib
-from typing import Optional
 
 import click
 
@@ -26,13 +23,13 @@
 
 @click.command()
 @click.pass_context
-@click.option("--tags", help="list of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags)
+@click.option("--tags", default=None, help="List of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags, show_default=True)
 @click.option("--device", "-d", help="Show a summary for this device", type=str, required=False)
 @click.option("--test", "-t", help="Show a summary for this test", type=str, required=False)
 @click.option(
     "--group-by", default=None, type=click.Choice(["device", "test"], case_sensitive=False), help="Group result by test or host. default none", required=False
 )
-def table(ctx: click.Context, tags: Optional[list[str]], device: Optional[str], test: Optional[str], group_by: str) -> None:
+def table(ctx: click.Context, tags: list[str], device: str | None, test: str | None, group_by: str) -> None:
     """ANTA command to check network states with table result"""
     print_settings(ctx)
     with anta_progress_bar() as AntaTest.progress:
@@ -43,7 +40,9 @@ def table(ctx: click.Context, tags: Optional[list[str]], device: Optional[str],
 
 @click.command()
 @click.pass_context
-@click.option("--tags", "-t", help="list of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags)
+@click.option(
+    "--tags", "-t", default=None, help="List of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags, show_default=True
+)
 @click.option(
     "--output",
     "-o",
@@ -52,7 +51,7 @@ def table(ctx: click.Context, tags: Optional[list[str]], device: Optional[str],
     required=False,
     help="Path to save report as a file",
 )
-def json(ctx: click.Context, tags: Optional[list[str]], output: Optional[pathlib.Path]) -> None:
+def json(ctx: click.Context, tags: list[str], output: pathlib.Path | None) -> None:
     """ANTA command to check network state with JSON result"""
     print_settings(ctx)
     with anta_progress_bar() as AntaTest.progress:
@@ -63,10 +62,12 @@ def json(ctx: click.Context, tags: Optional[list[str]], output: Optional[pathlib
 
 @click.command()
 @click.pass_context
-@click.option("--tags", "-t", help="List of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags)
+@click.option(
+    "--tags", "-t", default=None, help="List of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags, show_default=True
+)
 @click.option("--search", "-s", help="Regular expression to search in both name and test", type=str, required=False)
 @click.option("--skip-error", help="Hide tests in errors due to connectivity issue", default=False, is_flag=True, show_default=True, required=False)
-def text(ctx: click.Context, tags: Optional[list[str]], search: Optional[str], skip_error: bool) -> None:
+def text(ctx: click.Context, tags: list[str], search: str | None, skip_error: bool) -> None:
     """ANTA command to check network states with text result"""
     print_settings(ctx)
     with anta_progress_bar() as AntaTest.progress:
@@ -93,8 +94,10 @@ def text(ctx: click.Context, tags: Optional[list[str]], search: Optional[str], s
     required=False,
     help="Path to save report as a file",
 )
-@click.option("--tags", "-t", help="List of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags)
-def tpl_report(ctx: click.Context, tags: Optional[list[str]], template: pathlib.Path, output: Optional[pathlib.Path]) -> None:
+@click.option(
+    "--tags", "-t", default=None, help="List of tags using comma as separator: tag1,tag2,tag3", type=str, required=False, callback=parse_tags, show_default=True
+)
+def tpl_report(ctx: click.Context, tags: list[str], template: pathlib.Path, output: pathlib.Path | None) -> None:
     """ANTA command to check network state with templated report"""
     print_settings(ctx, template, output)
     with anta_progress_bar() as AntaTest.progress:

anta/cli/nrfu/utils.py

@@ -1,18 +1,16 @@
-#!/usr/bin/env python
 # Copyright (c) 2023 Arista Networks, Inc.
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the LICENSE file.
-# coding: utf-8 -*-
 
 """
 Utils functions to use with anta.cli.check.commands module.
 """
+from __future__ import annotations
 
 import json
 import logging
 import pathlib
 import re
-from typing import Optional
 
 import click
 import rich
@@ -27,7 +25,7 @@
 logger = logging.getLogger(__name__)
 
 
-def print_settings(context: click.Context, report_template: Optional[pathlib.Path] = None, report_output: Optional[pathlib.Path] = None) -> None:
+def print_settings(context: click.Context, report_template: pathlib.Path | None = None, report_output: pathlib.Path | None = None) -> None:
     """Print ANTA settings before running tests"""
     message = f"Running ANTA tests:\n- {context.obj['inventory']}\n- Tests catalog contains {len(context.obj['catalog'])} tests"
     if report_template:
@@ -38,7 +36,7 @@ def print_settings(context: click.Context, report_template: Optional[pathlib.Pat
     console.print()
 
 
-def print_table(results: ResultManager, device: Optional[str] = None, test: Optional[str] = None, group_by: Optional[str] = None) -> None:
+def print_table(results: ResultManager, device: str | None = None, test: str | None = None, group_by: str | None = None) -> None:
     """Print result in a table"""
     reporter = ReportTable()
     console.print()
@@ -54,7 +52,7 @@ def print_table(results: ResultManager, device: Optional[str] = None, test: Opti
         console.print(reporter.report_all(result_manager=results))
 
 
-def print_json(results: ResultManager, output: Optional[pathlib.Path] = None) -> None:
+def print_json(results: ResultManager, output: pathlib.Path | None = None) -> None:
     """Print result in a json format"""
     console.print()
     console.print(Panel("JSON results of all tests", style="cyan"))
@@ -64,7 +62,7 @@ def print_json(results: ResultManager, output: Optional[pathlib.Path] = None) ->
             fout.write(results.get_results(output_format="json"))
 
 
-def print_list(results: ResultManager, output: Optional[pathlib.Path] = None) -> None:
+def print_list(results: ResultManager, output: pathlib.Path | None = None) -> None:
     """Print result in a list"""
     console.print()
     console.print(Panel.fit("List results of all tests", style="cyan"))
@@ -74,17 +72,17 @@ def print_list(results: ResultManager, output: Optional[pathlib.Path] = None) ->
             fout.write(str(results.get_results(output_format="list")))
 
 
-def print_text(results: ResultManager, search: Optional[str] = None, skip_error: bool = False) -> None:
+def print_text(results: ResultManager, search: str | None = None, skip_error: bool = False) -> None:
     """Print results as simple text"""
     console.print()
-    regexp = re.compile(search if search else ".*")
+    regexp = re.compile(search or ".*")
     for line in results.get_results(output_format="list"):
         if any(regexp.match(entry) for entry in [line.name, line.test]) and (not skip_error or line.result != "error"):
             message = f" ({str(line.messages[0])})" if len(line.messages) > 0 else ""
             console.print(f"{line.name} :: {line.test} :: [{line.result}]{line.result.upper()}[/{line.result}]{message}", highlight=False)
 
 
-def print_jinja(results: ResultManager, template: pathlib.Path, output: Optional[pathlib.Path] = None) -> None:
+def print_jinja(results: ResultManager, template: pathlib.Path, output: pathlib.Path | None = None) -> None:
     """Print result based on template."""
     console.print()
     reporter = ReportJinja(template_path=template)

anta/device.py

@@ -22,7 +22,7 @@
 from httpx import ConnectError, HTTPError
 
 from anta import __DEBUG__
-from anta.models import DEFAULT_TAG, AntaCommand
+from anta.models import AntaCommand
 from anta.tools.misc import anta_log_exception, exc_to_str
 
 logger = logging.getLogger(__name__)
@@ -65,10 +65,6 @@ def __init__(self, name: str, tags: Optional[list[str]] = None, disable_cache: b
         if not disable_cache:
             self._init_cache()
 
-        # Ensure tag 'all' is always set
-        if DEFAULT_TAG not in self.tags:
-            self.tags.append(DEFAULT_TAG)
-
     def _init_cache(self) -> None:
         """
         Initialize cache for the device, can be overriden by subclasses to manipulate how it works

anta/models.py

@@ -33,7 +33,6 @@
 # https://stackoverflow.com/questions/74103528/type-hinting-an-instance-of-a-nested-class
 # N = TypeVar("N", bound="AntaTest.Input")
 
-DEFAULT_TAG = "all"
 
 # TODO - make this configurable - with an env var maybe?
 BLACKLIST_REGEX = [r"^reload.*", r"^conf\w*\s*(terminal|session)*", r"^wr\w*\s*\w+"]
@@ -289,6 +288,7 @@ class Input(BaseModel):
 
         model_config = ConfigDict(extra="forbid")
         result_overwrite: Optional[ResultOverwrite] = None
+        filters: Optional[Filters] = None
 
         class ResultOverwrite(BaseModel):
             """Test inputs model to overwrite result fields
@@ -303,6 +303,17 @@ class ResultOverwrite(BaseModel):
             categories: Optional[List[str]] = None
             custom_field: Optional[str] = None
 
+        class Filters(BaseModel):
+            """Runtime filters to map tests with list of tags or devices
+
+            Attributes:
+                devices: List of devices for the test.
+                tags: List of device's tags for the test.
+            """
+
+            devices: Optional[List[str]] = None
+            tags: Optional[List[str]] = None
+
     def __init__(
         self,
         device: AntaDevice,
@@ -476,8 +487,6 @@ def format_td(seconds: float, digits: int = 3) -> str:
             if self.result.result != "unset":
                 return self.result
 
-            # TODO maybe_skip decorators
-
             # Data
             if eos_data is not None:
                 self.save_commands_data(eos_data)

anta/runner.py

@@ -1,6 +1,7 @@
 # Copyright (c) 2023 Arista Networks, Inc.
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the LICENSE file.
+# pylint: disable=too-many-branches
 """
 ANTA runner function
 """
@@ -9,7 +10,7 @@
 import asyncio
 import itertools
 import logging
-from typing import Optional
+from typing import Union
 
 from anta.inventory import AntaInventory
 from anta.models import AntaTest
@@ -19,11 +20,16 @@
 logger = logging.getLogger(__name__)
 
 
+def filter_tags(tags_cli: Union[list[str], None], tags_device: list[str], tags_test: list[str]) -> bool:
+    """Implement filtering logic for tags"""
+    return (tags_cli is None or any(t for t in tags_cli if t in tags_device)) and any(t for t in tags_device if t in tags_test)
+
+
 async def main(
     manager: ResultManager,
     inventory: AntaInventory,
     tests: list[tuple[type[AntaTest], AntaTest.Input]],
-    tags: Optional[list[str]] = None,
+    tags: list[str],
     established_only: bool = True,
 ) -> None:
     """
@@ -67,13 +73,16 @@ async def main(
     for device, test in itertools.product(devices, tests):
         test_class = test[0]
         test_inputs = test[1]
-        try:
-            # Instantiate AntaTest object
-            test_instance = test_class(device=device, inputs=test_inputs)
-            coros.append(test_instance.test(eos_data=None))
-        except Exception as e:  # pylint: disable=broad-exception-caught
-            message = "Error when creating ANTA tests"
-            anta_log_exception(e, message, logger)
+        test_filters = test[1].get("filters", None) if test[1] is not None else None
+        test_tags = test_filters.get("tags", []) if test_filters is not None else []
+        if len(test_tags) == 0 or filter_tags(tags_cli=tags, tags_device=device.tags, tags_test=test_tags):
+            try:
+                # Instantiate AntaTest object
+                test_instance = test_class(device=device, inputs=test_inputs)
+                coros.append(test_instance.test(eos_data=None))
+            except Exception as e:  # pylint: disable=broad-exception-caught
+                message = "Error when creating ANTA tests"
+                anta_log_exception(e, message, logger)
 
     if AntaTest.progress is not None:
         AntaTest.nrfu_task = AntaTest.progress.add_task("Running NRFU Tests...", total=len(coros))