Repository: aristanetworks/anta (python)

Issue:
feat(anta.tests): VerifyLoggingErrors failure message should include the error logs
Missing details in the current failure message:

![Image](https://github.com/user-attachments/assets/2aea1062-69bd-407e-bd83-8c85e2f798cc)

Pull request:
feat(anta.tests): VerifyLoggingErrors failure message should include the error logs
# Description

VerifyLoggingErrors failure message should include the error logs

Fixes #1139 

# Checklist:

- [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`)
- [ ] 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`)

## Failure message:

<img width="1193" alt="Screenshot 2025-04-08 at 8 23 37 PM" src="https://github.com/user-attachments/assets/9fa41eba-6ab3-4349-b2a3-5a91f2b599ce" />



Patch:
diff --git a/.github/workflows/sonar.yml b/.github/workflows/sonar.yml
index d5d443e2..4562d25c 100644
--- a/.github/workflows/sonar.yml
+++ b/.github/workflows/sonar.yml
@@ -11,13 +11,6 @@ jobs:
     if: github.repository == 'aristanetworks/anta'
     runs-on: ubuntu-latest
     steps:
-      - name: Display workflow run
-        env:
-          WORKFLOW_RUN_JSON: ${{ toJson(github.event.workflow_run) }}
-        run: |
-            echo "--- Workflow Run JSON Start ---"
-            echo "$WORKFLOW_RUN_JSON"
-            echo "--- Workflow Run JSON End ---"
       - uses: actions/checkout@v4
         with:
           ref: ${{ github.event.workflow_run.head_sha }}
diff --git a/anta/cli/__init__.py b/anta/cli/__init__.py
index dd39f78b..689c4275 100644
--- a/anta/cli/__init__.py
+++ b/anta/cli/__init__.py
@@ -16,11 +16,14 @@ try:
 
 except ImportError as exc:
 
-    def build_cli(exception: Exception) -> Callable[[], None]:
+    def build_cli(exception: ImportError) -> Callable[[], None]:
         """Build CLI function using the caught exception."""
 
         def wrap() -> None:
             """Error message if any CLI dependency is missing."""
+            if not exception.name or "click" not in exception.name:
+                raise exception
+
             print(
                 "The ANTA command line client could not run because the required "
                 "dependencies were not installed.\nMake sure you've installed "
diff --git a/anta/custom_types.py b/anta/custom_types.py
index 5c87e29b..2245cb17 100644
--- a/anta/custom_types.py
+++ b/anta/custom_types.py
@@ -396,3 +396,5 @@ RedistributedProtocol = Annotated[
 ]
 RedistributedAfiSafi = Annotated[Literal["v4u", "v4m", "v6u", "v6m"], BeforeValidator(bgp_redistributed_route_proto_abbreviations)]
 NTPStratumLevel = Annotated[int, Field(ge=0, le=16)]
+PowerSupplyFanStatus = Literal["failed", "ok", "unknownHwStatus", "powerLoss", "unsupported"]
+PowerSupplyStatus = Literal["ok", "unknown", "powerLoss", "failed"]
diff --git a/anta/device.py b/anta/device.py
index 7c1e6f64..8ca4e3f3 100644
--- a/anta/device.py
+++ b/anta/device.py
@@ -539,29 +539,36 @@ class AsyncEOSDevice(AntaDevice):
         """Update attributes of an AsyncEOSDevice instance.
 
         This coroutine must update the following attributes of AsyncEOSDevice:
-        - is_online: When a device IP is reachable and a port can be open
+        - is_online: When a device eAPI HTTP endpoint is accessible
         - established: When a command execution succeeds
         - hw_model: The hardware model of the device
         """
         logger.debug("Refreshing device %s", self.name)
-        self.is_online = await self._session.check_connection()
-        if self.is_online:
-            show_version = AntaCommand(command="show version")
-            await self._collect(show_version)
-            if not show_version.collected:
-                logger.warning("Cannot get hardware information from device %s", self.name)
-            else:
-                self.hw_model = show_version.json_output.get("modelName", None)
-                if self.hw_model is None:
-                    logger.critical("Cannot parse 'show version' returned by device %s", self.name)
-                # in some cases it is possible that 'modelName' comes back empty
-                # and it is nice to get a meaninfule error message
-                elif self.hw_model == "":
-                    logger.critical("Got an empty 'modelName' in the 'show version' returned by device %s", self.name)
+        try:
+            self.is_online = await self._session.check_api_endpoint()
+        except HTTPError as e:
+            self.is_online = False
+            self.established = False
+            logger.warning("Could not connect to device %s: %s", self.name, e)
+            return
+
+        show_version = AntaCommand(command="show version")
+        await self._collect(show_version)
+        if not show_version.collected:
+            self.established = False
+            logger.warning("Cannot get hardware information from device %s", self.name)
+            return
+
+        self.hw_model = show_version.json_output.get("modelName", None)
+        if self.hw_model is None:
+            self.established = False
+            logger.critical("Cannot parse 'show version' returned by device %s", self.name)
+        # in some cases it is possible that 'modelName' comes back empty
+        elif self.hw_model == "":
+            self.established = False
+            logger.critical("Got an empty 'modelName' in the 'show version' returned by device %s", self.name)
         else:
-            logger.warning("Could not connect to device %s: cannot open eAPI port", self.name)
-
-        self.established = bool(self.is_online and self.hw_model)
+            self.established = True
 
     async def copy(self, sources: list[Path], destination: Path, direction: Literal["to", "from"] = "from") -> None:
         """Copy files to and from the device using asyncssh.scp().
diff --git a/asynceapi/device.py b/asynceapi/device.py
index a7702daa..504d072f 100644
--- a/asynceapi/device.py
+++ b/asynceapi/device.py
@@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any, Literal, overload
 # Public Imports
 # -----------------------------------------------------------------------------
 import httpx
+from typing_extensions import deprecated
 
 # -----------------------------------------------------------------------------
 # Private Imports
@@ -51,6 +52,7 @@ class Device(httpx.AsyncClient):
     """
 
     auth = None
+    EAPI_COMMAND_API_URL = "/command-api"
     EAPI_OFMT_OPTIONS = ("json", "text")
     EAPI_DEFAULT_OFMT = "json"
 
@@ -109,6 +111,7 @@ class Device(httpx.AsyncClient):
         super().__init__(**kwargs)
         self.headers["Content-Type"] = "application/json-rpc"
 
+    @deprecated("This method is deprecated, use `Device.check_api_endpoint` method instead. This will be removed in ANTA v2.0.0.", category=DeprecationWarning)
     async def check_connection(self) -> bool:
         """Check the target device to ensure that the eAPI port is open and accepting connections.
 
@@ -122,6 +125,22 @@ class Device(httpx.AsyncClient):
         """
         return await port_check_url(self.base_url)
 
+    async def check_api_endpoint(self) -> bool:
+        """Check the target device eAPI HTTP endpoint with a HEAD request.
+
+        It is recommended that a Caller checks the connection before involving cli commands,
+        but this step is not required.
+
+        Returns
+        -------
+        bool
+            True when the device eAPI HTTP endpoint is accessible (2xx status code),
+            otherwise an HTTPStatusError exception is raised.
+        """
+        response = await self.head(self.EAPI_COMMAND_API_URL, timeout=5)
+        response.raise_for_status()
+        return True
+
     # Single command, JSON output, no suppression
     @overload
     async def cli(
@@ -416,7 +435,7 @@ class Device(httpx.AsyncClient):
             The list of command results; either dict or text depending on the
             JSON-RPC format parameter.
         """
-        res = await self.post("/command-api", json=jsonrpc)
+        res = await self.post(self.EAPI_COMMAND_API_URL, json=jsonrpc)
         res.raise_for_status()
         body = res.json()
 
diff --git a/pyproject.toml b/pyproject.toml
index 68745437..b8d2524b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -56,7 +56,6 @@ requires-python = ">=3.9"
 [project.optional-dependencies]
 cli = [
   "click~=8.1.6",
-  "click-help-colors>=0.9",
 ]
 dev = [
   "bumpver>=2023.1129",


Test patch:
diff --git a/anta/tests/hardware.py b/anta/tests/hardware.py
index 9e0453f9..8bac413d 100644
--- a/anta/tests/hardware.py
+++ b/anta/tests/hardware.py
@@ -9,6 +9,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING, ClassVar
 
+from anta.custom_types import PowerSupplyFanStatus, PowerSupplyStatus
 from anta.decorators import skip_on_platforms
 from anta.models import AntaCommand, AntaTest
 
@@ -176,7 +177,7 @@ class VerifyEnvironmentCooling(AntaTest):
     class Input(AntaTest.Input):
         """Input model for the VerifyEnvironmentCooling test."""
 
-        states: list[str]
+        states: list[PowerSupplyFanStatus]
         """List of accepted states of fan status."""
 
     @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
@@ -225,7 +226,7 @@ class VerifyEnvironmentPower(AntaTest):
     class Input(AntaTest.Input):
         """Input model for the VerifyEnvironmentPower test."""
 
-        states: list[str]
+        states: list[PowerSupplyStatus]
         """List of accepted states list of power supplies status."""
 
     @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
diff --git a/anta/tests/logging.py b/anta/tests/logging.py
index f13860e2..f140ad4e 100644
--- a/anta/tests/logging.py
+++ b/anta/tests/logging.py
@@ -433,7 +433,7 @@ class VerifyLoggingErrors(AntaTest):
         if len(command_output) == 0:
             self.result.is_success()
         else:
-            self.result.is_failure("Device has reported syslog messages with a severity of ERRORS or higher")
+            self.result.is_failure(f"Device has reported syslog messages with a severity of ERRORS or higher:\n{command_output}")
 
 
 class VerifyLoggingEntries(AntaTest):
diff --git a/tests/conftest.py b/tests/conftest.py
index 362d9d0a..90862667 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -3,10 +3,8 @@
 # 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
@@ -42,7 +40,8 @@ def inventory(request: pytest.FixtureRequest) -> Iterator[AntaInventory]:
             )
     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:
+        with respx.mock:
+            respx.head(path="/command-api")
             respx.post(path="/command-api", headers={"Content-Type": "application/json-rpc"}, json__params__cmds__0__cmd="show version").respond(
                 json={
                     "result": [
@@ -54,5 +53,6 @@ def inventory(request: pytest.FixtureRequest) -> Iterator[AntaInventory]:
             )
             yield inv
     else:
-        with patch("asyncio.open_connection", AsyncMock(spec=asyncio.open_connection, side_effect=TimeoutError)):
+        with respx.mock:
+            respx.head(path="/command-api").respond(status_code=401)
             yield inv
diff --git a/tests/units/anta_tests/test_hardware.py b/tests/units/anta_tests/test_hardware.py
index 972b07f0..15e921e9 100644
--- a/tests/units/anta_tests/test_hardware.py
+++ b/tests/units/anta_tests/test_hardware.py
@@ -395,7 +395,7 @@ DATA: list[dict[str, Any]] = [
                         "status": "ok",
                         "fans": [
                             {
-                                "status": "Not Inserted",
+                                "status": "powerLoss",
                                 "uptime": 1682498935.9121106,
                                 "maxSpeed": 23000,
                                 "lastSpeedStableChangeTime": 1682499092.4665174,
@@ -490,7 +490,7 @@ DATA: list[dict[str, Any]] = [
                 "systemStatus": "coolingOk",
             },
         ],
-        "inputs": {"states": ["ok", "Not Inserted"]},
+        "inputs": {"states": ["ok", "powerLoss"]},
         "expected": {"result": "success"},
     },
     {
@@ -548,7 +548,7 @@ DATA: list[dict[str, Any]] = [
                         "status": "ok",
                         "fans": [
                             {
-                                "status": "down",
+                                "status": "unknownHwStatus",
                                 "uptime": 1682498923.9303148,
                                 "maxSpeed": 17500,
                                 "lastSpeedStableChangeTime": 1682498975.0139885,
@@ -584,7 +584,7 @@ DATA: list[dict[str, Any]] = [
                         "status": "ok",
                         "fans": [
                             {
-                                "status": "Not Inserted",
+                                "status": "powerLoss",
                                 "uptime": 1682498923.9383528,
                                 "maxSpeed": 17500,
                                 "lastSpeedStableChangeTime": 1682498975.0140095,
@@ -623,8 +623,8 @@ DATA: list[dict[str, Any]] = [
                 "systemStatus": "CoolingKo",
             },
         ],
-        "inputs": {"states": ["ok", "Not Inserted"]},
-        "expected": {"result": "failure", "messages": ["Fan Tray: 1 Fan: 1/1 - Invalid state - Expected: ok, Not Inserted Actual: down"]},
+        "inputs": {"states": ["ok", "powerLoss"]},
+        "expected": {"result": "failure", "messages": ["Fan Tray: 1 Fan: 1/1 - Invalid state - Expected: ok, powerLoss Actual: unknownHwStatus"]},
     },
     {
         "name": "failure-power-supply",
@@ -643,7 +643,7 @@ DATA: list[dict[str, Any]] = [
                         "status": "ok",
                         "fans": [
                             {
-                                "status": "down",
+                                "status": "unknownHwStatus",
                                 "uptime": 1682498937.0240965,
                                 "maxSpeed": 23000,
                                 "lastSpeedStableChangeTime": 1682499033.0403435,
@@ -717,7 +717,7 @@ DATA: list[dict[str, Any]] = [
                         "status": "ok",
                         "fans": [
                             {
-                                "status": "Not Inserted",
+                                "status": "powerLoss",
                                 "uptime": 1682498923.9383528,
                                 "maxSpeed": 17500,
                                 "lastSpeedStableChangeTime": 1682498975.0140095,
@@ -756,11 +756,11 @@ DATA: list[dict[str, Any]] = [
                 "systemStatus": "CoolingKo",
             },
         ],
-        "inputs": {"states": ["ok", "Not Inserted"]},
+        "inputs": {"states": ["ok", "powerLoss"]},
         "expected": {
             "result": "failure",
             "messages": [
-                "Power Slot: PowerSupply1 Fan: PowerSupply1/1 - Invalid state - Expected: ok, Not Inserted Actual: down",
+                "Power Slot: PowerSupply1 Fan: PowerSupply1/1 - Invalid state - Expected: ok, powerLoss Actual: unknownHwStatus",
             ],
         },
     },
@@ -827,7 +827,7 @@ DATA: list[dict[str, Any]] = [
                             "TempSensorP1/1": {"status": "ok", "temperature": 0.0},
                         },
                         "fans": {"FanP1/1": {"status": "ok", "speed": 33}},
-                        "state": "Not Inserted",
+                        "state": "powerLoss",
                         "inputCurrent": 0.0,
                         "dominant": False,
                         "inputVoltage": 0.0,
@@ -855,7 +855,7 @@ DATA: list[dict[str, Any]] = [
                 },
             },
         ],
-        "inputs": {"states": ["ok", "Not Inserted"]},
+        "inputs": {"states": ["ok", "powerLoss"]},
         "expected": {"result": "success"},
     },
     {
diff --git a/tests/units/anta_tests/test_logging.py b/tests/units/anta_tests/test_logging.py
index 5f795895..2628e7d4 100644
--- a/tests/units/anta_tests/test_logging.py
+++ b/tests/units/anta_tests/test_logging.py
@@ -277,7 +277,14 @@ DATA: list[dict[str, Any]] = [
             "Aug  2 19:57:42 DC1-LEAF1A Mlag: %FWK-3-SOCKET_CLOSE_REMOTE: Connection to Mlag (pid:27200) at tbt://192.168.0.1:4432/+n closed by peer (EOF)",
         ],
         "inputs": None,
-        "expected": {"result": "failure", "messages": ["Device has reported syslog messages with a severity of ERRORS or higher"]},
+        "expected": {
+            "result": "failure",
+            "messages": [
+                "Device has reported syslog messages with a severity of ERRORS or higher:\n"
+                "Aug  2 19:57:42 DC1-LEAF1A Mlag: %FWK-3-SOCKET_CLOSE_REMOTE: Connection to Mlag "
+                "(pid:27200) at tbt://192.168.0.1:4432/+n closed by peer (EOF)"
+            ],
+        },
     },
     {
         "name": "success",
diff --git a/tests/units/cli/conftest.py b/tests/units/cli/conftest.py
index 37646957..d8b308ad 100644
--- a/tests/units/cli/conftest.py
+++ b/tests/units/cli/conftest.py
@@ -123,7 +123,7 @@ def click_runner(capsys: pytest.CaptureFixture[str], anta_env: dict[str, str]) -
 
     # Patch asynceapi methods used by AsyncEOSDevice. See tests/units/test_device.py
     with (
-        patch("asynceapi.device.Device.check_connection", return_value=True),
+        patch("asynceapi.device.Device.check_api_endpoint", return_value=True),
         patch("asynceapi.device.Device.cli", side_effect=cli),
         patch("asyncssh.connect"),
         patch(
diff --git a/tests/units/cli/test__init__.py b/tests/units/cli/test__init__.py
index 308516f8..2c247311 100644
--- a/tests/units/cli/test__init__.py
+++ b/tests/units/cli/test__init__.py
@@ -6,35 +6,21 @@
 from __future__ import annotations
 
 import sys
-from importlib import reload
-from typing import TYPE_CHECKING, Any
+from typing import Any
 from unittest.mock import patch
 
 import pytest
 
-import anta.cli
 
-if TYPE_CHECKING:
-    from types import ModuleType
-
-builtins_import = __import__
-
-
-# Tried to achieve this with mock
-# http://materials-scientist.com/blog/2021/02/11/mocking-failing-module-import-python/
-def import_mock(name: str, *args: Any) -> ModuleType:  # noqa: ANN401
-    """Mock."""
-    if name == "click":
-        msg = "No module named 'click'"
-        raise ModuleNotFoundError(msg)
-    return builtins_import(name, *args)
-
-
-def test_cli_error_missing(capsys: pytest.CaptureFixture[Any]) -> None:
+# https://github.com/python/cpython/issues/88852
+@pytest.mark.skipif(sys.version_info <= (3, 11), reason="Unreliable behavior patching sys.modules before 3.11")
+def test_cli_error_missing_click(capsys: pytest.CaptureFixture[Any]) -> None:
     """Test ANTA errors out when anta[cli] was not installed."""
-    with patch.dict(sys.modules) as sys_modules, patch("builtins.__import__", import_mock):
-        del sys_modules["anta.cli._main"]
-        reload(anta.cli)
+    with patch.dict(sys.modules, {"click": None}) as sys_modules:
+        for k in list(sys_modules.keys()):
+            if k.startswith("anta."):
+                del sys_modules[k]
+        import anta.cli
 
         with pytest.raises(SystemExit) as e_info:
             anta.cli.cli()
@@ -53,3 +39,18 @@ def test_cli_error_missing(capsys: pytest.CaptureFixture[Any]) -> None:
         assert "Make sure you've installed everything with: pip install 'anta[cli]'" in captured.out
         assert "The caught exception was:" in captured.out
         assert e_info.value.code == 1
+
+
+# https://github.com/python/cpython/issues/88852
+@pytest.mark.skipif(sys.version_info <= (3, 11), reason="Unreliable behavior patching sys.modules before 3.11")
+def test_cli_error_missing_other() -> None:
+    """Test ANTA errors out when anta[cli] was not installed."""
+    with patch.dict(sys.modules, {"httpx": None}) as sys_modules:
+        # Need to clean up from previous runs a path that will trigger reimporting httpx
+        for k in list(sys_modules.keys()):
+            if k.startswith("anta."):
+                del sys_modules[k]
+        import anta.cli
+
+        with pytest.raises(ImportError, match="httpx"):
+            anta.cli.cli()
diff --git a/tests/units/test_device.py b/tests/units/test_device.py
index e65eeb2a..0ecba555 100644
--- a/tests/units/test_device.py
+++ b/tests/units/test_device.py
@@ -429,29 +429,8 @@ REFRESH_PARAMS: list[ParameterSet] = [
     pytest.param(
         {},
         (
-            {"return_value": False},
-            {
-                "return_value": {
-                    "mfgName": "Arista",
-                    "modelName": "DCS-7280CR3-32P4-F",
-                    "hardwareRevision": "11.00",
-                    "serialNumber": "JPE19500066",
-                    "systemMacAddress": "fc:bd:67:3d:13:c5",
-                    "hwMacAddress": "fc:bd:67:3d:13:c5",
-                    "configMacAddress": "00:00:00:00:00:00",
-                    "version": "4.31.1F-34361447.fraserrel (engineering build)",
-                    "architecture": "x86_64",
-                    "internalVersion": "4.31.1F-34361447.fraserrel",
-                    "internalBuildId": "4940d112-a2fc-4970-8b5a-a16cd03fd08c",
-                    "imageFormatVersion": "3.0",
-                    "imageOptimization": "Default",
-                    "bootupTimestamp": 1700729434.5892005,
-                    "uptime": 20666.78,
-                    "memTotal": 8099732,
-                    "memFree": 4989568,
-                    "isIntlVersion": False,
-                }
-            },
+            {"side_effect": HTTPError(message="Unauthorized")},
+            {},
         ),
         {"is_online": False, "established": False, "hw_model": None},
         id="is not online",
@@ -653,9 +632,9 @@ class TestAsyncEOSDevice:
     )
     async def test_refresh(self, async_device: AsyncEOSDevice, patch_kwargs: list[dict[str, Any]], expected: dict[str, Any]) -> None:
         """Test AsyncEOSDevice.refresh()."""
-        with patch.object(async_device._session, "check_connection", **patch_kwargs[0]), patch.object(async_device._session, "cli", **patch_kwargs[1]):
+        with patch.object(async_device._session, "check_api_endpoint", **patch_kwargs[0]), patch.object(async_device._session, "cli", **patch_kwargs[1]):
             await async_device.refresh()
-            async_device._session.check_connection.assert_called_once()  # type: ignore[attr-defined] # asynceapi.Device.check_connection is patched
+            async_device._session.check_api_endpoint.assert_called_once()  # type: ignore[attr-defined] # asynceapi.Device.check_api_endpoint is patched
             if expected["is_online"]:
                 async_device._session.cli.assert_called_once()  # type: ignore[attr-defined] # asynceapi.Device.cli is patched
             assert async_device.is_online == expected["is_online"]