fix(anta): Added support for dict commands in EapiCommandError

asynceapi/device.py

@@ -271,10 +271,11 @@ async def jsonrpc_exec(self, jsonrpc: dict[str, Any]) -> list[dict[str, Any] | s
         len_data = len(cmd_data)
         err_at = len_data - 1
         err_msg = err_data["message"]
+        failed_cmd = commands[err_at]
 
         raise EapiCommandError(
             passed=[get_output(cmd_data[cmd_i]) for cmd_i, cmd in enumerate(commands[:err_at])],
-            failed=commands[err_at]["cmd"],
+            failed=failed_cmd["cmd"] if isinstance(failed_cmd, dict) else failed_cmd,
             errors=cmd_data[err_at]["errors"],
             errmsg=err_msg,
             not_exec=commands[err_at + 1 :],

pyproject.toml

@@ -69,6 +69,7 @@ dev = [
   "pytest-cov>=4.1.0",
   "pytest-dependency",
   "pytest-html>=3.2.0",
+  "pytest-httpx>=0.30.0",
   "pytest-metadata>=3.0.0",
   "pytest>=7.4.0",
   "ruff>=0.5.4,<0.7.0",
@@ -181,7 +182,8 @@ filterwarnings = [
 
 [tool.coverage.run]
 branch = true
-source = ["anta"]
+# https://community.sonarsource.com/t/python-coverage-analysis-warning/62629/7
+include = ["anta/*", "asynceapi/*"]
 parallel = true
 relative_files = true
 

tests/units/asynceapi/__init__.py

@@ -0,0 +1,4 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Unit tests for the asynceapi client package used by ANTA."""

tests/units/asynceapi/conftest.py

@@ -0,0 +1,20 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Fixtures for the asynceapi client package."""
+
+import pytest
+
+from asynceapi import Device
+
+
+@pytest.fixture
+def asynceapi_device() -> Device:
+    """Return an asynceapi Device instance."""
+    return Device(
+        host="localhost",
+        username="admin",
+        password="admin",
+        proto="https",
+        port=443,
+    )

tests/units/asynceapi/test_data.py

@@ -0,0 +1,88 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Unit tests data for the asynceapi client package."""
+
+SUCCESS_EAPI_RESPONSE = {
+    "jsonrpc": "2.0",
+    "id": "EapiExplorer-1",
+    "result": [
+        {
+            "mfgName": "Arista",
+            "modelName": "cEOSLab",
+            "hardwareRevision": "",
+            "serialNumber": "5E9D49D20F09DA471333DD835835FD1A",
+            "systemMacAddress": "00:1c:73:2e:7b:a3",
+            "hwMacAddress": "00:00:00:00:00:00",
+            "configMacAddress": "00:00:00:00:00:00",
+            "version": "4.31.1F-34554157.4311F (engineering build)",
+            "architecture": "i686",
+            "internalVersion": "4.31.1F-34554157.4311F",
+            "internalBuildId": "47114ca4-ae9f-4f32-8c1f-2864db93b7e8",
+            "imageFormatVersion": "1.0",
+            "imageOptimization": "None",
+            "cEosToolsVersion": "(unknown)",
+            "kernelVersion": "6.5.0-44-generic",
+            "bootupTimestamp": 1723429239.9352903,
+            "uptime": 1300202.749528885,
+            "memTotal": 65832112,
+            "memFree": 41610316,
+            "isIntlVersion": False,
+        },
+        {
+            "utcTime": 1724729442.6863558,
+            "timezone": "EST",
+            "localTime": {
+                "year": 2024,
+                "month": 8,
+                "dayOfMonth": 26,
+                "hour": 22,
+                "min": 30,
+                "sec": 42,
+                "dayOfWeek": 0,
+                "dayOfYear": 239,
+                "daylightSavingsAdjust": 0,
+            },
+            "clockSource": {"local": True},
+        },
+    ],
+}
+"""Successful eAPI JSON response."""
+
+ERROR_EAPI_RESPONSE = {
+    "jsonrpc": "2.0",
+    "id": "EapiExplorer-1",
+    "error": {
+        "code": 1002,
+        "message": "CLI command 2 of 3 'bad command' failed: invalid command",
+        "data": [
+            {
+                "mfgName": "Arista",
+                "modelName": "cEOSLab",
+                "hardwareRevision": "",
+                "serialNumber": "5E9D49D20F09DA471333DD835835FD1A",
+                "systemMacAddress": "00:1c:73:2e:7b:a3",
+                "hwMacAddress": "00:00:00:00:00:00",
+                "configMacAddress": "00:00:00:00:00:00",
+                "version": "4.31.1F-34554157.4311F (engineering build)",
+                "architecture": "i686",
+                "internalVersion": "4.31.1F-34554157.4311F",
+                "internalBuildId": "47114ca4-ae9f-4f32-8c1f-2864db93b7e8",
+                "imageFormatVersion": "1.0",
+                "imageOptimization": "None",
+                "cEosToolsVersion": "(unknown)",
+                "kernelVersion": "6.5.0-44-generic",
+                "bootupTimestamp": 1723429239.9352903,
+                "uptime": 1300027.2297976017,
+                "memTotal": 65832112,
+                "memFree": 41595080,
+                "isIntlVersion": False,
+            },
+            {"errors": ["Invalid input (at token 1: 'bad')"]},
+        ],
+    },
+}
+"""Error eAPI JSON response."""
+
+JSONRPC_REQUEST_TEMPLATE = {"jsonrpc": "2.0", "method": "runCmds", "params": {"version": 1, "cmds": [], "format": "json"}, "id": "EapiExplorer-1"}
+"""Template for JSON-RPC eAPI request. `cmds` must be filled by the parametrize decorator."""

tests/units/asynceapi/test_device.py

@@ -0,0 +1,88 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Unit tests the asynceapi.device module."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from httpx import HTTPStatusError
+
+from asynceapi import Device, EapiCommandError
+
+from .test_data import ERROR_EAPI_RESPONSE, JSONRPC_REQUEST_TEMPLATE, SUCCESS_EAPI_RESPONSE
+
+if TYPE_CHECKING:
+    from pytest_httpx import HTTPXMock
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "cmds",
+    [
+        (["show version", "show clock"]),
+        ([{"cmd": "show version"}, {"cmd": "show clock"}]),
+        ([{"cmd": "show version"}, "show clock"]),
+    ],
+    ids=["simple_commands", "complex_commands", "mixed_commands"],
+)
+async def test_jsonrpc_exec_success(
+    asynceapi_device: Device,
+    httpx_mock: HTTPXMock,
+    cmds: list[str | dict[str, Any]],
+) -> None:
+    """Test the Device.jsonrpc_exec method with a successful response. Simple and complex commands are tested."""
+    jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy()
+    jsonrpc_request["params"]["cmds"] = cmds
+
+    httpx_mock.add_response(json=SUCCESS_EAPI_RESPONSE)
+
+    result = await asynceapi_device.jsonrpc_exec(jsonrpc=jsonrpc_request)
+
+    assert result == SUCCESS_EAPI_RESPONSE["result"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "cmds",
+    [
+        (["show version", "bad command", "show clock"]),
+        ([{"cmd": "show version"}, {"cmd": "bad command"}, {"cmd": "show clock"}]),
+        ([{"cmd": "show version"}, {"cmd": "bad command"}, "show clock"]),
+    ],
+    ids=["simple_commands", "complex_commands", "mixed_commands"],
+)
+async def test_jsonrpc_exec_eapi_command_error(
+    asynceapi_device: Device,
+    httpx_mock: HTTPXMock,
+    cmds: list[str | dict[str, Any]],
+) -> None:
+    """Test the Device.jsonrpc_exec method with an error response. Simple and complex commands are tested."""
+    jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy()
+    jsonrpc_request["params"]["cmds"] = cmds
+
+    error_eapi_response: dict[str, Any] = ERROR_EAPI_RESPONSE.copy()
+    httpx_mock.add_response(json=error_eapi_response)
+
+    with pytest.raises(EapiCommandError) as exc_info:
+        await asynceapi_device.jsonrpc_exec(jsonrpc=jsonrpc_request)
+
+    assert exc_info.value.passed == [error_eapi_response["error"]["data"][0]]
+    assert exc_info.value.failed == "bad command"
+    assert exc_info.value.errors == ["Invalid input (at token 1: 'bad')"]
+    assert exc_info.value.errmsg == "CLI command 2 of 3 'bad command' failed: invalid command"
+    assert exc_info.value.not_exec == [jsonrpc_request["params"]["cmds"][2]]
+
+
+@pytest.mark.asyncio
+async def test_jsonrpc_exec_http_status_error(asynceapi_device: Device, httpx_mock: HTTPXMock) -> None:
+    """Test the Device.jsonrpc_exec method with an HTTPStatusError."""
+    jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy()
+    jsonrpc_request["params"]["cmds"] = ["show version"]
+
+    httpx_mock.add_response(status_code=500, text="Internal Server Error")
+
+    with pytest.raises(HTTPStatusError):
+        await asynceapi_device.jsonrpc_exec(jsonrpc=jsonrpc_request)