Skip to content

Commit

Permalink
feat(anta.cli): anta nrfu enchancements: add --device and `--test…
Browse files Browse the repository at this point in the history
…` to filter tests and `--hide (error|failure|success|skipped)` to hide results. (#588)

* feat(anta): Implement `--device` and `--test` filters in `anta nrfu`
* feat(anta.cli): Implement `--hide` options in `anta nrfu`
* chore: Do not ignore TRY004 globally
* refactor(anta): anta.result_manager and anta.reporter
* refactor(anta): anta.tools
* refactor(anta): add set types where it makes sense

---------

Co-authored-by: Matthieu Tâche <[email protected]>
  • Loading branch information
titom73 and mtache authored Mar 29, 2024
1 parent 28c0e4f commit b2dadb1
Show file tree
Hide file tree
Showing 53 changed files with 1,368 additions and 1,313 deletions.
26 changes: 24 additions & 2 deletions anta/catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -336,11 +336,20 @@ def from_list(data: ListAntaTestTuples) -> AntaCatalog:
raise
return AntaCatalog(tests)

def get_tests_by_tags(self, tags: list[str], *, strict: bool = False) -> list[AntaTestDefinition]:
def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> list[AntaTestDefinition]:
"""Return all the tests that have matching tags in their input filters.
If strict=True, returns only tests that match all the tags provided as input.
If strict=True, return only tests that match all the tags provided as input.
If strict=False, return all the tests that match at least one tag provided as input.
Args:
----
tags: Tags of the tests to get.
strict: Specify if the returned tests must match all the tags provided.
Returns
-------
List of AntaTestDefinition that match the tags
"""
result: list[AntaTestDefinition] = []
for test in self.tests:
Expand All @@ -351,3 +360,16 @@ def get_tests_by_tags(self, tags: list[str], *, strict: bool = False) -> list[An
elif any(t in tags for t in f):
result.append(test)
return result

def get_tests_by_names(self, names: set[str]) -> list[AntaTestDefinition]:
"""Return all the tests that have matching a list of tests names.
Args:
----
names: Names of the tests to get.
Returns
-------
List of AntaTestDefinition that match the names
"""
return [test for test in self.tests if test.test.name in names]
1 change: 1 addition & 0 deletions anta/cli/debug/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ def run_cmd(
version: Literal["1", "latest"],
revision: int,
) -> None:
# pylint: disable=too-many-arguments
"""Run arbitrary command to an ANTA device."""
console.print(f"Run command [green]{command}[/green] on [red]{device.name}[/red]")
# I do not assume the following line, but click make me do it
Expand Down
2 changes: 1 addition & 1 deletion anta/cli/debug/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ def wrapper(
ctx: click.Context,
*args: tuple[Any],
inventory: AntaInventory,
tags: list[str] | None,
tags: set[str] | None,
device: str,
**kwargs: Any,
) -> Any:
Expand Down
6 changes: 3 additions & 3 deletions anta/cli/exec/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@

@click.command
@inventory_options
def clear_counters(inventory: AntaInventory, tags: list[str] | None) -> None:
def clear_counters(inventory: AntaInventory, tags: set[str] | None) -> None:
"""Clear counter statistics on EOS devices."""
asyncio.run(clear_counters_utils(inventory, tags=tags))

Expand All @@ -51,7 +51,7 @@ def clear_counters(inventory: AntaInventory, tags: list[str] | None) -> None:
default=f"anta_snapshot_{datetime.now(tz=timezone.utc).astimezone().strftime('%Y-%m-%d_%H_%M_%S')}",
show_default=True,
)
def snapshot(inventory: AntaInventory, tags: list[str] | None, commands_list: Path, output: Path) -> None:
def snapshot(inventory: AntaInventory, tags: set[str] | None, commands_list: Path, output: Path) -> None:
"""Collect commands output from devices in inventory."""
console.print(f"Collecting data for {commands_list}")
console.print(f"Output directory is {output}")
Expand Down Expand Up @@ -91,7 +91,7 @@ def snapshot(inventory: AntaInventory, tags: list[str] | None, commands_list: Pa
)
def collect_tech_support(
inventory: AntaInventory,
tags: list[str] | None,
tags: set[str] | None,
output: Path,
latest: int | None,
*,
Expand Down
12 changes: 6 additions & 6 deletions anta/cli/exec/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
logger = logging.getLogger(__name__)


async def clear_counters_utils(anta_inventory: AntaInventory, tags: list[str] | None = None) -> None:
async def clear_counters_utils(anta_inventory: AntaInventory, tags: set[str] | None = None) -> None:
"""Clear counters."""

async def clear(dev: AntaDevice) -> None:
Expand All @@ -44,7 +44,7 @@ async def clear(dev: AntaDevice) -> None:

logger.info("Connecting to devices...")
await anta_inventory.connect_inventory()
devices = anta_inventory.get_inventory(established_only=True, tags=tags).values()
devices = anta_inventory.get_inventory(established_only=True, tags=tags).devices
logger.info("Clearing counters on remote devices...")
await asyncio.gather(*(clear(device) for device in devices))

Expand All @@ -53,7 +53,7 @@ async def collect_commands(
inv: AntaInventory,
commands: dict[str, str],
root_dir: Path,
tags: list[str] | None = None,
tags: set[str] | None = None,
) -> None:
"""Collect EOS commands."""

Expand All @@ -78,7 +78,7 @@ async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "tex

logger.info("Connecting to devices...")
await inv.connect_inventory()
devices = inv.get_inventory(established_only=True, tags=tags).values()
devices = inv.get_inventory(established_only=True, tags=tags).devices
logger.info("Collecting commands from remote devices")
coros = []
if "json_format" in commands:
Expand All @@ -91,7 +91,7 @@ async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "tex
logger.error("Error when collecting commands: %s", str(r))


async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, *, configure: bool, tags: list[str] | None = None, latest: int | None = None) -> None:
async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, *, configure: bool, tags: set[str] | None = None, latest: int | None = None) -> None:
"""Collect scheduled show-tech on devices."""

async def collect(device: AntaDevice) -> None:
Expand Down Expand Up @@ -154,5 +154,5 @@ async def collect(device: AntaDevice) -> None:

logger.info("Connecting to devices...")
await inv.connect_inventory()
devices = inv.get_inventory(established_only=True, tags=tags).values()
devices = inv.get_inventory(established_only=True, tags=tags).devices
await asyncio.gather(*(collect(device) for device in devices))
15 changes: 8 additions & 7 deletions anta/cli/get/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
import json
import logging
from pathlib import Path
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any

import click
from cvprac.cvp_client import CvpClient
Expand All @@ -37,6 +37,7 @@
@click.option("--password", "-p", help="CloudVision password", type=str, required=True)
@click.option("--container", "-c", help="CloudVision container where devices are configured", type=str)
def from_cvp(ctx: click.Context, output: Path, host: str, username: str, password: str, container: str | None) -> None:
# pylint: disable=too-many-arguments
"""Build ANTA inventory from Cloudvision.
TODO - handle get_inventory and get_devices_in_container failure
Expand Down Expand Up @@ -91,7 +92,7 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i
@click.command
@inventory_options
@click.option("--connected/--not-connected", help="Display inventory after connection has been created", default=False, required=False)
def inventory(inventory: AntaInventory, tags: list[str] | None, *, connected: bool) -> None:
def inventory(inventory: AntaInventory, tags: set[str] | None, *, connected: bool) -> None:
"""Show inventory loaded in ANTA."""
# TODO: @gmuloc - tags come from context - we cannot have everything..
# ruff: noqa: ARG001
Expand All @@ -107,11 +108,11 @@ def inventory(inventory: AntaInventory, tags: list[str] | None, *, connected: bo

@click.command
@inventory_options
def tags(inventory: AntaInventory, tags: list[str] | None) -> None: # pylint: disable=unused-argument
def tags(inventory: AntaInventory, **kwargs: Any) -> None:
# pylint: disable=unused-argument
"""Get list of configured tags in user inventory."""
tags_found = []
tags: set[str] = set()
for device in inventory.values():
tags_found += device.tags
tags_found = sorted(set(tags_found))
tags.update(device.tags)
console.print("Tags found:")
console.print_json(json.dumps(tags_found, indent=2))
console.print_json(json.dumps(sorted(tags), indent=2))
2 changes: 1 addition & 1 deletion anta/cli/get/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ def create_inventory_from_cvp(inv: list[dict[str, Any]], output: Path) -> None:
AntaInventoryHost(
name=dev["hostname"],
host=dev["ipAddress"],
tags=[dev["containerName"].lower()],
tags={dev["containerName"].lower()},
)
)
write_inventory_to_file(hosts, output)
Expand Down
65 changes: 59 additions & 6 deletions anta/cli/nrfu/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,13 @@
from __future__ import annotations

import asyncio
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, get_args

import click

from anta.cli.nrfu import commands
from anta.cli.utils import AliasedGroup, catalog_options, inventory_options
from anta.custom_types import TestStatus
from anta.models import AntaTest
from anta.result_manager import ResultManager
from anta.runner import main
Expand Down Expand Up @@ -52,14 +53,65 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
return super().parse_args(ctx, args)


HIDE_STATUS: list[str] = list(get_args(TestStatus))
HIDE_STATUS.remove("unset")


@click.group(invoke_without_command=True, cls=IgnoreRequiredWithHelp)
@click.pass_context
@inventory_options
@catalog_options
@click.option("--ignore-status", help="Always exit with success", show_envvar=True, is_flag=True, default=False)
@click.option("--ignore-error", help="Only report failures and not errors", show_envvar=True, is_flag=True, default=False)
def nrfu(ctx: click.Context, inventory: AntaInventory, tags: list[str] | None, catalog: AntaCatalog, *, ignore_status: bool, ignore_error: bool) -> None:
"""Run ANTA tests on devices."""
@click.option(
"--device",
"-d",
help="Run tests on a specific device. Can be provided multiple times.",
type=str,
multiple=True,
required=False,
)
@click.option(
"--test",
help="Run a specific test. Can be provided multiple times.",
type=str,
multiple=True,
required=False,
)
@click.option(
"--ignore-status",
help="Exit code will always be 0.",
show_envvar=True,
is_flag=True,
default=False,
)
@click.option(
"--ignore-error",
help="Exit code will be 0 if all tests succeeded or 1 if any test failed.",
show_envvar=True,
is_flag=True,
default=False,
)
@click.option(
"--hide",
default=None,
type=click.Choice(HIDE_STATUS, case_sensitive=False),
multiple=True,
help="Group result by test or device.",
required=False,
)
# pylint: disable=too-many-arguments
def nrfu(
ctx: click.Context,
inventory: AntaInventory,
tags: set[str] | None,
catalog: AntaCatalog,
device: tuple[str],
test: tuple[str],
hide: tuple[str],
*,
ignore_status: bool,
ignore_error: bool,
) -> None:
"""Run ANTA tests on selected inventory devices."""
# If help is invoke somewhere, skip the command
if ctx.obj.get("_anta_help"):
return
Expand All @@ -68,9 +120,10 @@ def nrfu(ctx: click.Context, inventory: AntaInventory, tags: list[str] | None, c
ctx.obj["result_manager"] = ResultManager()
ctx.obj["ignore_status"] = ignore_status
ctx.obj["ignore_error"] = ignore_error
ctx.obj["hide"] = set(hide) if hide else None
print_settings(inventory, catalog)
with anta_progress_bar() as AntaTest.progress:
asyncio.run(main(ctx.obj["result_manager"], inventory, catalog, tags=tags))
asyncio.run(main(ctx.obj["result_manager"], inventory, catalog, tags=tags, devices=set(device) if device else None, tests=set(test) if test else None))
# Invoke `anta nrfu table` if no command is passed
if ctx.invoked_subcommand is None:
ctx.invoke(commands.table)
Expand Down
20 changes: 10 additions & 10 deletions anta/cli/nrfu/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

import logging
import pathlib
from typing import Literal

import click

Expand All @@ -19,18 +20,19 @@

@click.command()
@click.pass_context
@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",
help="Group result by test or device.",
required=False,
)
def table(ctx: click.Context, device: str | None, test: str | None, group_by: str) -> None:
def table(
ctx: click.Context,
group_by: Literal["device", "test"] | None,
) -> None:
"""ANTA command to check network states with table result."""
print_table(results=ctx.obj["result_manager"], device=device, group_by=group_by, test=test)
print_table(ctx, group_by=group_by)
exit_with_code(ctx)


Expand All @@ -46,17 +48,15 @@ def table(ctx: click.Context, device: str | None, test: str | None, group_by: st
)
def json(ctx: click.Context, output: pathlib.Path | None) -> None:
"""ANTA command to check network state with JSON result."""
print_json(results=ctx.obj["result_manager"], output=output)
print_json(ctx, output=output)
exit_with_code(ctx)


@click.command()
@click.pass_context
@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, search: str | None, *, skip_error: bool) -> None:
def text(ctx: click.Context) -> None:
"""ANTA command to check network states with text result."""
print_text(results=ctx.obj["result_manager"], search=search, skip_error=skip_error)
print_text(ctx)
exit_with_code(ctx)


Expand Down
Loading

0 comments on commit b2dadb1

Please sign in to comment.