ci: use pytest-benchmark

.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

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
 ]

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

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",

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,28 +71,28 @@ 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
     """
     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]