Merge branch 'main' into fix_ospf_module_failure_msg

.github/workflows/sonar.yml

@@ -20,7 +20,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           ref: ${{ github.event.pull_request.head.sha }}
-          fetch-depth: 0  # Shallow clones should be disabled for a better relevancy of analysis
+          fetch-depth: 0 # Shallow clones should be disabled for a better relevancy of analysis
       - name: Setup Python
         uses: actions/setup-python@v5
         with:
@@ -30,7 +30,7 @@ jobs:
       - name: "Run pytest via tox for ${{ matrix.python }}"
         run: tox
       - name: SonarCloud Scan
-        uses: SonarSource/sonarcloud-github-action@master
+        uses: SonarSource/sonarqube-scan-action@v5.0.0
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

.pre-commit-config.yaml

@@ -43,16 +43,16 @@ repos:
           - --allow-past-years
           - --fuzzy-match-generates-todo
           - --comment-style
-          - '<!--|  ~|  -->'
+          - "<!--|  ~|  -->"
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.9.6
+    rev: v0.9.7
     hooks:
-    - id: ruff
-      name: Run Ruff linter
-      args: [ --fix ]
-    - id: ruff-format
-      name: Run Ruff formatter
+      - id: ruff
+        name: Run Ruff linter
+        args: [--fix]
+      - id: ruff-format
+        name: Run Ruff formatter
 
   - repo: https://github.com/pycqa/pylint
     rev: "v3.3.4"
@@ -62,9 +62,9 @@ repos:
         description: This hook runs pylint.
         types: [python]
         args:
-          - -rn                      # Only display messages
-          - -sn                      # Don't display the score
-          - --rcfile=pyproject.toml  # Link to config file
+          - -rn # Only display messages
+          - -sn # Don't display the score
+          - --rcfile=pyproject.toml # Link to config file
         additional_dependencies:
           - anta[cli]
           - types-PyYAML
@@ -123,5 +123,14 @@ repos:
         pass_filenames: false
         additional_dependencies:
           - anta[cli]
-          # TODO: next can go once we have it added to anta properly
-          - numpydoc
+      - id: doc-snippets
+        name: Generate doc snippets
+        entry: >-
+          sh -c "docs/scripts/generate_doc_snippets.py"
+        language: python
+        types: [python]
+        files: anta/cli/
+        verbose: true
+        pass_filenames: false
+        additional_dependencies:
+          - anta[cli]

anta/cli/nrfu/__init__.py

@@ -42,9 +42,10 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
             if "--help" not in args:
                 raise
 
-            # remove the required params so that help can display
+            # Fake presence of the required params so that help can display
             for param in self.params:
-                param.required = False
+                if param.required:
+                    param.value_is_missing = lambda value: False  # type: ignore[method-assign] # noqa: ARG005
 
             return super().parse_args(ctx, args)
 

anta/cli/nrfu/commands.py

@@ -45,7 +45,10 @@ def table(ctx: click.Context, group_by: Literal["device", "test"] | None) -> Non
     help="Path to save report as a JSON file",
 )
 def json(ctx: click.Context, output: pathlib.Path | None) -> None:
-    """ANTA command to check network state with JSON results."""
+    """ANTA command to check network state with JSON results.
+
+    If no `--output` is specified, the output is printed to stdout.
+    """
     run_tests(ctx)
     print_json(ctx, output=output)
     exit_with_code(ctx)
@@ -72,11 +75,11 @@ def text(ctx: click.Context) -> None:
         path_type=pathlib.Path,
     ),
     show_envvar=True,
-    required=False,
+    required=True,
     help="Path to save report as a CSV file",
 )
 def csv(ctx: click.Context, csv_output: pathlib.Path) -> None:
-    """ANTA command to check network states with CSV result."""
+    """ANTA command to check network state with CSV report."""
     run_tests(ctx)
     save_to_csv(ctx, csv_file=csv_output)
     exit_with_code(ctx)

anta/custom_types.py

@@ -24,6 +24,13 @@
 """Match hostname like `my-hostname`, `my-hostname-1`, `my-hostname-1-2`."""
 
 
+# Regular expression for BGP redistributed routes
+REGEX_IPV4_UNICAST = r"ipv4[-_ ]?unicast$"
+REGEX_IPV4_MULTICAST = r"ipv4[-_ ]?multicast$"
+REGEX_IPV6_UNICAST = r"ipv6[-_ ]?unicast$"
+REGEX_IPV6_MULTICAST = r"ipv6[-_ ]?multicast$"
+
+
 def aaa_group_prefix(v: str) -> str:
     """Prefix the AAA method with 'group' if it is known."""
     built_in_methods = ["local", "none", "logging"]
@@ -92,10 +99,10 @@ def bgp_multiprotocol_capabilities_abbreviations(value: str) -> str:
     patterns = {
         f"{r'dynamic[-_ ]?path[-_ ]?selection$'}": "dps",
         f"{r'dps$'}": "dps",
-        f"{r'ipv4[-_ ]?unicast$'}": "ipv4Unicast",
-        f"{r'ipv6[-_ ]?unicast$'}": "ipv6Unicast",
-        f"{r'ipv4[-_ ]?multicast$'}": "ipv4Multicast",
-        f"{r'ipv6[-_ ]?multicast$'}": "ipv6Multicast",
+        f"{REGEX_IPV4_UNICAST}": "ipv4Unicast",
+        f"{REGEX_IPV6_UNICAST}": "ipv6Unicast",
+        f"{REGEX_IPV4_MULTICAST}": "ipv4Multicast",
+        f"{REGEX_IPV6_MULTICAST}": "ipv6Multicast",
         f"{r'ipv4[-_ ]?labeled[-_ ]?Unicast$'}": "ipv4MplsLabels",
         f"{r'ipv4[-_ ]?mpls[-_ ]?labels$'}": "ipv4MplsLabels",
         f"{r'ipv6[-_ ]?labeled[-_ ]?Unicast$'}": "ipv6MplsLabels",
@@ -132,6 +139,54 @@ def validate_regex(value: str) -> str:
     return value
 
 
+def bgp_redistributed_route_proto_abbreviations(value: str) -> str:
+    """Abbreviations for different BGP redistributed route protocols.
+
+    Handles different separators (hyphen, underscore, space) and case sensitivity.
+
+    Examples
+    --------
+    ```python
+    >>> bgp_redistributed_route_proto_abbreviations("IPv4 Unicast")
+    'v4u'
+    >>> bgp_redistributed_route_proto_abbreviations("IPv4-multicast")
+    'v4m'
+    >>> bgp_redistributed_route_proto_abbreviations("IPv6_multicast")
+    'v6m'
+    >>> bgp_redistributed_route_proto_abbreviations("ipv6unicast")
+    'v6u'
+    ```
+    """
+    patterns = {REGEX_IPV4_UNICAST: "v4u", REGEX_IPV4_MULTICAST: "v4m", REGEX_IPV6_UNICAST: "v6u", REGEX_IPV6_MULTICAST: "v6m"}
+
+    for pattern, replacement in patterns.items():
+        match = re.match(pattern, value, re.IGNORECASE)
+        if match:
+            return replacement
+
+    return value
+
+
+def update_bgp_redistributed_proto_user(value: str) -> str:
+    """Update BGP redistributed route `User` proto with EOS SDK.
+
+    Examples
+    --------
+    ```python
+    >>> update_bgp_redistributed_proto_user("User")
+    'EOS SDK'
+    >>> update_bgp_redistributed_proto_user("Bgp")
+    'Bgp'
+    >>> update_bgp_redistributed_proto_user("RIP")
+    'RIP'
+    ```
+    """
+    if value == "User":
+        value = "EOS SDK"
+
+    return value
+
+
 # AntaTest.Input types
 AAAAuthMethod = Annotated[str, AfterValidator(aaa_group_prefix)]
 Vlan = Annotated[int, Field(ge=0, le=4094)]
@@ -319,3 +374,23 @@ def snmp_v3_prefix(auth_type: Literal["auth", "priv", "noauth"]) -> str:
     "inVersionErrs", "inBadCommunityNames", "inBadCommunityUses", "inParseErrs", "outTooBigErrs", "outNoSuchNameErrs", "outBadValueErrs", "outGeneralErrs"
 ]
 SnmpVersionV3AuthType = Annotated[Literal["auth", "priv", "noauth"], AfterValidator(snmp_v3_prefix)]
+RedistributedProtocol = Annotated[
+    Literal[
+        "AttachedHost",
+        "Bgp",
+        "Connected",
+        "Dynamic",
+        "IS-IS",
+        "OSPF Internal",
+        "OSPF External",
+        "OSPF Nssa-External",
+        "OSPFv3 Internal",
+        "OSPFv3 External",
+        "OSPFv3 Nssa-External",
+        "RIP",
+        "Static",
+        "User",
+    ],
+    AfterValidator(update_bgp_redistributed_proto_user),
+]
+RedistributedAfiSafi = Annotated[Literal["v4u", "v4m", "v6u", "v6m"], BeforeValidator(bgp_redistributed_route_proto_abbreviations)]

anta/device.py

@@ -32,6 +32,10 @@
 # https://github.com/pyca/cryptography/issues/7236#issuecomment-1131908472
 CLIENT_KEYS = asyncssh.public_key.load_default_keypairs()
 
+# Limit concurrency to 100 requests (HTTPX default) to avoid high-concurrency performance issues
+# See: https://github.com/encode/httpx/issues/3215
+MAX_CONCURRENT_REQUESTS = 100
+
 
 class AntaCache:
     """Class to be used as cache.
@@ -296,6 +300,7 @@ async def copy(self, sources: list[Path], destination: Path, direction: Literal[
         raise NotImplementedError(msg)
 
 
+# pylint: disable=too-many-instance-attributes
 class AsyncEOSDevice(AntaDevice):
     """Implementation of AntaDevice for EOS using aio-eapi.
 
@@ -388,6 +393,10 @@ def __init__(  # noqa: PLR0913
             host=host, port=ssh_port, username=username, password=password, client_keys=CLIENT_KEYS, **ssh_params
         )
 
+        # In Python 3.9, Semaphore must be created within a running event loop
+        # TODO: Once we drop Python 3.9 support, initialize the semaphore here
+        self._command_semaphore: asyncio.Semaphore | None = None
+
     def __rich_repr__(self) -> Iterator[tuple[str, Any]]:
         """Implement Rich Repr Protocol.
 
@@ -431,6 +440,15 @@ def _keys(self) -> tuple[Any, ...]:
         """
         return (self._session.host, self._session.port)
 
+    async def _get_semaphore(self) -> asyncio.Semaphore:
+        """Return the semaphore, initializing it if needed.
+
+        TODO: Remove this method once we drop Python 3.9 support.
+        """
+        if self._command_semaphore is None:
+            self._command_semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
+        return self._command_semaphore
+
     async def _collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None:
         """Collect device command output from EOS using aio-eapi.
 
@@ -445,57 +463,63 @@ 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]] = []
-        if self.enable and self._enable_password is not None:
-            commands.append(
-                {
-                    "cmd": "enable",
-                    "input": str(self._enable_password),
-                },
-            )
-        elif self.enable:
-            # No password
-            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(
-                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:
-            # This block catches exceptions related to EOS issuing an error.
-            self._log_eapi_command_error(command, e)
-        except TimeoutException as e:
-            # This block catches Timeout exceptions.
-            command.errors = [exc_to_str(e)]
-            timeouts = self._session.timeout.as_dict()
-            logger.error(
-                "%s occurred while sending a command to %s. Consider increasing the timeout.\nCurrent timeouts: Connect: %s | Read: %s | Write: %s | Pool: %s",
-                exc_to_str(e),
-                self.name,
-                timeouts["connect"],
-                timeouts["read"],
-                timeouts["write"],
-                timeouts["pool"],
-            )
-        except (ConnectError, OSError) as e:
-            # This block catches OSError and socket issues related exceptions.
-            command.errors = [exc_to_str(e)]
-            if (isinstance(exc := e.__cause__, httpcore.ConnectError) and isinstance(os_error := exc.__context__, OSError)) or isinstance(os_error := e, OSError):  # pylint: disable=no-member
-                if isinstance(os_error.__cause__, OSError):
-                    os_error = os_error.__cause__
-                logger.error("A local OS error occurred while connecting to %s: %s.", self.name, os_error)
-            else:
+        semaphore = await self._get_semaphore()
+
+        async with semaphore:
+            commands: list[dict[str, str | int]] = []
+            if self.enable and self._enable_password is not None:
+                commands.append(
+                    {
+                        "cmd": "enable",
+                        "input": str(self._enable_password),
+                    },
+                )
+            elif self.enable:
+                # No password
+                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(
+                    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:
+                # This block catches exceptions related to EOS issuing an error.
+                self._log_eapi_command_error(command, e)
+            except TimeoutException as e:
+                # This block catches Timeout exceptions.
+                command.errors = [exc_to_str(e)]
+                timeouts = self._session.timeout.as_dict()
+                logger.error(
+                    "%s occurred while sending a command to %s. Consider increasing the timeout.\nCurrent timeouts: Connect: %s | Read: %s | Write: %s | Pool: %s",
+                    exc_to_str(e),
+                    self.name,
+                    timeouts["connect"],
+                    timeouts["read"],
+                    timeouts["write"],
+                    timeouts["pool"],
+                )
+            except (ConnectError, OSError) as e:
+                # This block catches OSError and socket issues related exceptions.
+                command.errors = [exc_to_str(e)]
+                # pylint: disable=no-member
+                if (isinstance(exc := e.__cause__, httpcore.ConnectError) and isinstance(os_error := exc.__context__, OSError)) or isinstance(
+                    os_error := e, OSError
+                ):
+                    if isinstance(os_error.__cause__, OSError):
+                        os_error = os_error.__cause__
+                    logger.error("A local OS error occurred while connecting to %s: %s.", self.name, os_error)
+                else:
+                    anta_log_exception(e, f"An error occurred while issuing an eAPI request to {self.name}", logger)
+            except HTTPError as e:
+                # This block catches most of the httpx Exceptions and logs a general message.
+                command.errors = [exc_to_str(e)]
                 anta_log_exception(e, f"An error occurred while issuing an eAPI request to {self.name}", logger)
-        except HTTPError as e:
-            # This block catches most of the httpx Exceptions and logs a general message.
-            command.errors = [exc_to_str(e)]
-            anta_log_exception(e, f"An error occurred while issuing an eAPI request to {self.name}", logger)
-        logger.debug("%s: %s", self.name, command)
+            logger.debug("%s: %s", self.name, command)
 
     def _log_eapi_command_error(self, command: AntaCommand, e: asynceapi.EapiCommandError) -> None:
         """Appropriately log the eapi command error."""

anta/input_models/avt.py

@@ -33,4 +33,4 @@ def __str__(self) -> str:
         AVT CONTROL-PLANE-PROFILE VRF: default (Destination: 10.101.255.2, Next-hop: 10.101.255.1)
 
         """
-        return f"AVT {self.avt_name} VRF: {self.vrf} (Destination: {self.destination}, Next-hop: {self.next_hop})"
+        return f"AVT: {self.avt_name} VRF: {self.vrf} Destination: {self.destination} Next-hop: {self.next_hop}"