Skip to content

Commit

Permalink
refactor: cleanup pytest
Browse files Browse the repository at this point in the history
  • Loading branch information
mtache committed Sep 12, 2024
1 parent a69d679 commit c35f184
Show file tree
Hide file tree
Showing 3 changed files with 17 additions and 31 deletions.
16 changes: 0 additions & 16 deletions tests/lib/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@

from __future__ import annotations

from pathlib import Path
from typing import Any


Expand All @@ -14,11 +13,6 @@ def generate_test_ids_dict(val: dict[str, Any], key: str = "name") -> str:
return val.get(key, "unamed_test")


def generate_test_ids_list(val: list[dict[str, Any]], key: str = "name") -> list[str]:
"""generate_test_ids Helper to generate test ID for parametrize."""
return [entry.get(key, "unamed_test") for entry in val]


def generate_test_ids(data: list[dict[str, Any]]) -> list[str]:
"""Build id for a unit test of an AntaTest subclass.
Expand All @@ -29,13 +23,3 @@ def generate_test_ids(data: list[dict[str, Any]]) -> list[str]:
}
"""
return [f"{val['test'].module}.{val['test'].__name__}-{val['name']}" for val in data]


def default_anta_env() -> dict[str, str | None]:
"""Return a default_anta_environement which can be passed to a cliRunner.invoke method."""
return {
"ANTA_USERNAME": "anta",
"ANTA_PASSWORD": "formica",
"ANTA_INVENTORY": str(Path(__file__).parent.parent / "data" / "test_inventory.yml"),
"ANTA_CATALOG": str(Path(__file__).parent.parent / "data" / "test_catalog.yml"),
}
4 changes: 3 additions & 1 deletion tests/units/test_catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,9 @@ def test_parse(self, filename: str, file_format: Literal["yaml", "json"], tests:
assert inputs == catalog.tests[test_id].inputs

@pytest.mark.parametrize(("filename", "file_format", "tests"), INIT_CATALOG_PARAMS)
def test_from_list(self, tests: list[tuple[type[AntaTest], AntaTest.Input | dict[str, Any] | None]]) -> None:
def test_from_list( # pylint: disable=unused-argument
self, filename: str, file_format: Literal["yaml", "json"], tests: list[tuple[type[AntaTest], AntaTest.Input | dict[str, Any] | None]]
) -> None:
"""Instantiate AntaCatalog from a list."""
catalog: AntaCatalog = AntaCatalog.from_list(tests)

Expand Down
28 changes: 14 additions & 14 deletions tests/units/test_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
if TYPE_CHECKING:
from _pytest.mark.structures import ParameterSet

INIT_DATA: list[ParameterSet] = [
INIT_PARAMS: list[ParameterSet] = [
pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta"}, {"name": "42.42.42.42"}, id="no name, no port"),
pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta", "port": 666}, {"name": "42.42.42.42:666"}, id="no name, port"),
pytest.param(
Expand All @@ -33,7 +33,7 @@
{"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "test.anta.ninja", "insecure": True}, {"name": "test.anta.ninja"}, id="insecure"
),
]
EQUALITY_DATA: list[ParameterSet] = [
EQUALITY_PARAMS: list[ParameterSet] = [
pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta"}, {"host": "42.42.42.42", "username": "anta", "password": "blah"}, True, id="equal"),
pytest.param(
{"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "device1"},
Expand All @@ -51,7 +51,7 @@
{"host": "42.42.42.41", "username": "anta", "password": "anta"}, {"host": "42.42.42.42", "username": "anta", "password": "anta"}, False, id="not-equal-host"
),
]
ASYNCEAPI_COLLECT_DATA: list[ParameterSet] = [
ASYNCEAPI_COLLECT_PARAMS: list[ParameterSet] = [
pytest.param(
{},
{
Expand Down Expand Up @@ -302,12 +302,12 @@
id="httpx.ConnectError",
),
]
ASYNCEAPI_COPY_DATA: list[ParameterSet] = [
ASYNCEAPI_COPY_PARAMS: list[ParameterSet] = [
pytest.param({}, {"sources": [Path("/mnt/flash"), Path("/var/log/agents")], "destination": Path(), "direction": "from"}, id="from"),
pytest.param({}, {"sources": [Path("/mnt/flash"), Path("/var/log/agents")], "destination": Path(), "direction": "to"}, id="to"),
pytest.param({}, {"sources": [Path("/mnt/flash"), Path("/var/log/agents")], "destination": Path(), "direction": "wrong"}, id="wrong"),
]
REFRESH_DATA: list[ParameterSet] = [
REFRESH_PARAMS: list[ParameterSet] = [
pytest.param(
{},
(
Expand Down Expand Up @@ -431,7 +431,7 @@
id="httpx.ConnectError",
),
]
COLLECT_DATA: list[ParameterSet] = [
COLLECT_PARAMS: list[ParameterSet] = [
pytest.param(
{"disable_cache": False},
{"command": "show version", "use_cache": True},
Expand Down Expand Up @@ -459,7 +459,7 @@
),
pytest.param({"disable_cache": True}, {"command": "show version", "use_cache": False}, {}, id="device cache disabled, command cache disabled"),
]
CACHE_STATS_DATA: list[ParameterSet] = [
CACHE_STATS_PARAMS: list[ParameterSet] = [
pytest.param({"disable_cache": False}, {"total_commands_sent": 0, "cache_hits": 0, "cache_hit_ratio": "0.00%"}, id="with_cache"),
pytest.param({"disable_cache": True}, None, id="without_cache"),
]
Expand All @@ -468,7 +468,7 @@
class TestAntaDevice:
"""Test for anta.device.AntaDevice Abstract class."""

@pytest.mark.parametrize(("device", "command", "expected"), COLLECT_DATA, indirect=["device"])
@pytest.mark.parametrize(("device", "command", "expected"), COLLECT_PARAMS, indirect=["device"])
async def test_collect(self, device: AntaDevice, command: dict[str, Any], expected: dict[str, Any]) -> None:
"""Test AntaDevice.collect behavior."""
cmd = AntaCommand(command=command["command"], use_cache=command["use_cache"])
Expand Down Expand Up @@ -503,7 +503,7 @@ async def test_collect(self, device: AntaDevice, command: dict[str, Any], expect
assert device.cache is None
device._collect.assert_called_once_with(command=cmd, collection_id=None) # type: ignore[attr-defined] # pylint: disable=protected-access

@pytest.mark.parametrize(("device", "expected"), CACHE_STATS_DATA, indirect=["device"])
@pytest.mark.parametrize(("device", "expected"), CACHE_STATS_PARAMS, indirect=["device"])
def test_cache_statistics(self, device: AntaDevice, expected: dict[str, Any] | None) -> None:
"""Verify that when cache statistics attribute does not exist.
Expand All @@ -515,7 +515,7 @@ def test_cache_statistics(self, device: AntaDevice, expected: dict[str, Any] | N
class TestAsyncEOSDevice:
"""Test for anta.device.AsyncEOSDevice."""

@pytest.mark.parametrize(("device", "expected"), INIT_DATA)
@pytest.mark.parametrize(("device", "expected"), INIT_PARAMS)
def test__init__(self, device: dict[str, Any], expected: dict[str, Any]) -> None:
"""Test the AsyncEOSDevice constructor."""
dev = AsyncEOSDevice(**device)
Expand All @@ -532,7 +532,7 @@ def test__init__(self, device: dict[str, Any], expected: dict[str, Any]) -> None
with patch("anta.device.__DEBUG__", new=True):
rprint(dev)

@pytest.mark.parametrize(("device1", "device2", "expected"), EQUALITY_DATA)
@pytest.mark.parametrize(("device1", "device2", "expected"), EQUALITY_PARAMS)
def test__eq(self, device1: dict[str, Any], device2: dict[str, Any], expected: bool) -> None:
"""Test the AsyncEOSDevice equality."""
dev1 = AsyncEOSDevice(**device1)
Expand All @@ -544,7 +544,7 @@ def test__eq(self, device1: dict[str, Any], device2: dict[str, Any], expected: b

@pytest.mark.parametrize(
("async_device", "patch_kwargs", "expected"),
REFRESH_DATA,
REFRESH_PARAMS,
indirect=["async_device"],
)
async def test_refresh(self, async_device: AsyncEOSDevice, patch_kwargs: list[dict[str, Any]], expected: dict[str, Any]) -> None:
Expand All @@ -561,7 +561,7 @@ async def test_refresh(self, async_device: AsyncEOSDevice, patch_kwargs: list[di

@pytest.mark.parametrize(
("async_device", "command", "expected"),
ASYNCEAPI_COLLECT_DATA,
ASYNCEAPI_COLLECT_PARAMS,
indirect=["async_device"],
)
async def test__collect(self, async_device: AsyncEOSDevice, command: dict[str, Any], expected: dict[str, Any]) -> None:
Expand Down Expand Up @@ -592,7 +592,7 @@ async def test__collect(self, async_device: AsyncEOSDevice, command: dict[str, A

@pytest.mark.parametrize(
("async_device", "copy"),
ASYNCEAPI_COPY_DATA,
ASYNCEAPI_COPY_PARAMS,
indirect=["async_device"],
)
async def test_copy(self, async_device: AsyncEOSDevice, copy: dict[str, Any]) -> None:
Expand Down

0 comments on commit c35f184

Please sign in to comment.