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
|
# 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.
"""See https://docs.pytest.org/en/stable/reference/fixtures.html#conftest-py-sharing-fixtures-across-multiple-files."""
import asyncio
from collections.abc import Iterator
from pathlib import Path
from unittest.mock import AsyncMock, Mock, patch
import pytest
import respx
from anta.device import AsyncEOSDevice
from anta.inventory import AntaInventory
DATA_DIR: Path = Path(__file__).parent.resolve() / "data"
@pytest.fixture
def inventory(request: pytest.FixtureRequest) -> Iterator[AntaInventory]:
"""Generate an ANTA inventory."""
user = "admin"
password = "password" # noqa: S105
params = request.param if hasattr(request, "param") else {}
count = params.get("count", 1)
disable_cache = params.get("disable_cache", True)
reachable = params.get("reachable", True)
if "filename" in params:
inv = AntaInventory.parse(DATA_DIR / params["filename"], username=user, password=password, disable_cache=disable_cache)
else:
inv = AntaInventory()
for i in range(count):
inv.add_device(
AsyncEOSDevice(
host=f"device-{i}.anta.arista.com",
username=user,
password=password,
name=f"device-{i}",
disable_cache=disable_cache,
)
)
if reachable:
# This context manager makes all devices reachable
with patch("asyncio.open_connection", AsyncMock(spec=asyncio.open_connection, return_value=(Mock(), Mock()))), respx.mock:
respx.post(path="/command-api", headers={"Content-Type": "application/json-rpc"}, json__params__cmds__0__cmd="show version").respond(
json={
"result": [
{
"modelName": "pytest",
}
],
}
)
yield inv
else:
with patch("asyncio.open_connection", AsyncMock(spec=asyncio.open_connection, side_effect=TimeoutError)):
yield inv
|
