Repository: aristanetworks/anta (python)

Issue:
Handle OSError better for get-from-cvp
reported from the field

```
           CRITICAL Uncaught Exception when running ANTA CLI       logger.py:116
                    Please reach out to the maintainer team or
                    open an issue on Github:
                    https://github.com/arista-netdevops-community/
                    anta.
                    PermissionError: [Errno 13] Permission denied:
                    'cvp_duts.yaml'
```

Pull request:
fix(anta.cli): Better error handling for OSError when writing inventory
# Description

Catch OSErrors when writing inv

Fixes #772 

# Checklist:

<!-- Delete not relevant items !-->

- [x] My code follows the style guidelines of this project
- [x] I have performed a self-review of my code
- [x] I have run pre-commit for code linting and typing (`pre-commit run`)
- [x] I have made corresponding changes to the documentation
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] New and existing unit tests pass locally with my changes (`tox -e testenv`)


Patch:
diff --git a/anta/cli/get/commands.py b/anta/cli/get/commands.py
index 2bdd9cb9..3cc91261 100644
--- a/anta/cli/get/commands.py
+++ b/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)
 
diff --git a/anta/cli/get/utils.py b/anta/cli/get/utils.py
index 6f5d7d0f..d21dc541 100644
--- a/anta/cli/get/utils.py
+++ b/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:


Test patch:
diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py
index b87c394f..bc1acbb8 100644
--- a/anta/tests/interfaces.py
+++ b/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 VerifyInterfaceIPv4(AntaTest):
 
             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]:
diff --git a/examples/tests.yaml b/examples/tests.yaml
index 58d34c42..9f1aea99 100644
--- a/examples/tests.yaml
+++ b/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.
diff --git a/tests/units/anta_tests/test_interfaces.py b/tests/units/anta_tests/test_interfaces.py
index 271683b0..f3b4ee0c 100644
--- a/tests/units/anta_tests/test_interfaces.py
+++ b/tests/units/anta_tests/test_interfaces.py
@@ -1969,8 +1969,8 @@ DATA: list[dict[str, Any]] = [
                 "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 @@ DATA: list[dict[str, Any]] = [
         ],
         "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"]},
             ]
         },
diff --git a/tests/units/cli/get/test_commands.py b/tests/units/cli/get/test_commands.py
index 8edfa73b..0e263f7d 100644
--- a/tests/units/cli/get/test_commands.py
+++ b/tests/units/cli/get/test_commands.py
@@ -114,6 +114,27 @@ def test_from_cvp(
     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"),
     [