WIP

anta/catalog.py

@@ -6,7 +6,11 @@
 """
 from __future__ import annotations
 
+import importlib
 import logging
+from typing import Any, Optional, cast
+
+from yaml import safe_load
 
 from anta.device import AsyncEOSDevice
 from anta.models import AntaTest
@@ -15,18 +19,134 @@
 logger = logging.getLogger(__name__)
 
 
-def is_catalog_valid(catalog: list[tuple[AntaTest, AntaTest.Input]]) -> ResultManager:
+class AntaCatalog:
     """
-    TODO - for now a test requires a device but this may be revisited in the future
+    Class representing an ANTA Catalog
+
+    Attributes:
+        name: Catalog name
+        filename Optional[str]: The path from which the catalog was loaded
+        tests: list[tuple[AntaTest, AntaTest.Input]]: A list of Tuple containing an AntaTest and the associated input
     """
-    # Mock device
-    mock_device = AsyncEOSDevice(name="mock", host="127.0.0.1", username="mock", password="mock")
-
-    manager = ResultManager()
-    # Instantiate each test to verify the Inputs are correct
-    for test_class, test_inputs in catalog:
-        # TODO - this is the same code with typing as in runner.py but somehow mypy complains that test_class
-        # ot type AntaTest is not callable
-        test_instance = test_class(device=mock_device, inputs=test_inputs)  # type: ignore[operator]
-        manager.add_test_result(test_instance.result)
-    return manager
+
+    def __init__(self, name: str, filename: Optional[str] = None) -> None:
+        """
+        Constructor of AntaCatalog
+
+        Args:
+            name: Device name
+            filname: Optional name - if provided tests are loaded
+        """
+        self.name: str = name
+        self.filename: Optional[str] = filename
+        self.tests: list[tuple[AntaTest, AntaTest.Input]] = []
+
+    def parse_catalog_file(self: AntaCatalog) -> None:
+        """
+        Parse a file
+        """
+        if self.filename is None:
+            return
+        try:
+            with open(self.filename, "r", encoding="UTF-8") as file:
+                data = safe_load(file)
+                self.parse_catalog(data)
+        # pylint: disable-next=broad-exception-caught
+        except Exception:
+            logger.critical(f"Something went wrong while parsing {self.filename}")
+            raise
+
+    def parse_catalog(self: AntaCatalog, test_catalog: dict[str, Any], package: str | None = None) -> None:
+        """
+        Function to parse the catalog and return a list of tests with their inputs
+
+        A valid test catalog must follow the following structure:
+            <Python module>:
+                - <AntaTest subclass>:
+                    <AntaTest.Input compliant dictionary>
+
+        Example:
+            anta.tests.connectivity:
+                - VerifyReachability:
+                    hosts:
+                        - dst: 8.8.8.8
+                          src: 172.16.0.1
+                        - dst: 1.1.1.1
+                          src: 172.16.0.1
+                    result_overwrite:
+                        categories:
+                            - "Overwritten category 1"
+                        description: "Test with overwritten description"
+                        custom_field: "Test run by John Doe"
+
+        Also supports nesting for Python module definition:
+            anta.tests:
+                connectivity:
+                    - VerifyReachability:
+                        hosts:
+                            - dst: 8.8.8.8
+                              src: 172.16.0.1
+                            - dst: 1.1.1.1
+                              src: 172.16.0.1
+                        result_overwrite:
+                            categories:
+                                - "Overwritten category 1"
+                            description: "Test with overwritten description"
+                            custom_field: "Test run by John Doe"
+
+        Args:
+            test_catalog: Python dictionary representing the test catalog YAML file
+
+        """
+        # pylint: disable=broad-exception-raised
+        if not test_catalog:
+            return
+
+        for key, value in test_catalog.items():
+            # Required to manage iteration within a tests module
+            if package is not None:
+                key = ".".join([package, key])
+            try:
+                module = importlib.import_module(f"{key}")
+            except ModuleNotFoundError:
+                logger.critical(f"No test module named '{key}'")
+                raise
+
+            if isinstance(value, list):
+                # This is a list of tests
+                for test in value:
+                    for test_name, inputs in test.items():
+                        # A test must be a subclass of AntaTest as defined in the Python module
+                        try:
+                            test = getattr(module, test_name)
+                        except AttributeError:
+                            logger.critical(f"Wrong test name '{test_name}' in '{module.__name__}'")
+                            raise
+                        if not issubclass(test, AntaTest):
+                            logger.critical(f"'{test.__module__}.{test.__name__}' is not an AntaTest subclass")
+                            raise Exception()
+                        # Test inputs can be either None or a dictionary
+                        if inputs is None or isinstance(inputs, dict):
+                            self.tests.append((cast(AntaTest, test), cast(AntaTest.Input, inputs)))
+                        else:
+                            logger.critical(f"'{test.__module__}.{test.__name__}' inputs must be a dictionary")
+                            raise Exception()
+            if isinstance(value, dict):
+                # This is an inner Python module
+                self.parse_catalog(value, package=module.__name__)
+
+    def check(self: AntaCatalog) -> ResultManager:
+        """
+        TODO - for now a test requires a device but this may be revisited in the future
+        """
+        # Mock device
+        mock_device = AsyncEOSDevice(name="mock", host="127.0.0.1", username="mock", password="mock")
+
+        manager = ResultManager()
+        # Instantiate each test to verify the Inputs are correct
+        for test_class, test_inputs in self.tests:
+            # TODO - this is the same code with typing as in runner.py but somehow mypy complains that test_class
+            # ot type AntaTest is not callable
+            test_instance = test_class(device=mock_device, inputs=test_inputs)  # type: ignore[operator]
+            manager.add_test_result(test_instance.result)
+        return manager

anta/cli/__init__.py

@@ -10,11 +10,12 @@
 
 import logging
 import pathlib
-from typing import Any, Callable, Literal
+from typing import Any, Literal
 
 import click
 
 from anta import __version__
+from anta.catalog import AntaCatalog
 from anta.cli.check import commands as check_commands
 from anta.cli.debug import commands as debug_commands
 from anta.cli.exec import commands as exec_commands
@@ -23,7 +24,6 @@
 from anta.cli.utils import AliasedGroup, IgnoreRequiredWithHelp, parse_catalog, parse_inventory
 from anta.loader import setup_logging
 from anta.result_manager import ResultManager
-from anta.result_manager.models import TestResult
 
 
 @click.group(cls=IgnoreRequiredWithHelp)
@@ -148,7 +148,7 @@ def anta(
     required=True,
     callback=parse_catalog,
 )
-def _nrfu(ctx: click.Context, catalog: list[tuple[Callable[..., TestResult], dict[Any, Any]]]) -> None:
+def _nrfu(ctx: click.Context, catalog: AntaCatalog) -> None:
     """Run NRFU against inventory devices"""
     ctx.obj["catalog"] = catalog
     ctx.obj["result_manager"] = ResultManager()

anta/cli/check/commands.py

@@ -12,10 +12,9 @@
 
 import click
 
-from anta.catalog import is_catalog_valid
+from anta.catalog import AntaCatalog
 from anta.cli.console import console
 from anta.cli.utils import parse_catalog
-from anta.models import AntaTest
 
 logger = logging.getLogger(__name__)
 
@@ -31,12 +30,12 @@
     required=True,
     callback=parse_catalog,
 )
-def catalog(ctx: click.Context, catalog: list[tuple[AntaTest, AntaTest.Input]]) -> None:
+def catalog(ctx: click.Context, catalog: AntaCatalog) -> None:
     """
     Check that the catalog is valid
     """
     logger.info(f"Checking syntax of catalog {ctx.obj['catalog_path']}")
-    manager = is_catalog_valid(catalog)
+    manager = catalog.check()
     if manager.error_status:
         console.print(f"[bold][red]Catalog {ctx.obj['catalog_path']} is invalid")
         # TODO print nice report

anta/cli/utils.py

@@ -14,9 +14,8 @@
 from typing import TYPE_CHECKING, Any
 
 import click
-from yaml import safe_load
 
-import anta.loader
+from anta.catalog import AntaCatalog
 from anta.inventory import AntaInventory
 from anta.tools.misc import anta_log_exception
 
@@ -25,8 +24,6 @@
 if TYPE_CHECKING:
     from click import Option
 
-    from anta.models import AntaTest
-
 
 class ExitCode(enum.IntEnum):
     """
@@ -82,7 +79,7 @@ def parse_tags(ctx: click.Context, param: Option, value: str) -> list[str] | Non
     return None
 
 
-def parse_catalog(ctx: click.Context, param: Option, value: str) -> list[tuple[AntaTest, dict[str, Any] | None]]:
+def parse_catalog(ctx: click.Context, param: Option, value: str) -> AntaCatalog:
     # pylint: disable=unused-argument
     """
     Click option callback to parse an ANTA tests catalog YAML file
@@ -91,21 +88,20 @@ def parse_catalog(ctx: click.Context, param: Option, value: str) -> list[tuple[A
     """
     if ctx.obj.get("_anta_help"):
         # Currently looking for help for a subcommand so no
-        # need to parse the Catalog - return an empty list
-        return []
+        # need to parse the Catalog - return an empty catalog
+        return AntaCatalog("dummy")
+    # Storing catalog path
+    ctx.obj["catalog_path"] = value
     try:
-        with open(value, "r", encoding="UTF-8") as file:
-            data = safe_load(file)
+        catalog = AntaCatalog("cli", value)
     # TODO catch proper exception
     # pylint: disable-next=broad-exception-caught
     except Exception as e:
         message = f"Unable to parse ANTA Tests Catalog file '{value}'"
         anta_log_exception(e, message, logger)
         ctx.fail(message)
 
-    # Storing catalog path
-    ctx.obj["catalog_path"] = value
-    return anta.loader.parse_catalog(data)
+    return catalog
 
 
 def exit_with_code(ctx: click.Context) -> None:

anta/loader.py

@@ -6,16 +6,12 @@
 """
 from __future__ import annotations
 
-import importlib
 import logging
-import sys
 from pathlib import Path
-from typing import Any
 
 from rich.logging import RichHandler
 
 from anta import __DEBUG__
-from anta.models import AntaTest
 
 logger = logging.getLogger(__name__)
 
@@ -42,7 +38,7 @@ def setup_logging(level: str = logging.getLevelName(logging.INFO), file: Path |
     # Init root logger
     root = logging.getLogger()
     # In ANTA debug mode, level is overriden to DEBUG
-    loglevel = getattr(logging, level.upper()) if not __DEBUG__ else logging.DEBUG
+    loglevel = logging.DEBUG if __DEBUG__ else getattr(logging, level.upper())
     root.setLevel(loglevel)
     # Silence the logging of chatty Python modules when level is INFO
     if loglevel == logging.INFO:
@@ -73,85 +69,3 @@ def setup_logging(level: str = logging.getLevelName(logging.INFO), file: Path |
 
     if __DEBUG__:
         logger.debug("ANTA Debug Mode enabled")
-
-
-def parse_catalog(test_catalog: dict[str, Any], package: str | None = None) -> list[tuple[AntaTest, dict[str, Any] | None]]:
-    """
-    Function to parse the catalog and return a list of tests with their inputs
-
-    A valid test catalog must follow the following structure:
-        <Python module>:
-            - <AntaTest subclass>:
-                <AntaTest.Input compliant dictionary>
-
-    Example:
-        anta.tests.connectivity:
-            - VerifyReachability:
-                hosts:
-                    - dst: 8.8.8.8
-                      src: 172.16.0.1
-                    - dst: 1.1.1.1
-                      src: 172.16.0.1
-                result_overwrite:
-                    categories:
-                        - "Overwritten category 1"
-                    description: "Test with overwritten description"
-                    custom_field: "Test run by John Doe"
-
-    Also supports nesting for Python module definition:
-        anta.tests:
-            connectivity:
-                - VerifyReachability:
-                    hosts:
-                        - dst: 8.8.8.8
-                          src: 172.16.0.1
-                        - dst: 1.1.1.1
-                          src: 172.16.0.1
-                    result_overwrite:
-                        categories:
-                            - "Overwritten category 1"
-                        description: "Test with overwritten description"
-                        custom_field: "Test run by John Doe"
-
-    Args:
-        test_catalog: Python dictionary representing the test catalog YAML file
-
-    Returns:
-        tests: List of tuples (test, inputs) where test is a reference of an AntaTest subclass
-              and inputs is a dictionary
-    """
-    tests: list[tuple[AntaTest, dict[str, Any] | None]] = []
-    if not test_catalog:
-        return tests
-    for key, value in test_catalog.items():
-        # Required to manage iteration within a tests module
-        if package is not None:
-            key = ".".join([package, key])
-        try:
-            module = importlib.import_module(f"{key}")
-        except ModuleNotFoundError:
-            logger.critical(f"No test module named '{key}'")
-            sys.exit(1)
-        if isinstance(value, list):
-            # This is a list of tests
-            for test in value:
-                for test_name, inputs in test.items():
-                    # A test must be a subclass of AntaTest as defined in the Python module
-                    try:
-                        test = getattr(module, test_name)
-                    except AttributeError:
-                        logger.critical(f"Wrong test name '{test_name}' in '{module.__name__}'")
-                        sys.exit(1)
-                    if not issubclass(test, AntaTest):
-                        logger.critical(f"'{test.__module__}.{test.__name__}' is not an AntaTest subclass")
-                        sys.exit(1)
-                    # Test inputs can be either None or a dictionary
-                    if inputs is None or isinstance(inputs, dict):
-                        tests.append((test, inputs))
-                    else:
-                        logger.critical(f"'{test.__module__}.{test.__name__}' inputs must be a dictionary")
-                        sys.exit(1)
-        if isinstance(value, dict):
-            # This is an inner Python module
-            tests.extend(parse_catalog(value, package=module.__name__))
-    return tests