Repository: aristanetworks/anta (python)

Issue:
bug(anta): anta._runner module should not depend on anta.cli when used outside of ANTA CLI
```
  File "/home/ansible/.ansible/collections/ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/get_anta_results.py", line 47, in <module>
    from anta.runner import main as anta_runner
  File "/home/ansible/.local/lib/python3.10/site-packages/anta/runner.py", line 16, in <module>
    from anta._runner import AntaRunFilters, AntaRunner
  File "/home/ansible/.local/lib/python3.10/site-packages/anta/_runner.py", line 19, in <module>
    from anta.cli.console import console
  File "/home/ansible/.local/lib/python3.10/site-packages/anta/cli/__init__.py", line 15, in <module>
    from ._main import anta, cli
  File "/home/ansible/.local/lib/python3.10/site-packages/anta/cli/_main.py", line 15, in <module>
    from anta.cli.check import check as check_command
  File "/home/ansible/.local/lib/python3.10/site-packages/anta/cli/check/__init__.py", line 16, in <module>
    check.add_command(commands.catalog)
AttributeError: 'function' object has no attribute 'add_command'
```

Workaround is to make sure `anta[cli]` is also installed: https://anta.arista.com/stable/requirements-and-installation/#install-cli-from-pypi-server

Pull request:
fix(anta): Remove anta.cli import from anta._runner
# Description

Remove the use of `anta.cli.console` in the runner to avoid being dependant of `anta.cli` when using ANTA as a library.

Fixes #1226 

# Checklist:

<!-- Delete not relevant items !-->

- [X] My code follows the style guidelines of this project
- [X] I have performed a self-review of my code
- [X] I have run pre-commit for code linting and typing (`pre-commit run`)
- [X] I have made corresponding changes to the documentation
- [X] I have added tests that prove my fix is effective or that my feature works
- [X] New and existing unit tests pass locally with my changes (`tox -e testenv`)


Patch:
diff --git a/anta/_runner.py b/anta/_runner.py
index 109108be..eaa1cf9a 100644
--- a/anta/_runner.py
+++ b/anta/_runner.py
@@ -16,7 +16,6 @@ from typing import TYPE_CHECKING, Any
 from pydantic import BaseModel, ConfigDict
 
 from anta import GITHUB_SUGGESTION
-from anta.cli.console import console
 from anta.inventory import AntaInventory
 from anta.logger import anta_log_exception
 from anta.models import AntaTest
@@ -287,7 +286,7 @@ class AntaRunner:
             return ctx
 
         if AntaTest.progress is not None:
-            AntaTest.nrfu_task = AntaTest.progress.add_task("Running NRFU Tests...", total=ctx.total_tests_scheduled)
+            AntaTest.nrfu_task = AntaTest.progress.add_task("Running NRFU Tests ...", total=ctx.total_tests_scheduled)
 
         with Catchtime(logger=logger, message="Running Tests"):
             sem = Semaphore(self._settings.max_concurrency)
@@ -415,48 +414,35 @@ class AntaRunner:
         for coro in coros:
             # Get the AntaTest instance from the coroutine locals, can be in `args` when decorated
             coro_locals = getcoroutinelocals(coro)
-            test = coro_locals.get("self") or coro_locals.get("args", (None))[0]
+            test = coro_locals.get("self") or coro_locals.get("args")
             if isinstance(test, AntaTest):
                 ctx.manager.add(test.result)
+            elif test and isinstance(test, tuple) and isinstance(test[0], AntaTest):
+                ctx.manager.add(test[0].result)
             else:
                 logger.error("Coroutine %s does not have an AntaTest instance.", coro)
             coro.close()
 
     def _log_run_information(self, ctx: AntaRunContext) -> None:
         """Log ANTA run information and potential resource limit warnings."""
-        # 34 is an estimate of the combined length of timestamp, log level name, filename and spacing added by the Rich logger
-        width = min(int(console.width) - 34, len("  Potential connections needed: 100000000\n"))
-
-        # Build device information
-        device_lines = [
-            "Devices:",
-            f"  Total in initial inventory: {ctx.total_devices_in_inventory}",
-        ]
+        logger.info("Initial inventory contains %s devices", ctx.total_devices_in_inventory)
+
         if ctx.total_devices_filtered_by_tags > 0:
-            device_lines.append(f"  Excluded by tags: {ctx.total_devices_filtered_by_tags}")
+            device_list_str = ", ".join(sorted(ctx.devices_filtered_at_setup))
+            logger.info("%d devices excluded by name/tag filters: %s", ctx.total_devices_filtered_by_tags, device_list_str)
+
         if ctx.total_devices_unreachable > 0:
-            device_lines.append(f"  Failed to connect: {ctx.total_devices_unreachable}")
-        device_lines.append(f"  Selected for testing: {ctx.total_devices_selected_for_testing}")
-        joined_device_lines = "\n".join(device_lines)
-
-        # Build title
-        title = " ANTA NRFU Dry Run Information " if ctx.dry_run else " ANTA NRFU Run Information "
-        formatted_title_line = f"{title:-^{width}}"
-
-        # Log run information
-        run_info = "\n".join(
-            [
-                f"{formatted_title_line}",
-                f"{joined_device_lines}",
-                f"Total number of selected tests: {ctx.total_tests_scheduled}",
-                f"{'':-^{width}}",
-            ]
-        )
-        logger.info(run_info)
-        logger.debug("Max concurrent tests: %d", self._settings.max_concurrency)
-        if potential_connections := ctx.selected_inventory.max_potential_connections:
-            logger.debug("Potential connections needed: %d", potential_connections)
-        logger.debug("File descriptors limit: %d", self._settings.file_descriptor_limit)
+            device_list_str = ", ".join(sorted(ctx.devices_unreachable_at_setup))
+            logger.info("%d devices found unreachable after connection attempts: %s", ctx.total_devices_unreachable, device_list_str)
+
+        logger.info("%d devices selected for testing", ctx.total_devices_selected_for_testing)
+        logger.info("%d total tests scheduled across all selected devices", ctx.total_tests_scheduled)
+
+        # Log debugs for runner settings
+        logger.debug("Max concurrent tests configured: %d", self._settings.max_concurrency)
+        if (potential_connections := ctx.selected_inventory.max_potential_connections) is not None:
+            logger.debug("Potential device connections estimated for this run: %d", potential_connections)
+        logger.debug("System file descriptor limit configured: %d", self._settings.file_descriptor_limit)
 
         # Log warnings for potential resource limits
         if ctx.total_tests_scheduled > self._settings.max_concurrency:
diff --git a/asynceapi/device.py b/asynceapi/device.py
index 504d072f..32529e5e 100644
--- a/asynceapi/device.py
+++ b/asynceapi/device.py
@@ -51,7 +51,6 @@ class Device(httpx.AsyncClient):
     httpx.AsyncClient, so any initialization options can be passed directly.
     """
 
-    auth = None
     EAPI_COMMAND_API_URL = "/command-api"
     EAPI_OFMT_OPTIONS = ("json", "text")
     EAPI_DEFAULT_OFMT = "json"
@@ -103,10 +102,8 @@ class Device(httpx.AsyncClient):
         kwargs.setdefault("base_url", httpx.URL(f"{proto}://{self.host}:{self.port}"))
         kwargs.setdefault("verify", False)
 
-        if username and password:
-            self.auth = httpx.BasicAuth(username, password)
-
-        kwargs.setdefault("auth", self.auth)
+        auth_object = httpx.BasicAuth(username, password) if username and password else None
+        kwargs.setdefault("auth", auth_object)
 
         super().__init__(**kwargs)
         self.headers["Content-Type"] = "application/json-rpc"


Test patch:
diff --git a/tests/units/test__runner.py b/tests/units/test__runner.py
index 46c63ed0..efb5f454 100644
--- a/tests/units/test__runner.py
+++ b/tests/units/test__runner.py
@@ -310,11 +310,28 @@ class TestAntaRunner:
         await runner.run(inventory, catalog, dry_run=True)
 
         expected_output = [
-            "ANTA NRFU Dry Run Information",
-            "Devices:",
-            "  Total in initial inventory: 3",
-            "  Selected for testing: 3",
-            "Total number of selected tests: 27",
+            "Initial inventory contains 3 devices",
+            "3 devices selected for testing",
+            "27 total tests scheduled across all selected devices",
+        ]
+        for line in expected_output:
+            assert line in caplog.text
+
+    async def test_log_run_information_filters(self, caplog: pytest.LogCaptureFixture) -> None:
+        """Test AntaRunner._log_run_information with filters."""
+        caplog.set_level(logging.INFO)
+
+        inventory = AntaInventory.parse(filename=DATA_DIR / "test_inventory_with_tags.yml", username="anta", password="anta")
+        catalog = AntaCatalog.parse(filename=DATA_DIR / "test_catalog_with_tags.yml")
+        runner = AntaRunner()
+        filters = AntaRunFilters(devices={"spine1"})
+        await runner.run(inventory, catalog, filters=filters, dry_run=True)
+
+        expected_output = [
+            "Initial inventory contains 3 devices",
+            "2 devices excluded by name/tag filters: leaf1, leaf2",
+            "1 devices selected for testing",
+            "9 total tests scheduled across all selected devices",
         ]
         for line in expected_output:
             assert line in caplog.text
@@ -350,6 +367,36 @@ class TestAntaRunner:
         assert warning_msg in ctx.warnings_at_setup
         assert warning_msg in caplog.messages
 
+    async def test_log_run_information_from_context(self, caplog: pytest.LogCaptureFixture) -> None:
+        """Test AntaRunner._log_run_information from a fake context."""
+        caplog.set_level(logging.INFO)
+
+        inventory = AntaInventory.parse(filename=DATA_DIR / "test_inventory_with_tags.yml", username="anta", password="anta")
+        catalog = AntaCatalog.parse(filename=DATA_DIR / "test_catalog_with_tags.yml")
+        manager = ResultManager()
+        filters = AntaRunFilters()
+        context = AntaRunContext(
+            inventory=inventory,
+            catalog=catalog,
+            manager=manager,
+            filters=filters,
+        )
+        context.devices_filtered_at_setup = ["leaf1"]
+        context.devices_unreachable_at_setup = ["leaf2"]
+        context.selected_inventory.add_device(inventory["spine1"])
+
+        AntaRunner()._log_run_information(context)
+
+        expected_output = [
+            "Initial inventory contains 3 devices",
+            "1 devices excluded by name/tag filters: leaf1",
+            "1 devices found unreachable after connection attempts: leaf2",
+            "1 devices selected for testing",
+            "0 total tests scheduled across all selected devices",
+        ]
+        for line in expected_output:
+            assert line in caplog.text
+
     @pytest.mark.parametrize(("inventory"), [{"count": 3}], indirect=True)
     @respx.mock
     async def test_run(self, inventory: AntaInventory) -> None: