Merge branch 'main' into refactor/runner_limit

.pre-commit-config.yaml

@@ -46,7 +46,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.3
+    rev: v0.8.4
     hooks:
     - id: ruff
       name: Run Ruff linter
@@ -86,7 +86,7 @@ repos:
         types: [text]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.13.0
+    rev: v1.14.0
     hooks:
       - id: mypy
         name: Check typing with mypy

anta/cli/get/commands.py

@@ -75,7 +75,11 @@ def from_cvp(ctx: click.Context, output: Path, host: str, username: str, passwor
         # Get devices under a container
         logger.info("Getting inventory for container %s from CloudVision instance '%s'", container, host)
         cvp_inventory = clnt.api.get_devices_in_container(container)
-    create_inventory_from_cvp(cvp_inventory, output)
+    try:
+        create_inventory_from_cvp(cvp_inventory, output)
+    except OSError as e:
+        logger.error(str(e))
+        ctx.exit(ExitCode.USAGE_ERROR)
 
 
 @click.command
@@ -101,7 +105,7 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i
             output=output,
             ansible_group=ansible_group,
         )
-    except ValueError as e:
+    except (ValueError, OSError) as e:
         logger.error(str(e))
         ctx.exit(ExitCode.USAGE_ERROR)
 

anta/cli/get/utils.py

@@ -122,11 +122,28 @@ def get_cv_token(cvp_ip: str, cvp_username: str, cvp_password: str, *, verify_ce
 
 
 def write_inventory_to_file(hosts: list[AntaInventoryHost], output: Path) -> None:
-    """Write a file inventory from pydantic models."""
+    """Write a file inventory from pydantic models.
+
+    Parameters
+    ----------
+    hosts:
+        the list of AntaInventoryHost to write to an inventory file
+    output:
+        the Path where the inventory should be written.
+
+    Raises
+    ------
+    OSError
+        When anything goes wrong while writing the file.
+    """
     i = AntaInventoryInput(hosts=hosts)
-    with output.open(mode="w", encoding="UTF-8") as out_fd:
-        out_fd.write(yaml.dump({AntaInventory.INVENTORY_ROOT_KEY: yaml.safe_load(i.yaml())}))
-    logger.info("ANTA inventory file has been created: '%s'", output)
+    try:
+        with output.open(mode="w", encoding="UTF-8") as out_fd:
+            out_fd.write(yaml.dump({AntaInventory.INVENTORY_ROOT_KEY: yaml.safe_load(i.yaml())}))
+        logger.info("ANTA inventory file has been created: '%s'", output)
+    except OSError as exc:
+        msg = f"Could not write inventory to path '{output}'."
+        raise OSError(msg) from exc
 
 
 def create_inventory_from_cvp(inv: list[dict[str, Any]], output: Path) -> None:

anta/tests/interfaces.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import re
-from ipaddress import IPv4Network
+from ipaddress import IPv4Interface
 from typing import Any, ClassVar
 
 from pydantic import BaseModel, Field
@@ -629,9 +629,9 @@ class VerifyInterfaceIPv4(AntaTest):
       - VerifyInterfaceIPv4:
           interfaces:
             - name: Ethernet2
-              primary_ip: 172.30.11.0/31
+              primary_ip: 172.30.11.1/31
               secondary_ips:
-                - 10.10.10.0/31
+                - 10.10.10.1/31
                 - 10.10.10.10/31
     ```
     """
@@ -651,9 +651,9 @@ class InterfaceDetail(BaseModel):
 
             name: Interface
             """Name of the interface."""
-            primary_ip: IPv4Network
+            primary_ip: IPv4Interface
             """Primary IPv4 address in CIDR notation."""
-            secondary_ips: list[IPv4Network] | None = None
+            secondary_ips: list[IPv4Interface] | None = None
             """Optional list of secondary IPv4 addresses in CIDR notation."""
 
     def render(self, template: AntaTemplate) -> list[AntaCommand]:

examples/tests.yaml

@@ -217,9 +217,9 @@ anta.tests.interfaces:
       # Verifies the interface IPv4 addresses.
       interfaces:
         - name: Ethernet2
-          primary_ip: 172.30.11.0/31
+          primary_ip: 172.30.11.1/31
           secondary_ips:
-            - 10.10.10.0/31
+            - 10.10.10.1/31
             - 10.10.10.10/31
   - VerifyInterfaceUtilization:
       # Verifies that the utilization of interfaces is below a certain threshold.

tests/units/anta_tests/test_interfaces.py

@@ -1969,8 +1969,8 @@
                 "interfaces": {
                     "Ethernet2": {
                         "interfaceAddress": {
-                            "primaryIp": {"address": "172.30.11.0", "maskLen": 31},
-                            "secondaryIpsOrderedList": [{"address": "10.10.10.0", "maskLen": 31}, {"address": "10.10.10.10", "maskLen": 31}],
+                            "primaryIp": {"address": "172.30.11.1", "maskLen": 31},
+                            "secondaryIpsOrderedList": [{"address": "10.10.10.1", "maskLen": 31}, {"address": "10.10.10.10", "maskLen": 31}],
                         }
                     }
                 }
@@ -1988,7 +1988,7 @@
         ],
         "inputs": {
             "interfaces": [
-                {"name": "Ethernet2", "primary_ip": "172.30.11.0/31", "secondary_ips": ["10.10.10.0/31", "10.10.10.10/31"]},
+                {"name": "Ethernet2", "primary_ip": "172.30.11.1/31", "secondary_ips": ["10.10.10.1/31", "10.10.10.10/31"]},
                 {"name": "Ethernet12", "primary_ip": "172.30.11.10/31", "secondary_ips": ["10.10.10.10/31", "10.10.10.20/31"]},
             ]
         },

tests/units/cli/get/test_commands.py

@@ -114,6 +114,27 @@ def mock_cvp_connect(_self: CvpClient, *_args: str, **_kwargs: str) -> None:
     assert result.exit_code == ExitCode.OK
 
 
+def test_from_cvp_os_error(tmp_path: Path, click_runner: CliRunner, caplog: pytest.LogCaptureFixture) -> None:
+    """Test from_cvp when an OSError occurs."""
+    output: Path = tmp_path / "output.yml"
+    cli_args = ["get", "from-cvp", "--output", str(output), "--host", "42.42.42.42", "--username", "anta", "--password", "anta"]
+
+    with (
+        patch("anta.cli.get.commands.get_cv_token", autospec=True, side_effect=None),
+        patch("cvprac.cvp_client.CvpClient.connect", autospec=True, side_effect=None) as mocked_cvp_connect,
+        patch("cvprac.cvp_client.CvpApi.get_inventory", autospec=True, return_value=[]) as mocked_get_inventory,
+        patch("cvprac.cvp_client.CvpApi.get_devices_in_container", autospec=True, return_value=[]),
+        patch("anta.cli.get.utils.Path.open", side_effect=OSError("Permission denied")),
+    ):
+        result = click_runner.invoke(anta, cli_args)
+
+    mocked_cvp_connect.assert_called_once()
+    mocked_get_inventory.assert_called_once()
+    assert not output.exists()
+    assert "Could not write inventory to path" in caplog.text
+    assert result.exit_code == ExitCode.USAGE_ERROR
+
+
 @pytest.mark.parametrize(
     ("ansible_inventory", "ansible_group", "expected_exit", "expected_log"),
     [

tests/units/reporter/test_md_reporter.py

@@ -5,7 +5,7 @@
 
 from __future__ import annotations
 
-from io import StringIO
+from io import BytesIO, TextIOWrapper
 from pathlib import Path
 
 import pytest
@@ -46,7 +46,7 @@ def generate_section(self) -> None:
 
     results = ResultManager()
 
-    with StringIO() as mock_file:
+    with TextIOWrapper(BytesIO(b"1 2 3")) as mock_file:
         report = FakeMDReportBase(mock_file, results)
         assert report.generate_heading_name() == "Fake MD Report Base"