1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
|
# Copyright (c) 2023-2024 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 = redefined-outer-name
"""Click commands to execute EOS commands on remote devices."""
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING, Literal
import click
from anta.cli.console import console
from anta.cli.debug.utils import debug_options
from anta.cli.utils import ExitCode
from anta.models import AntaCommand, AntaTemplate
if TYPE_CHECKING:
from anta.device import AntaDevice
logger = logging.getLogger(__name__)
@click.command
@debug_options
@click.pass_context
@click.option("--command", "-c", type=str, required=True, help="Command to run")
def run_cmd(
ctx: click.Context,
device: AntaDevice,
command: str,
ofmt: Literal["json", "text"],
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
v: Literal[1, "latest"] = version if version == "latest" else 1
c = AntaCommand(command=command, ofmt=ofmt, version=v, revision=revision)
asyncio.run(device.collect(c))
if not c.collected:
console.print(f"[bold red] Command '{c.command}' failed to execute!")
ctx.exit(ExitCode.USAGE_ERROR)
elif ofmt == "json":
console.print(c.json_output)
elif ofmt == "text":
console.print(c.text_output)
@click.command
@debug_options
@click.pass_context
@click.option(
"--template",
"-t",
type=str,
required=True,
help="Command template to run. E.g. 'show vlan {vlan_id}'",
)
@click.argument("params", required=True, nargs=-1)
def run_template(
ctx: click.Context,
device: AntaDevice,
template: str,
params: list[str],
ofmt: Literal["json", "text"],
version: Literal["1", "latest"],
revision: int,
) -> None:
# pylint: disable=too-many-arguments
"""Run arbitrary templated command to an ANTA device.
Takes a list of arguments (keys followed by a value) to build a dictionary used as template parameters.
Example:
-------
anta debug run-template -d leaf1a -t 'show vlan {vlan_id}' vlan_id 1
"""
template_params = dict(zip(params[::2], params[1::2]))
console.print(f"Run templated command [blue]'{template}'[/blue] with [orange]{template_params}[/orange] on [red]{device.name}[/red]")
# I do not assume the following line, but click make me do it
v: Literal[1, "latest"] = version if version == "latest" else 1
t = AntaTemplate(template=template, ofmt=ofmt, version=v, revision=revision)
c = t.render(**template_params)
asyncio.run(device.collect(c))
if not c.collected:
console.print(f"[bold red] Command '{c.command}' failed to execute!")
ctx.exit(ExitCode.USAGE_ERROR)
elif ofmt == "json":
console.print(c.json_output)
elif ofmt == "text":
console.print(c.text_output)
|