diff --git a/anta/catalog.py b/anta/catalog.py index 5cf6b5b12..c64011706 100644 --- a/anta/catalog.py +++ b/anta/catalog.py @@ -250,7 +250,7 @@ def from_dict(data: RawCatalogInput, filename: str | Path | None = None) -> Anta try: catalog_data = AntaCatalogFile(**data) # type: ignore[arg-type] except ValidationError as e: - anta_log_exception(e, f"Test catalog{f' (from {filename})' if filename is not None else ''} is invalid!", logger) + anta_log_exception(e, f"Test catalog is invalid!{f' (from {filename})' if filename is not None else ''}", logger) raise for t in catalog_data.root.values(): tests.extend(t) diff --git a/anta/cli/exec/utils.py b/anta/cli/exec/utils.py index 15322e207..d52967fa8 100644 --- a/anta/cli/exec/utils.py +++ b/anta/cli/exec/utils.py @@ -15,11 +15,11 @@ from typing import Literal from aioeapi import EapiCommandError +from httpx import ConnectError, HTTPError from anta.device import AntaDevice, AsyncEOSDevice from anta.inventory import AntaInventory from anta.models import AntaCommand -from anta.tools.misc import anta_log_exception EOS_SCHEDULED_TECH_SUPPORT = "/mnt/flash/schedule/tech-support" @@ -88,8 +88,7 @@ async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "tex res = await asyncio.gather(*coros, return_exceptions=True) for r in res: if isinstance(r, Exception): - message = "Error when collecting commands" - anta_log_exception(r, message, logger) + logger.error(f"Error when collecting commands: {str(r)}") async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, configure: bool, tags: list[str] | None = None, latest: int | None = None) -> None: @@ -141,7 +140,7 @@ async def collect(device: AntaDevice) -> None: ) logger.warning(f"Configuring 'aaa authorization exec default local' on device {device.name}") command = AntaCommand(command="show running-config | include aaa authorization exec default local", ofmt="text") - await device.session.cli(commands=commands) # type: ignore[attr-defined] + await device._session.cli(commands=commands) # pylint: disable=protected-access logger.info(f"Configured 'aaa authorization exec default local' on device {device.name}") else: logger.error(f"Unable to collect tech-support on {device.name}: configuration 'aaa authorization exec default local' is not present") @@ -151,12 +150,8 @@ async def collect(device: AntaDevice) -> None: await device.copy(sources=filenames, destination=outdir, direction="from") logger.info(f"Collected {len(filenames)} scheduled tech-support from {device.name}") - except EapiCommandError as e: - logger.error(f"Unable to collect tech-support on {device.name}: {e.errmsg}") - # In this case we want to catch all exceptions - except Exception as e: # pylint: disable=broad-except - message = f"Unable to collect tech-support on device {device.name}" - anta_log_exception(e, message, logger) + except (EapiCommandError, HTTPError, ConnectError) as e: + logger.error(f"Unable to collect tech-support on {device.name}: {str(e)}") logger.info("Connecting to devices...") await inv.connect_inventory() diff --git a/anta/cli/utils.py b/anta/cli/utils.py index 43a8ba564..98e19cc53 100644 --- a/anta/cli/utils.py +++ b/anta/cli/utils.py @@ -19,7 +19,7 @@ from anta.catalog import AntaCatalog from anta.inventory import AntaInventory -from anta.tools.misc import anta_log_exception +from anta.inventory.exceptions import InventoryIncorrectSchema, InventoryRootKeyError logger = logging.getLogger(__name__) @@ -55,7 +55,7 @@ def parse_inventory(ctx: click.Context, path: Path) -> AntaInventory: return AntaInventory() try: inventory = AntaInventory.parse( - inventory_file=str(path), + filename=str(path), username=ctx.params["username"], password=ctx.params["password"], enable=ctx.params["enable"], @@ -64,10 +64,8 @@ def parse_inventory(ctx: click.Context, path: Path) -> AntaInventory: insecure=ctx.params["insecure"], disable_cache=ctx.params["disable_cache"], ) - except Exception as e: # pylint: disable=broad-exception-caught - message = f"Unable to parse ANTA Inventory file '{path}'" - anta_log_exception(e, message, logger) - ctx.fail(message) + except (ValidationError, YAMLError, OSError, InventoryIncorrectSchema, InventoryRootKeyError): + ctx.exit(ExitCode.USAGE_ERROR) return inventory @@ -95,7 +93,7 @@ def parse_catalog(ctx: click.Context, param: Option, value: Path) -> AntaCatalog try: catalog: AntaCatalog = AntaCatalog.parse(value) except (ValidationError, YAMLError, OSError): - ctx.fail("Unable to load ANTA Test Catalog") + ctx.exit(ExitCode.USAGE_ERROR) return catalog diff --git a/anta/inventory/__init__.py b/anta/inventory/__init__.py index e33ae92b6..fe297d1f4 100644 --- a/anta/inventory/__init__.py +++ b/anta/inventory/__init__.py @@ -13,7 +13,7 @@ from typing import Any, Optional from pydantic import ValidationError -from yaml import safe_load +from yaml import YAMLError, safe_load from anta.device import AntaDevice, AsyncEOSDevice from anta.inventory.exceptions import InventoryIncorrectSchema, InventoryRootKeyError @@ -138,7 +138,7 @@ def _parse_ranges(inventory_input: AntaInventoryInput, inventory: AntaInventory, @staticmethod def parse( - inventory_file: str, + filename: str, username: str, password: str, enable: bool = False, @@ -153,7 +153,7 @@ def parse( The inventory devices are AsyncEOSDevice instances. Args: - inventory_file (str): Path to inventory YAML file where user has described his inputs + filename (str): Path to device inventory YAML file username (str): Username to use to connect to devices password (str): Password to use to connect to devices enable (bool): Whether or not the commands need to be run in enable mode towards the devices @@ -165,7 +165,6 @@ def parse( Raises: InventoryRootKeyError: Root key of inventory is missing. InventoryIncorrectSchema: Inventory file is not following AntaInventory Schema. - InventoryUnknownFormat: Output format is not supported. """ inventory = AntaInventory() @@ -180,18 +179,24 @@ def parse( } kwargs = {k: v for k, v in kwargs.items() if v is not None} - with open(inventory_file, "r", encoding="UTF-8") as file: - data = safe_load(file) + try: + with open(file=filename, mode="r", encoding="UTF-8") as file: + data = safe_load(file) + except (YAMLError, OSError) as e: + message = f"Unable to parse ANTA Device Inventory file '{filename}'" + anta_log_exception(e, message, logger) + raise + + if AntaInventory.INVENTORY_ROOT_KEY not in data: + exc = InventoryRootKeyError(f"Inventory root key ({AntaInventory.INVENTORY_ROOT_KEY}) is not defined in your inventory") + anta_log_exception(exc, f"Device inventory is invalid! (from {filename})", logger) + raise exc - # Load data using Pydantic try: inventory_input = AntaInventoryInput(**data[AntaInventory.INVENTORY_ROOT_KEY]) - except KeyError as exc: - logger.error(f"Inventory root key is missing: {AntaInventory.INVENTORY_ROOT_KEY}") - raise InventoryRootKeyError(f"Inventory root key ({AntaInventory.INVENTORY_ROOT_KEY}) is not defined in your inventory") from exc - except ValidationError as exc: - logger.error("Inventory data are not compliant with inventory models") - raise InventoryIncorrectSchema(f"Inventory is not following the schema: {str(exc)}") from exc + except ValidationError as e: + anta_log_exception(e, f"Device inventory is invalid! (from {filename})", logger) + raise # Read data from input AntaInventory._parse_hosts(inventory_input, inventory, **kwargs) diff --git a/anta/result_manager/__init__.py b/anta/result_manager/__init__.py index 883d6f6c6..4cd22a266 100644 --- a/anta/result_manager/__init__.py +++ b/anta/result_manager/__init__.py @@ -26,7 +26,7 @@ class ResultManager: Create Inventory: inventory_anta = AntaInventory.parse( - inventory_file='examples/inventory.yml', + filename='examples/inventory.yml', username='ansible', password='ansible', timeout=0.5 diff --git a/docs/advanced_usages/as-python-lib.md b/docs/advanced_usages/as-python-lib.md index de4f1eb2e..04c5d82ae 100644 --- a/docs/advanced_usages/as-python-lib.md +++ b/docs/advanced_usages/as-python-lib.md @@ -69,7 +69,7 @@ async def main(inv: AntaInventory) -> None: if __name__ == "__main__": # Create the AntaInventory instance inventory = AntaInventory.parse( - inventory_file="inv.yml", + filename="inv.yml", username="arista", password="@rista123", timeout=15, @@ -126,7 +126,7 @@ async def main(inv: AntaInventory, commands: list[str]) -> dict[str, list[AntaCo if __name__ == "__main__": # Create the AntaInventory instance inventory = AntaInventory.parse( - inventory_file="inv.yml", + filename="inv.yml", username="arista", password="@rista123", timeout=15, diff --git a/tests/lib/fixture.py b/tests/lib/fixture.py index 929568a80..234f922e6 100644 --- a/tests/lib/fixture.py +++ b/tests/lib/fixture.py @@ -128,7 +128,7 @@ def test_inventory() -> AntaInventory: """ env = default_anta_env() return AntaInventory.parse( - inventory_file=env["ANTA_INVENTORY"], + filename=env["ANTA_INVENTORY"], username=env["ANTA_USERNAME"], password=env["ANTA_PASSWORD"], ) diff --git a/tests/units/cli/check/test_commands.py b/tests/units/cli/check/test_commands.py index 6cb316a5a..125a319d2 100644 --- a/tests/units/cli/check/test_commands.py +++ b/tests/units/cli/check/test_commands.py @@ -25,7 +25,7 @@ "catalog_path, expected_exit, expected_output", [ pytest.param("ghost_catalog.yml", 2, "Error: Invalid value for '--catalog'", id="catalog does not exist"), - pytest.param("test_catalog_with_undefined_module.yml", 2, "Unable to load ANTA Test Catalog", id="catalog is not valid"), + pytest.param("test_catalog_with_undefined_module.yml", 4, "Test catalog is invalid!", id="catalog is not valid"), pytest.param("test_catalog.yml", 0, f"Catalog {DATA_DIR}/test_catalog.yml is valid", id="catalog valid"), ], ) diff --git a/tests/units/inventory/test_inventory.py b/tests/units/inventory/test_inventory.py index 0fcc80257..6bb5453dc 100644 --- a/tests/units/inventory/test_inventory.py +++ b/tests/units/inventory/test_inventory.py @@ -53,7 +53,7 @@ def test_init_valid(self, test_definition: dict[str, Any], tmp_path: Path) -> No """ inventory_file = self.create_inventory(content=test_definition["input"], tmp_path=tmp_path) try: - AntaInventory.parse(inventory_file=inventory_file, username="arista", password="arista123") + AntaInventory.parse(filename=inventory_file, username="arista", password="arista123") except ValidationError as exc: logging.error("Exceptions is: %s", str(exc)) assert False @@ -77,11 +77,5 @@ def test_init_invalid(self, test_definition: dict[str, Any], tmp_path: Path) -> """ inventory_file = self.create_inventory(content=test_definition["input"], tmp_path=tmp_path) - try: - AntaInventory.parse(inventory_file=inventory_file, username="arista", password="arista123") - except InventoryIncorrectSchema as exc: - logging.warning("Exception is: %s", exc) - except InventoryRootKeyError as exc: - logging.warning("Exception is: %s", exc) - else: - assert False + with pytest.raises((InventoryIncorrectSchema, InventoryRootKeyError, ValidationError)): + AntaInventory.parse(filename=inventory_file, username="arista", password="arista123") diff --git a/tests/units/inventory/test_models.py b/tests/units/inventory/test_models.py index 912294133..7d8ab7d42 100644 --- a/tests/units/inventory/test_models.py +++ b/tests/units/inventory/test_models.py @@ -293,7 +293,6 @@ def test_anta_inventory_intput_invalid(self, inventory_def: dict[str, Any]) -> N } """ - # inventory_file = self.create_inventory(content=inventory_def['input'], tmp_path=tmp_path) try: if "hosts" in inventory_def["input"].keys(): logging.info(