fix(anta.cli): Better error handling for OSError when writing invento…

…ry (#976)

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:

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"),
     [