narrow broad excepts

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)

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()

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
 
 

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)

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

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,

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

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

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

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(