diff --git a/anta/cli/exec/utils.py b/anta/cli/exec/utils.py index 1f874b0cd..ecef27857 100644 --- a/anta/cli/exec/utils.py +++ b/anta/cli/exec/utils.py @@ -23,6 +23,7 @@ if TYPE_CHECKING: from anta.inventory import AntaInventory + from asynceapi._types import EapiComplexCommand, EapiSimpleCommand EOS_SCHEDULED_TECH_SUPPORT = "/mnt/flash/schedule/tech-support" INVALID_CHAR = "`~!@#$/" @@ -135,7 +136,7 @@ async def collect(device: AntaDevice) -> None: ) logger.warning(msg) - commands = [] + commands: list[EapiSimpleCommand | EapiComplexCommand] = [] # TODO: @mtache - add `config` field to `AntaCommand` object to handle this use case. # Otherwise mypy complains about enable as it is only implemented for AsyncEOSDevice # TODO: Should enable be also included in AntaDevice? diff --git a/anta/device.py b/anta/device.py index 3624fdb2e..b636ebf13 100644 --- a/anta/device.py +++ b/anta/device.py @@ -26,6 +26,8 @@ from collections.abc import Iterator from pathlib import Path + from asynceapi._types import EapiComplexCommand, EapiSimpleCommand + logger = logging.getLogger(__name__) # Do not load the default keypairs multiple times due to a performance issue introduced in cryptography 37.0 @@ -445,7 +447,7 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No collection_id An identifier used to build the eAPI request ID. """ - commands: list[dict[str, str | int]] = [] + commands: list[EapiComplexCommand | EapiSimpleCommand] = [] if self.enable and self._enable_password is not None: commands.append( { @@ -458,12 +460,12 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No commands.append({"cmd": "enable"}) commands += [{"cmd": command.command, "revision": command.revision}] if command.revision else [{"cmd": command.command}] try: - response: list[dict[str, Any] | str] = await self._session.cli( + response = await self._session.cli( commands=commands, ofmt=command.ofmt, version=command.version, req_id=f"ANTA-{collection_id}-{id(command)}" if collection_id else f"ANTA-{id(command)}", - ) # type: ignore[assignment] # multiple commands returns a list + ) # Do not keep response of 'enable' command command.output = response[-1] except asynceapi.EapiCommandError as e: diff --git a/asynceapi/_types.py b/asynceapi/_types.py new file mode 100644 index 000000000..e81377c1c --- /dev/null +++ b/asynceapi/_types.py @@ -0,0 +1,43 @@ +# Copyright (c) 2024-2025 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Type definitions used for the asynceapi package.""" + +from __future__ import annotations + +from typing import Any, Literal, NotRequired, TypedDict + +EapiJsonOutput = dict[str, Any] +"""Type definition of an eAPI JSON output response.""" +EapiTextOutput = str +"""Type definition of an eAPI text output response.""" +EapiSimpleCommand = str +"""Type definition of an eAPI simple command.""" + + +class EapiComplexCommand(TypedDict): + """Type definition of an eAPI complex command.""" + + cmd: str + input: NotRequired[str] + revision: NotRequired[int] + + +class JsonRpc(TypedDict): + """Type definition of a JSON-RPC payload.""" + + jsonrpc: Literal["2.0"] + method: Literal["runCmds"] + params: JsonRpcParams + id: NotRequired[int | str] + + +class JsonRpcParams(TypedDict): + """Type definition of JSON-RPC parameters.""" + + version: NotRequired[int | Literal["latest"]] + cmds: list[EapiSimpleCommand | EapiComplexCommand] + format: NotRequired[Literal["json", "text"]] + autoComplete: NotRequired[bool] + expandAliases: NotRequired[bool] + timestamps: NotRequired[bool] diff --git a/asynceapi/config_session.py b/asynceapi/config_session.py index 470a0d69c..e5e1d0851 100644 --- a/asynceapi/config_session.py +++ b/asynceapi/config_session.py @@ -10,9 +10,10 @@ from __future__ import annotations import re -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING if TYPE_CHECKING: + from ._types import EapiComplexCommand, EapiJsonOutput, EapiSimpleCommand from .device import Device # ----------------------------------------------------------------------------- @@ -78,7 +79,7 @@ def device(self) -> Device: # Public Methods # ------------------------------------------------------------------------- - async def status_all(self) -> dict[str, Any]: + async def status_all(self) -> EapiJsonOutput: """Get the status of all the session config on the device. Run the following command on the device: @@ -86,7 +87,7 @@ async def status_all(self) -> dict[str, Any]: Returns ------- - dict[str, Any] + EapiJsonOutput Dictionary of native EOS eAPI response; see `status` method for details. @@ -116,9 +117,9 @@ async def status_all(self) -> dict[str, Any]: } ``` """ - return await self._cli("show configuration sessions detail") # type: ignore[return-value] # json outformat returns dict[str, Any] + return await self._cli(command="show configuration sessions detail") - async def status(self) -> dict[str, Any] | None: + async def status(self) -> EapiJsonOutput | None: """Get the status of a session config on the device. Run the following command on the device: @@ -129,7 +130,7 @@ async def status(self) -> dict[str, Any] | None: Returns ------- - dict[str, Any] | None + EapiJsonOutput | None Dictionary instance of the session status. If the session does not exist, then this method will return None. @@ -201,7 +202,7 @@ async def push(self, content: list[str] | str, *, replace: bool = False) -> None # prepare the initial set of command to enter the config session and # rollback clean if the `replace` argument is True. - commands: list[str | dict[str, Any]] = [self._cli_config_session] + commands: list[EapiSimpleCommand | EapiComplexCommand] = [self._cli_config_session] if replace: commands.append(self.CLI_CFG_FACTORY_RESET) @@ -232,7 +233,7 @@ async def commit(self, timer: str | None = None) -> None: if timer: command += f" timer {timer}" - await self._cli(command) + await self._cli(command=command) async def abort(self) -> None: """Abort the configuration session. @@ -240,7 +241,7 @@ async def abort(self) -> None: Run the following command on the device: # configure session abort """ - await self._cli(f"{self._cli_config_session} abort") + await self._cli(command=f"{self._cli_config_session} abort") async def diff(self) -> str: """Return the "diff" of the session config relative to the running config. @@ -257,7 +258,7 @@ async def diff(self) -> str: ---------- * https://www.gnu.org/software/diffutils/manual/diffutils.txt """ - return await self._cli(f"show session-config named {self.name} diffs", ofmt="text") # type: ignore[return-value] # text outformat returns str + return await self._cli(command=f"show session-config named {self.name} diffs", ofmt="text") async def load_file(self, filename: str, *, replace: bool = False) -> None: """Load the configuration from into the session configuration. @@ -281,12 +282,12 @@ async def load_file(self, filename: str, *, replace: bool = False) -> None: If there are any issues with loading the configuration file then a RuntimeError is raised with the error messages content. """ - commands: list[str | dict[str, Any]] = [self._cli_config_session] + commands: list[EapiSimpleCommand | EapiComplexCommand] = [self._cli_config_session] if replace: commands.append(self.CLI_CFG_FACTORY_RESET) commands.append(f"copy {filename} session-config") - res: list[dict[str, Any]] = await self._cli(commands=commands) # type: ignore[assignment] # JSON outformat of multiple commands returns list[dict[str, Any]] + res = await self._cli(commands=commands) checks_re = re.compile(r"error|abort|invalid", flags=re.IGNORECASE) messages = res[-1]["messages"] @@ -295,4 +296,4 @@ async def load_file(self, filename: str, *, replace: bool = False) -> None: async def write(self) -> None: """Save the running config to the startup config by issuing the command "write" to the device.""" - await self._cli("write") + await self._cli(command="write") diff --git a/asynceapi/device.py b/asynceapi/device.py index 9b68dc586..d23024767 100644 --- a/asynceapi/device.py +++ b/asynceapi/device.py @@ -10,7 +10,7 @@ from __future__ import annotations from socket import getservbyname -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal, overload # ----------------------------------------------------------------------------- # Public Imports @@ -25,7 +25,7 @@ from .errors import EapiCommandError if TYPE_CHECKING: - from collections.abc import Sequence + from ._types import EapiComplexCommand, EapiJsonOutput, EapiSimpleCommand, EapiTextOutput, JsonRpc # ----------------------------------------------------------------------------- # Exports @@ -121,18 +121,148 @@ async def check_connection(self) -> bool: """ return await port_check_url(self.base_url) + # Single command, JSON output, no suppression + @overload async def cli( self, - command: str | dict[str, Any] | None = None, - commands: Sequence[str | dict[str, Any]] | None = None, - ofmt: str | None = None, - version: int | str | None = "latest", + *, + command: EapiSimpleCommand | EapiComplexCommand, + commands: None = None, + ofmt: Literal["json"] = "json", + version: int | Literal["latest"] = "latest", + suppress_error: Literal[False] = False, + auto_complete: bool = False, + expand_aliases: bool = False, + timestamps: bool = False, + req_id: int | str | None = None, + ) -> EapiJsonOutput: ... + + # Multiple commands, JSON output, no suppression + @overload + async def cli( + self, + *, + command: None = None, + commands: list[EapiSimpleCommand | EapiComplexCommand], + ofmt: Literal["json"] = "json", + version: int | Literal["latest"] = "latest", + suppress_error: Literal[False] = False, + auto_complete: bool = False, + expand_aliases: bool = False, + timestamps: bool = False, + req_id: int | str | None = None, + ) -> list[EapiJsonOutput]: ... + + # Single command, TEXT output, no suppression + @overload + async def cli( + self, + *, + command: EapiSimpleCommand | EapiComplexCommand, + commands: None = None, + ofmt: Literal["text"], + version: int | Literal["latest"] = "latest", + suppress_error: Literal[False] = False, + auto_complete: bool = False, + expand_aliases: bool = False, + timestamps: bool = False, + req_id: int | str | None = None, + ) -> EapiTextOutput: ... + + # Multiple commands, TEXT output, no suppression + @overload + async def cli( + self, + *, + command: None = None, + commands: list[EapiSimpleCommand | EapiComplexCommand], + ofmt: Literal["text"], + version: int | Literal["latest"] = "latest", + suppress_error: Literal[False] = False, + auto_complete: bool = False, + expand_aliases: bool = False, + timestamps: bool = False, + req_id: int | str | None = None, + ) -> list[EapiTextOutput]: ... + + # Single command, JSON output, with suppression + @overload + async def cli( + self, + *, + command: EapiSimpleCommand | EapiComplexCommand, + commands: None = None, + ofmt: Literal["json"] = "json", + version: int | Literal["latest"] = "latest", + suppress_error: Literal[True], + auto_complete: bool = False, + expand_aliases: bool = False, + timestamps: bool = False, + req_id: int | str | None = None, + ) -> EapiJsonOutput | None: ... + + # Multiple commands, JSON output, with suppression + @overload + async def cli( + self, + *, + command: None = None, + commands: list[EapiSimpleCommand | EapiComplexCommand], + ofmt: Literal["json"] = "json", + version: int | Literal["latest"] = "latest", + suppress_error: Literal[True], + auto_complete: bool = False, + expand_aliases: bool = False, + timestamps: bool = False, + req_id: int | str | None = None, + ) -> list[EapiJsonOutput] | None: ... + + # Single command, TEXT output, with suppression + @overload + async def cli( + self, + *, + command: EapiSimpleCommand | EapiComplexCommand, + commands: None = None, + ofmt: Literal["text"], + version: int | Literal["latest"] = "latest", + suppress_error: Literal[True], + auto_complete: bool = False, + expand_aliases: bool = False, + timestamps: bool = False, + req_id: int | str | None = None, + ) -> EapiTextOutput | None: ... + + # Multiple commands, TEXT output, with suppression + @overload + async def cli( + self, + *, + command: None = None, + commands: list[EapiSimpleCommand | EapiComplexCommand], + ofmt: Literal["text"], + version: int | Literal["latest"] = "latest", + suppress_error: Literal[True], + auto_complete: bool = False, + expand_aliases: bool = False, + timestamps: bool = False, + req_id: int | str | None = None, + ) -> list[EapiTextOutput] | None: ... + + # Actual implementation + async def cli( + self, + command: EapiSimpleCommand | EapiComplexCommand | None = None, + commands: list[EapiSimpleCommand | EapiComplexCommand] | None = None, + ofmt: Literal["json", "text"] = "json", + version: int | Literal["latest"] = "latest", *, suppress_error: bool = False, auto_complete: bool = False, expand_aliases: bool = False, + timestamps: bool = False, req_id: int | str | None = None, - ) -> list[dict[str, Any] | str] | dict[str, Any] | str | None: + ) -> EapiJsonOutput | EapiTextOutput | list[EapiJsonOutput] | list[EapiTextOutput] | None: """Execute one or more CLI commands. Parameters @@ -143,6 +273,7 @@ async def cli( A list of commands to execute; results in a list of output responses. ofmt Either 'json' or 'text'; indicates the output format for the CLI commands. + eAPI defaults to 'json'. version By default the eAPI will use "version 1" for all API object models. This driver will, by default, always set version to "latest" so @@ -158,33 +289,54 @@ async def cli( response = dev.cli(..., suppress_error=True) auto_complete - Enabled/disables the command auto-compelete feature of the EAPI. Per the + Enabled/disables the command auto-compelete feature of the eAPI. Per the documentation: Allows users to use shorthand commands in eAPI calls. With this parameter included a user can send 'sh ver' via eAPI to get the output of 'show version'. expand_aliases - Enables/disables the command use of User defined alias. Per the + Enables/disables the command use of user-defined alias. Per the documentation: Allowed users to provide the expandAliases parameter to eAPI calls. This allows users to use aliased commands via the API. For example if an alias is configured as 'sv' for 'show version' then an API call with sv and the expandAliases parameter will return the output of show version. + timestamps + If True, return the per-command execution time. req_id A unique identifier that will be echoed back by the switch. May be a string or number. Returns ------- - list[dict[str, Any] | str] | dict[str, Any] | str | None - One or List of output responses, per the description above. + dict[str, Any] + Single command, JSON output, suppress_error=False + list[dict[str, Any]] + Multiple commands, JSON output, suppress_error=False + str + Single command, TEXT output, suppress_error=False + list[str] + Multiple commands, TEXT output, suppress_error=False + dict[str, Any] | None + Single command, JSON output, suppress_error=True + list[dict[str, Any]] | None + Multiple commands, JSON output, suppress_error=True + str | None + Single command, TEXT output, suppress_error=True + list[str] | None + Multiple commands, TEXT output, suppress_error=True """ - if not any((command, commands)): + if command and commands: + msg = "Cannot provide both 'command' and 'commands'" + raise RuntimeError(msg) + + cmds = [command] if command else commands + if not cmds: msg = "Required 'command' or 'commands'" raise RuntimeError(msg) jsonrpc = self._jsonrpc_command( - commands=[command] if command else commands, ofmt=ofmt, version=version, auto_complete=auto_complete, expand_aliases=expand_aliases, req_id=req_id + commands=cmds, ofmt=ofmt, version=version, auto_complete=auto_complete, expand_aliases=expand_aliases, timestamps=timestamps, req_id=req_id ) try: @@ -197,14 +349,15 @@ async def cli( def _jsonrpc_command( self, - commands: Sequence[str | dict[str, Any]] | None = None, - ofmt: str | None = None, - version: int | str | None = "latest", + commands: list[EapiSimpleCommand | EapiComplexCommand], + ofmt: Literal["json", "text"] = "json", + version: int | Literal["latest"] = "latest", *, auto_complete: bool = False, expand_aliases: bool = False, + timestamps: bool = False, req_id: int | str | None = None, - ) -> dict[str, Any]: + ) -> JsonRpc: """Create the JSON-RPC command dictionary object. Parameters @@ -213,6 +366,7 @@ def _jsonrpc_command( A list of commands to execute; results in a list of output responses. ofmt Either 'json' or 'text'; indicates the output format for the CLI commands. + eAPI defaults to 'json'. version By default the eAPI will use "version 1" for all API object models. This driver will, by default, always set version to "latest" so @@ -232,6 +386,8 @@ def _jsonrpc_command( For example if an alias is configured as 'sv' for 'show version' then an API call with sv and the expandAliases parameter will return the output of show version. + timestamps + If True, return the per-command execution time. req_id A unique identifier that will be echoed back by the switch. May be a string or number. @@ -241,25 +397,21 @@ def _jsonrpc_command( dict containing the JSON payload to run the command. """ - cmd: dict[str, Any] = { + return { "jsonrpc": "2.0", "method": "runCmds", "params": { "version": version, "cmds": commands, - "format": ofmt or self.EAPI_DEFAULT_OFMT, + "format": ofmt, + "autoComplete": auto_complete, + "expandAliases": expand_aliases, + "timestamps": timestamps, }, "id": req_id or id(self), } - if auto_complete is not None: - cmd["params"].update({"autoComplete": auto_complete}) - - if expand_aliases is not None: - cmd["params"].update({"expandAliases": expand_aliases}) - - return cmd - async def jsonrpc_exec(self, jsonrpc: dict[str, Any]) -> list[dict[str, Any] | str]: + async def jsonrpc_exec(self, jsonrpc: JsonRpc) -> list[EapiJsonOutput] | list[EapiTextOutput]: """Execute the JSON-RPC dictionary object. Parameters @@ -315,7 +467,7 @@ async def jsonrpc_exec(self, jsonrpc: dict[str, Any]) -> list[dict[str, Any] | s failed_cmd = commands[err_at] raise EapiCommandError( - passed=[get_output(cmd_data[cmd_i]) for cmd_i, cmd in enumerate(commands[:err_at])], + passed=[get_output(cmd_data[i]) for i in range(err_at)], failed=failed_cmd["cmd"] if isinstance(failed_cmd, dict) else failed_cmd, errors=cmd_data[err_at]["errors"], errmsg=err_msg, diff --git a/asynceapi/errors.py b/asynceapi/errors.py index ee1faf2ea..50b02c6fd 100644 --- a/asynceapi/errors.py +++ b/asynceapi/errors.py @@ -6,13 +6,16 @@ from __future__ import annotations -from typing import Any +from typing import TYPE_CHECKING import httpx +if TYPE_CHECKING: + from ._types import EapiComplexCommand, EapiJsonOutput, EapiSimpleCommand, EapiTextOutput + class EapiCommandError(RuntimeError): - """Exception class for EAPI command errors. + """Exception class for eAPI command errors. Attributes ---------- @@ -23,7 +26,14 @@ class EapiCommandError(RuntimeError): not_exec: a list of commands that were not executed """ - def __init__(self, failed: str, errors: list[str], errmsg: str, passed: list[str | dict[str, Any]], not_exec: list[dict[str, Any]]) -> None: + def __init__( + self, + failed: str, + errors: list[str], + errmsg: str, + passed: list[EapiJsonOutput] | list[EapiTextOutput], + not_exec: list[EapiSimpleCommand | EapiComplexCommand], + ) -> None: """Initialize for the EapiCommandError exception.""" self.failed = failed self.errmsg = errmsg diff --git a/tests/units/asynceapi/test_data.py b/tests/units/asynceapi/test_data.py index f164c1178..a1400fe80 100644 --- a/tests/units/asynceapi/test_data.py +++ b/tests/units/asynceapi/test_data.py @@ -3,7 +3,9 @@ # that can be found in the LICENSE file. """Unit tests data for the asynceapi client package.""" -SUCCESS_EAPI_RESPONSE = { +from asynceapi._types import EapiJsonOutput, JsonRpc + +SUCCESS_EAPI_RESPONSE: EapiJsonOutput = { "jsonrpc": "2.0", "id": "EapiExplorer-1", "result": [ @@ -49,7 +51,7 @@ } """Successful eAPI JSON response.""" -ERROR_EAPI_RESPONSE = { +ERROR_EAPI_RESPONSE: EapiJsonOutput = { "jsonrpc": "2.0", "id": "EapiExplorer-1", "error": { @@ -84,5 +86,5 @@ } """Error eAPI JSON response.""" -JSONRPC_REQUEST_TEMPLATE = {"jsonrpc": "2.0", "method": "runCmds", "params": {"version": 1, "cmds": [], "format": "json"}, "id": "EapiExplorer-1"} +JSONRPC_REQUEST_TEMPLATE: JsonRpc = {"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.""" diff --git a/tests/units/asynceapi/test_device.py b/tests/units/asynceapi/test_device.py index da7af1233..b2293d917 100644 --- a/tests/units/asynceapi/test_device.py +++ b/tests/units/asynceapi/test_device.py @@ -5,7 +5,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING import pytest from httpx import HTTPStatusError @@ -17,6 +17,8 @@ if TYPE_CHECKING: from pytest_httpx import HTTPXMock + from asynceapi._types import EapiComplexCommand, EapiSimpleCommand + @pytest.mark.parametrize( "cmds", @@ -30,10 +32,10 @@ async def test_jsonrpc_exec_success( asynceapi_device: Device, httpx_mock: HTTPXMock, - cmds: list[str | dict[str, Any]], + cmds: list[EapiSimpleCommand | EapiComplexCommand], ) -> 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 = JSONRPC_REQUEST_TEMPLATE.copy() jsonrpc_request["params"]["cmds"] = cmds httpx_mock.add_response(json=SUCCESS_EAPI_RESPONSE) @@ -55,13 +57,13 @@ async def test_jsonrpc_exec_success( async def test_jsonrpc_exec_eapi_command_error( asynceapi_device: Device, httpx_mock: HTTPXMock, - cmds: list[str | dict[str, Any]], + cmds: list[EapiSimpleCommand | EapiComplexCommand], ) -> 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 = JSONRPC_REQUEST_TEMPLATE.copy() jsonrpc_request["params"]["cmds"] = cmds - error_eapi_response: dict[str, Any] = ERROR_EAPI_RESPONSE.copy() + error_eapi_response = ERROR_EAPI_RESPONSE.copy() httpx_mock.add_response(json=error_eapi_response) with pytest.raises(EapiCommandError) as exc_info: @@ -76,7 +78,7 @@ async def test_jsonrpc_exec_eapi_command_error( 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 = JSONRPC_REQUEST_TEMPLATE.copy() jsonrpc_request["params"]["cmds"] = ["show version"] httpx_mock.add_response(status_code=500, text="Internal Server Error")