diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 4e981ae8b..4d87bbc0d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -69,6 +69,7 @@ repos: - types-pyOpenSSL - pylint_pydantic - pytest + - pytest-asyncio - repo: https://github.com/codespell-project/codespell rev: v2.3.0 diff --git a/pyproject.toml b/pyproject.toml index e8b4feba2..4b34fa152 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -66,8 +66,9 @@ dev = [ "pylint-pydantic>=0.2.4", "pylint>=2.17.5", "pytest-asyncio>=0.21.1", + "pytest-benchmark>=4.0.0", "pytest-cov>=4.1.0", - "pytest-dependency", + "pytest-dependency>=0.5.1", "pytest-html>=3.2.0", "pytest-metadata>=3.0.0", "pytest>=7.4.0", @@ -381,6 +382,9 @@ runtime-evaluated-base-classes = ["pydantic.BaseModel", "anta.models.AntaTest.In "S106", # Passwords are indeed hardcoded in tests "S108", # Probable insecure usage of temporary file or directory ] +"tests/lib/fixture.py" = [ + "ANN401", # Ok to use Any type hint in decorators +] "tests/units/anta_tests/test_interfaces.py" = [ "S104", # False positive for 0.0.0.0 bindings in test inputs ] diff --git a/tests/lib/fixture.py b/tests/lib/fixture.py index 17943edc3..b73ed4c00 100644 --- a/tests/lib/fixture.py +++ b/tests/lib/fixture.py @@ -5,15 +5,19 @@ from __future__ import annotations +import asyncio import logging import shutil +from pathlib import Path from typing import TYPE_CHECKING, Any, Callable from unittest.mock import patch import pytest +import pytest_asyncio from click.testing import CliRunner, Result import asynceapi +from anta.catalog import AntaCatalog from anta.cli.console import console from anta.device import AntaDevice, AsyncEOSDevice from anta.inventory import AntaInventory @@ -23,7 +27,6 @@ if TYPE_CHECKING: from collections.abc import Iterator - from pathlib import Path from anta.models import AntaCommand @@ -62,7 +65,7 @@ def device(request: pytest.FixtureRequest) -> Iterator[AntaDevice]: """Return an AntaDevice instance with mocked abstract method.""" - def _collect(command: AntaCommand, *args: Any, **kwargs: Any) -> None: # noqa: ARG001, ANN401 #pylint: disable=unused-argument + def _collect(command: AntaCommand, *args: Any, **kwargs: Any) -> None: # noqa: ARG001 # pylint: disable=unused-argument command.output = COMMAND_OUTPUT kwargs = {"name": DEVICE_NAME, "hw_model": DEVICE_HW_MODEL} @@ -79,9 +82,12 @@ def _collect(command: AntaCommand, *args: Any, **kwargs: Any) -> None: # noqa: @pytest.fixture() -def test_inventory() -> AntaInventory: +def test_inventory(request: pytest.FixtureRequest) -> AntaInventory: """Return the test_inventory.""" env = default_anta_env() + if hasattr(request, "param"): + # Fixture is parametrized indirectly with a specific test inventory filename + env["ANTA_INVENTORY"] = str(Path(__file__).parent.parent / "data" / request.param) assert env["ANTA_INVENTORY"] assert env["ANTA_USERNAME"] assert env["ANTA_PASSWORD"] is not None @@ -92,6 +98,17 @@ def test_inventory() -> AntaInventory: ) +@pytest.fixture() +def test_catalog(request: pytest.FixtureRequest) -> AntaCatalog: + """Return the test_catalog.""" + env = default_anta_env() + if hasattr(request, "param"): + # Fixture is parametrized indirectly with a specific test inventory filename + env["ANTA_CATALOG"] = str(Path(__file__).parent.parent / "data" / request.param) + assert env["ANTA_CATALOG"] + return AntaCatalog.parse(filename=env["ANTA_CATALOG"]) + + # tests.unit.test_device.py fixture @pytest.fixture() def async_device(request: pytest.FixtureRequest) -> AsyncEOSDevice: @@ -179,8 +196,8 @@ class AntaCliRunner(CliRunner): def invoke( self, - *args: Any, # noqa: ANN401 - **kwargs: Any, # noqa: ANN401 + *args: Any, + **kwargs: Any, ) -> Result: # Inject default env if not provided kwargs["env"] = kwargs["env"] if "env" in kwargs else default_anta_env() @@ -201,7 +218,7 @@ def cli( commands: list[dict[str, Any]] | None = None, ofmt: str = "json", _version: int | str | None = "latest", - **_kwargs: Any, # noqa: ANN401 + **_kwargs: Any, ) -> dict[str, Any] | list[dict[str, Any]]: def get_output(command: str | dict[str, Any]) -> dict[str, Any]: if isinstance(command, dict): @@ -242,3 +259,26 @@ def get_output(command: str | dict[str, Any]) -> dict[str, Any]: ): console._color_system = None # pylint: disable=protected-access yield AntaCliRunner() + + +@pytest_asyncio.fixture() +async def aio_benchmark(benchmark: Callable[..., Any]) -> Callable[..., Any]: + """Fixture to benchmark a coroutine function. + + https://github.com/ionelmc/pytest-benchmark/issues/66#issuecomment-2058337929 + """ + + async def run_async_coroutine(func: Any, *args: Any, **kwargs: Any) -> Any: + return await func(*args, **kwargs) + + def _wrapper(func: Any, *args: Any, **kwargs: Any) -> Any: + if asyncio.iscoroutinefunction(func): + + @benchmark + def _() -> asyncio.Future[None]: + future = asyncio.ensure_future(run_async_coroutine(func, *args, **kwargs)) + return asyncio.get_event_loop().run_until_complete(future) + else: + benchmark(func, *args, **kwargs) + + return _wrapper diff --git a/tests/lib/utils.py b/tests/lib/utils.py index ba669c287..bf0f2d48c 100644 --- a/tests/lib/utils.py +++ b/tests/lib/utils.py @@ -32,7 +32,7 @@ def generate_test_ids(data: list[dict[str, Any]]) -> list[str]: def default_anta_env() -> dict[str, str | None]: - """Return a default_anta_environement which can be passed to a cliRunner.invoke method.""" + """Return a default ANTA environment used for unit testing.""" return { "ANTA_USERNAME": "anta", "ANTA_PASSWORD": "formica", diff --git a/tests/units/test_runner.py b/tests/units/test_runner.py index 955149d09..4a334ca46 100644 --- a/tests/units/test_runner.py +++ b/tests/units/test_runner.py @@ -8,6 +8,7 @@ import logging import resource from pathlib import Path +from typing import TYPE_CHECKING from unittest.mock import patch import pytest @@ -18,10 +19,30 @@ from anta.result_manager import ResultManager from anta.runner import adjust_rlimit_nofile, main, prepare_tests -from .test_models import FakeTest +if TYPE_CHECKING: + from typing import Any, Callable DATA_DIR: Path = Path(__file__).parent.parent.resolve() / "data" -FAKE_CATALOG: AntaCatalog = AntaCatalog.from_list([(FakeTest, None)]) + + +@pytest.mark.parametrize( + ("test_inventory", "test_catalog"), + [ + pytest.param("test_inventory_large.yml", "test_catalog_large.yml", id="large-50_devices-7688_tests"), + pytest.param("test_inventory_medium.yml", "test_catalog_medium.yml", id="medium-6_devices-228_tests"), + pytest.param("test_inventory.yml", "test_catalog.yml", id="small-3_devices-3_tests"), + ], + indirect=True, +) +def test_runner(caplog: pytest.LogCaptureFixture, test_inventory: AntaInventory, test_catalog: AntaCatalog, aio_benchmark: Callable[..., Any]) -> None: + """Test and benchmark ANTA runner. + + caplog is the pytest fixture to capture logs. + """ + logger.setup_logging(logger.Log.INFO) + caplog.set_level(logging.INFO) + manager = ResultManager() + aio_benchmark(main, manager, test_inventory, test_catalog) @pytest.mark.asyncio() @@ -41,7 +62,7 @@ async def test_runner_empty_tests(caplog: pytest.LogCaptureFixture, test_invento @pytest.mark.asyncio() -async def test_runner_empty_inventory(caplog: pytest.LogCaptureFixture) -> None: +async def test_runner_empty_inventory(caplog: pytest.LogCaptureFixture, test_catalog: AntaCatalog) -> None: """Test that when the Inventory is empty, a log is raised. caplog is the pytest fixture to capture logs @@ -50,14 +71,14 @@ async def test_runner_empty_inventory(caplog: pytest.LogCaptureFixture) -> None: caplog.set_level(logging.INFO) manager = ResultManager() inventory = AntaInventory() - await main(manager, inventory, FAKE_CATALOG) + await main(manager, inventory, test_catalog) assert len(caplog.record_tuples) == 3 - assert "The inventory is empty, exiting" in caplog.records[1].message + assert "The inventory is empty, exiting" in caplog.records[0].message @pytest.mark.asyncio() -async def test_runner_no_selected_device(caplog: pytest.LogCaptureFixture, test_inventory: AntaInventory) -> None: - """Test that when the list of established device. +async def test_runner_no_selected_device(caplog: pytest.LogCaptureFixture, test_inventory: AntaInventory, test_catalog: AntaCatalog) -> None: + """Test that when there is no reachable device, a log is raised. caplog is the pytest fixture to capture logs test_inventory is a fixture that gives a default inventory for tests @@ -65,13 +86,13 @@ async def test_runner_no_selected_device(caplog: pytest.LogCaptureFixture, test_ logger.setup_logging(logger.Log.INFO) caplog.set_level(logging.INFO) manager = ResultManager() - await main(manager, test_inventory, FAKE_CATALOG) + await main(manager, test_inventory, test_catalog) assert "No reachable device was found." in [record.message for record in caplog.records] # Reset logs and run with tags caplog.clear() - await main(manager, test_inventory, FAKE_CATALOG, tags={"toto"}) + await main(manager, test_inventory, test_catalog, tags={"toto"}) assert "No reachable device matching the tags {'toto'} was found." in [record.message for record in caplog.records]