diff --git a/anta/cli/get/commands.py b/anta/cli/get/commands.py index 2bdd9cb9f..3cc912610 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 6f5d7d0ff..d21dc5411 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: diff --git a/tests/units/cli/get/test_commands.py b/tests/units/cli/get/test_commands.py index 8edfa73b5..0e263f7d2 100644 --- a/tests/units/cli/get/test_commands.py +++ b/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"), [