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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
# Copyright (c) 2023-2025 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the LICENSE file.
"""See https://docs.pytest.org/en/stable/reference/fixtures.html#conftest-py-sharing-fixtures-across-multiple-files."""
from __future__ import annotations
import logging
import shutil
from typing import TYPE_CHECKING, Any, Literal
from unittest.mock import patch
import pytest
from click.testing import CliRunner, Result
import asynceapi
from anta.cli.console import console
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
logger = logging.getLogger(__name__)
MOCK_CLI_JSON: dict[str, asynceapi.EapiCommandError | dict[str, Any]] = {
"show version": {
"modelName": "DCS-7280CR3-32P4-F",
"version": "4.31.1F",
},
"enable": {},
"clear counters": {},
"clear hardware counter drop": {},
"undefined": asynceapi.EapiCommandError(
passed=[],
failed="show version",
errors=["Authorization denied for command 'show version'"],
errmsg="Invalid command",
not_exec=[],
),
"show interfaces": {"interfaces": {}},
"ping": { # NOTE: yeah this is the same for every one byt that's mocking.
"messages": [
(
"PING 10.0.0.2 (10.0.0.2) from 10.0.0.5 : 72(100) bytes of data.\n"
"80 bytes from 10.0.0.2: icmp_seq=1 ttl=64 time=0.247 ms\n"
"80 bytes from 10.0.0.2: icmp_seq=2 ttl=64 time=0.072 ms\n\n"
"--- 10.0.0.2 ping statistics ---\n"
"2 packets transmitted, 2 received, 0% packet loss, time 0ms\n"
"rtt min/avg/max/mdev = 0.072/0.159/0.247/0.088 ms, ipg/ewma 0.370/0.225 ms\n\n"
)
]
},
}
MOCK_CLI_TEXT: dict[str, asynceapi.EapiCommandError | str] = {
"show version": "Arista cEOSLab",
"bash timeout 10 ls -1t /mnt/flash/schedule/tech-support": "dummy_tech-support_2023-12-01.1115.log.gz\ndummy_tech-support_2023-12-01.1015.log.gz",
"bash timeout 10 ls -1t /mnt/flash/schedule/tech-support | head -1": "dummy_tech-support_2023-12-01.1115.log.gz",
"show running-config | include aaa authorization exec default": "aaa authorization exec default local",
}
@pytest.fixture
def temp_env(anta_env: dict[str, str], tmp_path: Path) -> dict[str, str]:
"""Fixture that create a temporary ANTA inventory.
The inventory can be overridden and returns the corresponding environment variables.
"""
anta_inventory = str(anta_env["ANTA_INVENTORY"])
temp_inventory = tmp_path / "test_inventory.yml"
shutil.copy(anta_inventory, temp_inventory)
anta_env["ANTA_INVENTORY"] = str(temp_inventory)
return anta_env
@pytest.fixture
# Disabling C901 - too complex as we like our runner like this
def click_runner(capsys: pytest.CaptureFixture[str], anta_env: dict[str, str]) -> Iterator[CliRunner]: # noqa: C901
"""Return a click.CliRunner for cli testing."""
class AntaCliRunner(CliRunner):
"""Override CliRunner to inject specific variables for ANTA."""
def invoke(self, *args: Any, **kwargs: Any) -> Result: # noqa: ANN401
# Inject default env vars if not provided
kwargs["env"] = anta_env | kwargs.get("env", {})
# Deterministic terminal width
kwargs["env"]["COLUMNS"] = "165"
kwargs["auto_envvar_prefix"] = "ANTA"
# Way to fix https://github.com/pallets/click/issues/824
with capsys.disabled():
result = super().invoke(*args, **kwargs)
# disabling T201 as we want to print here
print("--- CLI Output ---") # noqa: T201
print(result.output) # noqa: T201
return result
def cli(
command: str | None = None,
commands: list[dict[str, Any]] | None = None,
ofmt: Literal["json", "text"] = "json",
_version: int | str | None = "latest",
**_kwargs: Any, # noqa: ANN401
) -> dict[str, Any] | list[dict[str, Any]]:
def get_output(command: str | dict[str, Any]) -> dict[str, Any]:
if isinstance(command, dict):
command_str: str = command["cmd"]
else:
command_str = command
mock_cli: dict[str, Any]
if ofmt == "json":
mock_cli = MOCK_CLI_JSON
elif ofmt == "text":
mock_cli = MOCK_CLI_TEXT
for mock_cmd, output in mock_cli.items():
if command_str == mock_cmd or (command_str.startswith("ping") and mock_cmd == "ping"):
logger.info("Mocking command %s", mock_cmd)
if isinstance(output, asynceapi.EapiCommandError):
raise output
return output
message = f"Command '{command_str}' is not mocked"
logger.critical(message)
raise NotImplementedError(message)
res: dict[str, Any] | list[dict[str, Any]]
if command is not None:
logger.debug("Mock input %s", command)
res = get_output(command)
elif commands is not None:
logger.debug("Mock input %s", commands)
res = list(map(get_output, commands))
else:
msg = "Either 'command' or 'commands' must not be None"
raise ValueError(msg)
logger.debug("Mock output %s", res)
return res
# Patch asynceapi methods used by AsyncEOSDevice. See tests/units/test_device.py
with (
patch("asynceapi.device.Device.check_api_endpoint", return_value=True),
patch("asynceapi.device.Device.cli", side_effect=cli),
patch("asyncssh.connect"),
patch(
"asyncssh.scp",
),
):
console._color_system = None
yield AntaCliRunner()
|
