From 7fec0636328410bfabf6c282cbd67073b117bf0f Mon Sep 17 00:00:00 2001 From: gmuloc Date: Mon, 9 Oct 2023 15:38:07 +0200 Subject: [PATCH] WIP --- anta/catalog.py | 146 +++++++++++++++++++++++++++++++++---- anta/cli/__init__.py | 6 +- anta/cli/check/commands.py | 7 +- anta/cli/utils.py | 20 ++--- anta/loader.py | 88 +--------------------- anta/runner.py | 9 ++- 6 files changed, 153 insertions(+), 123 deletions(-) diff --git a/anta/catalog.py b/anta/catalog.py index ea5926c18..ccb3c7fb0 100644 --- a/anta/catalog.py +++ b/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: + : + - : + + + 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 diff --git a/anta/cli/__init__.py b/anta/cli/__init__.py index 8d486c95a..c26866462 100644 --- a/anta/cli/__init__.py +++ b/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() diff --git a/anta/cli/check/commands.py b/anta/cli/check/commands.py index 070699f00..f542658e3 100644 --- a/anta/cli/check/commands.py +++ b/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 diff --git a/anta/cli/utils.py b/anta/cli/utils.py index 97bb033c4..98d6c3302 100644 --- a/anta/cli/utils.py +++ b/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,11 +88,12 @@ 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: @@ -103,9 +101,7 @@ def parse_catalog(ctx: click.Context, param: Option, value: str) -> list[tuple[A 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: diff --git a/anta/loader.py b/anta/loader.py index 95d4e9347..da8b75abd 100644 --- a/anta/loader.py +++ b/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: - : - - : - - - 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 diff --git a/anta/runner.py b/anta/runner.py index 31a852ac2..687422472 100644 --- a/anta/runner.py +++ b/anta/runner.py @@ -11,6 +11,7 @@ import logging from typing import Optional +from anta.catalog import AntaCatalog from anta.inventory import AntaInventory from anta.models import AntaTest from anta.result_manager import ResultManager @@ -22,7 +23,7 @@ async def main( manager: ResultManager, inventory: AntaInventory, - tests: list[tuple[type[AntaTest], AntaTest.Input]], + catalog: AntaCatalog, tags: Optional[list[str]] = None, established_only: bool = True, ) -> None: @@ -33,7 +34,7 @@ async def main( Args: manager: ResultManager object to populate with the test results. inventory: AntaInventory object that includes the device(s). - tests: ANTA test catalog. Output of anta.loader.parse_catalog(). + catalog: AntaCatalog object that includes the list of tests. tags: List of tags to filter devices from the inventory. Defaults to None. established_only: Include only established device(s). Defaults to True. @@ -41,7 +42,7 @@ async def main( any: ResultManager object gets updated with the test results. """ - if not tests: + if not catalog.tests: logger.info("The list of tests is empty, exiting") return @@ -64,7 +65,7 @@ async def main( coros = [] - for device, test in itertools.product(devices, tests): + for device, test in itertools.product(devices, catalog.tests): test_class = test[0] test_inputs = test[1] try: