Adding upstream version 0.14.0.upstream/0.14.0

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 107 insertions, 114 deletions

diff --git a/tests/units/test_device.py b/tests/units/test_device.py
index 845da2b..c901a3d 100644
--- a/tests/units/test_device.py
+++ b/tests/units/test_device.py
@@ -1,20 +1,17 @@
 # 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.
-"""
-test anta.device.py
-"""
+"""test anta.device.py."""
 
 from __future__ import annotations
 
 import asyncio
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 from unittest.mock import patch
 
 import httpx
 import pytest
-from _pytest.mark.structures import ParameterSet
 from asyncssh import SSHClientConnection, SSHClientConnectionOptions
 from rich import print as rprint
 
@@ -24,6 +21,9 @@ from anta.models import AntaCommand
 from tests.lib.fixture import COMMAND_OUTPUT
 from tests.lib.utils import generate_test_ids_list
 
+if TYPE_CHECKING:
+    from _pytest.mark.structures import ParameterSet
+
 INIT_DATA: list[dict[str, Any]] = [
     {
         "name": "no name, no port",
@@ -155,8 +155,8 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
                         "memTotal": 8099732,
                         "memFree": 4989568,
                         "isIntlVersion": False,
-                    }
-                ]
+                    },
+                ],
             },
         },
         "expected": {
@@ -211,7 +211,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
                         "memFree": 4989568,
                         "isIntlVersion": False,
                     },
-                ]
+                ],
             },
         },
         "expected": {
@@ -266,7 +266,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
                         "memFree": 4989568,
                         "isIntlVersion": False,
                     },
-                ]
+                ],
             },
         },
         "expected": {
@@ -322,7 +322,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
                         "memFree": 4989568,
                         "isIntlVersion": False,
                     },
-                ]
+                ],
             },
         },
         "expected": {
@@ -356,8 +356,12 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
             "command": "show version",
             "patch_kwargs": {
                 "side_effect": aioeapi.EapiCommandError(
-                    passed=[], failed="show version", errors=["Authorization denied for command 'show version'"], errmsg="Invalid command", not_exec=[]
-                )
+                    passed=[],
+                    failed="show version",
+                    errors=["Authorization denied for command 'show version'"],
+                    errmsg="Invalid command",
+                    not_exec=[],
+                ),
             },
         },
         "expected": {"output": None, "errors": ["Authorization denied for command 'show version'"]},
@@ -369,7 +373,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
             "command": "show version",
             "patch_kwargs": {"side_effect": httpx.HTTPError(message="404")},
         },
-        "expected": {"output": None, "errors": ["404"]},
+        "expected": {"output": None, "errors": ["HTTPError: 404"]},
     },
     {
         "name": "httpx.ConnectError",
@@ -378,7 +382,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
             "command": "show version",
             "patch_kwargs": {"side_effect": httpx.ConnectError(message="Cannot open port")},
         },
-        "expected": {"output": None, "errors": ["Cannot open port"]},
+        "expected": {"output": None, "errors": ["ConnectError: Cannot open port"]},
     },
 ]
 AIOEAPI_COPY_DATA: list[dict[str, Any]] = [
@@ -387,7 +391,7 @@ AIOEAPI_COPY_DATA: list[dict[str, Any]] = [
         "device": {},
         "copy": {
             "sources": [Path("/mnt/flash"), Path("/var/log/agents")],
-            "destination": Path("."),
+            "destination": Path(),
             "direction": "from",
         },
     },
@@ -396,7 +400,7 @@ AIOEAPI_COPY_DATA: list[dict[str, Any]] = [
         "device": {},
         "copy": {
             "sources": [Path("/mnt/flash"), Path("/var/log/agents")],
-            "destination": Path("."),
+            "destination": Path(),
             "direction": "to",
         },
     },
@@ -405,7 +409,7 @@ AIOEAPI_COPY_DATA: list[dict[str, Any]] = [
         "device": {},
         "copy": {
             "sources": [Path("/mnt/flash"), Path("/var/log/agents")],
-            "destination": Path("."),
+            "destination": Path(),
             "direction": "wrong",
         },
     },
@@ -417,26 +421,28 @@ REFRESH_DATA: list[dict[str, Any]] = [
         "patch_kwargs": (
             {"return_value": True},
             {
-                "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,
-                }
+                "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,
+                    }
+                ],
             },
         ),
         "expected": {"is_online": True, "established": True, "hw_model": "DCS-7280CR3-32P4-F"},
@@ -466,7 +472,7 @@ REFRESH_DATA: list[dict[str, Any]] = [
                     "memTotal": 8099732,
                     "memFree": 4989568,
                     "isIntlVersion": False,
-                }
+                },
             },
         ),
         "expected": {"is_online": False, "established": False, "hw_model": None},
@@ -477,25 +483,27 @@ REFRESH_DATA: list[dict[str, Any]] = [
         "patch_kwargs": (
             {"return_value": True},
             {
-                "return_value": {
-                    "mfgName": "Arista",
-                    "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,
-                }
+                "return_value": [
+                    {
+                        "mfgName": "Arista",
+                        "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,
+                    }
+                ],
             },
         ),
         "expected": {"is_online": True, "established": False, "hw_model": None},
@@ -507,8 +515,12 @@ REFRESH_DATA: list[dict[str, Any]] = [
             {"return_value": True},
             {
                 "side_effect": aioeapi.EapiCommandError(
-                    passed=[], failed="show version", errors=["Authorization denied for command 'show version'"], errmsg="Invalid command", not_exec=[]
-                )
+                    passed=[],
+                    failed="show version",
+                    errors=["Authorization denied for command 'show version'"],
+                    errmsg="Invalid command",
+                    not_exec=[],
+                ),
             },
         ),
         "expected": {"is_online": True, "established": False, "hw_model": None},
@@ -599,21 +611,17 @@ CACHE_STATS_DATA: list[ParameterSet] = [
 
 
 class TestAntaDevice:
-    """
-    Test for anta.device.AntaDevice Abstract class
-    """
+    """Test for anta.device.AntaDevice Abstract class."""
 
-    @pytest.mark.asyncio
+    @pytest.mark.asyncio()
     @pytest.mark.parametrize(
-        "device, command_data, expected_data",
-        map(lambda d: (d["device"], d["command"], d["expected"]), COLLECT_DATA),
+        ("device", "command_data", "expected_data"),
+        ((d["device"], d["command"], d["expected"]) for d in COLLECT_DATA),
         indirect=["device"],
         ids=generate_test_ids_list(COLLECT_DATA),
     )
     async def test_collect(self, device: AntaDevice, command_data: dict[str, Any], expected_data: dict[str, Any]) -> None:
-        """
-        Test AntaDevice.collect behavior
-        """
+        """Test AntaDevice.collect behavior."""
         command = AntaCommand(command=command_data["command"], use_cache=command_data["use_cache"])
 
         # Dummy output for cache hit
@@ -646,32 +654,21 @@ class TestAntaDevice:
             assert device.cache is None
             device._collect.assert_called_once_with(command=command)  # type: ignore[attr-defined]  # pylint: disable=protected-access
 
-    @pytest.mark.parametrize("device, expected", CACHE_STATS_DATA, indirect=["device"])
+    @pytest.mark.parametrize(("device", "expected"), CACHE_STATS_DATA, indirect=["device"])
     def test_cache_statistics(self, device: AntaDevice, expected: dict[str, Any] | None) -> None:
-        """
-        Verify that when cache statistics attribute does not exist
-        TODO add a test where cache has some value
-        """
-        assert device.cache_statistics == expected
+        """Verify that when cache statistics attribute does not exist.
 
-    def test_supports(self, device: AntaDevice) -> None:
+        TODO add a test where cache has some value.
         """
-        Test if the supports() method
-        """
-        command = AntaCommand(command="show hardware counter drop", errors=["Unavailable command (not supported on this hardware platform) (at token 2: 'counter')"])
-        assert device.supports(command) is False
-        command = AntaCommand(command="show hardware counter drop")
-        assert device.supports(command) is True
+        assert device.cache_statistics == expected
 
 
 class TestAsyncEOSDevice:
-    """
-    Test for anta.device.AsyncEOSDevice
-    """
+    """Test for anta.device.AsyncEOSDevice."""
 
     @pytest.mark.parametrize("data", INIT_DATA, ids=generate_test_ids_list(INIT_DATA))
     def test__init__(self, data: dict[str, Any]) -> None:
-        """Test the AsyncEOSDevice constructor"""
+        """Test the AsyncEOSDevice constructor."""
         device = AsyncEOSDevice(**data["device"])
 
         assert device.name == data["expected"]["name"]
@@ -683,12 +680,12 @@ class TestAsyncEOSDevice:
             assert device.cache_locks is not None
         hash(device)
 
-        with patch("anta.device.__DEBUG__", True):
+        with patch("anta.device.__DEBUG__", new=True):
             rprint(device)
 
     @pytest.mark.parametrize("data", EQUALITY_DATA, ids=generate_test_ids_list(EQUALITY_DATA))
     def test__eq(self, data: dict[str, Any]) -> None:
-        """Test the AsyncEOSDevice equality"""
+        """Test the AsyncEOSDevice equality."""
         device1 = AsyncEOSDevice(**data["device1"])
         device2 = AsyncEOSDevice(**data["device2"])
         if data["expected"]:
@@ -696,49 +693,45 @@ class TestAsyncEOSDevice:
         else:
             assert device1 != device2
 
-    @pytest.mark.asyncio
+    @pytest.mark.asyncio()
     @pytest.mark.parametrize(
-        "async_device, patch_kwargs, expected",
-        map(lambda d: (d["device"], d["patch_kwargs"], d["expected"]), REFRESH_DATA),
+        ("async_device", "patch_kwargs", "expected"),
+        ((d["device"], d["patch_kwargs"], d["expected"]) for d in REFRESH_DATA),
         ids=generate_test_ids_list(REFRESH_DATA),
         indirect=["async_device"],
     )
     async def test_refresh(self, async_device: AsyncEOSDevice, patch_kwargs: list[dict[str, Any]], expected: dict[str, Any]) -> None:
         # pylint: disable=protected-access
-        """Test AsyncEOSDevice.refresh()"""
-        with patch.object(async_device._session, "check_connection", **patch_kwargs[0]):
-            with patch.object(async_device._session, "cli", **patch_kwargs[1]):
-                await async_device.refresh()
-                async_device._session.check_connection.assert_called_once()
-                if expected["is_online"]:
-                    async_device._session.cli.assert_called_once()
-                assert async_device.is_online == expected["is_online"]
-                assert async_device.established == expected["established"]
-                assert async_device.hw_model == expected["hw_model"]
+        """Test AsyncEOSDevice.refresh()."""
+        with patch.object(async_device._session, "check_connection", **patch_kwargs[0]), patch.object(async_device._session, "cli", **patch_kwargs[1]):
+            await async_device.refresh()
+            async_device._session.check_connection.assert_called_once()
+            if expected["is_online"]:
+                async_device._session.cli.assert_called_once()
+            assert async_device.is_online == expected["is_online"]
+            assert async_device.established == expected["established"]
+            assert async_device.hw_model == expected["hw_model"]
 
-    @pytest.mark.asyncio
+    @pytest.mark.asyncio()
     @pytest.mark.parametrize(
-        "async_device, command, expected",
-        map(lambda d: (d["device"], d["command"], d["expected"]), AIOEAPI_COLLECT_DATA),
+        ("async_device", "command", "expected"),
+        ((d["device"], d["command"], d["expected"]) for d in AIOEAPI_COLLECT_DATA),
         ids=generate_test_ids_list(AIOEAPI_COLLECT_DATA),
         indirect=["async_device"],
     )
     async def test__collect(self, async_device: AsyncEOSDevice, command: dict[str, Any], expected: dict[str, Any]) -> None:
         # pylint: disable=protected-access
-        """Test AsyncEOSDevice._collect()"""
-        if "revision" in command:
-            cmd = AntaCommand(command=command["command"], revision=command["revision"])
-        else:
-            cmd = AntaCommand(command=command["command"])
+        """Test AsyncEOSDevice._collect()."""
+        cmd = AntaCommand(command=command["command"], revision=command["revision"]) if "revision" in command else AntaCommand(command=command["command"])
         with patch.object(async_device._session, "cli", **command["patch_kwargs"]):
             await async_device.collect(cmd)
-            commands = []
+            commands: list[dict[str, Any]] = []
             if async_device.enable and async_device._enable_password is not None:
                 commands.append(
                     {
                         "cmd": "enable",
                         "input": str(async_device._enable_password),
-                    }
+                    },
                 )
             elif async_device.enable:
                 # No password
@@ -751,15 +744,15 @@ class TestAsyncEOSDevice:
             assert cmd.output == expected["output"]
             assert cmd.errors == expected["errors"]
 
-    @pytest.mark.asyncio
+    @pytest.mark.asyncio()
     @pytest.mark.parametrize(
-        "async_device, copy",
-        map(lambda d: (d["device"], d["copy"]), AIOEAPI_COPY_DATA),
+        ("async_device", "copy"),
+        ((d["device"], d["copy"]) for d in AIOEAPI_COPY_DATA),
         ids=generate_test_ids_list(AIOEAPI_COPY_DATA),
         indirect=["async_device"],
     )
     async def test_copy(self, async_device: AsyncEOSDevice, copy: dict[str, Any]) -> None:
-        """Test AsyncEOSDevice.copy()"""
+        """Test AsyncEOSDevice.copy()."""
         conn = SSHClientConnection(asyncio.get_event_loop(), SSHClientConnectionOptions())
         with patch("asyncssh.connect") as connect_mock:
             connect_mock.return_value.__aenter__.return_value = conn