From f128bdcae1286c4fa4d38aee8a0952b87e83a015 Mon Sep 17 00:00:00 2001 From: gmuloc Date: Tue, 10 Sep 2024 16:30:09 +0200 Subject: [PATCH] Refactor: Address PR comments --- anta/catalog.py | 86 ++++++++++++++-------- anta/cli/get/utils.py | 27 ++++--- anta/cli/nrfu/utils.py | 6 +- anta/cli/utils.py | 3 +- anta/custom_types.py | 14 ++-- anta/decorators.py | 24 ++++-- anta/device.py | 126 +++++++++++++++++++++----------- anta/inventory/__init__.py | 84 ++++++++++++++------- anta/inventory/models.py | 37 +++++++--- anta/logger.py | 15 ++-- anta/models.py | 123 ++++++++++++++++++++----------- anta/reporter/__init__.py | 67 +++++++++++------ anta/reporter/csv_reporter.py | 39 ++++++---- anta/reporter/md_reporter.py | 36 +++++---- anta/result_manager/__init__.py | 41 +++++++---- anta/result_manager/models.py | 39 ++++++---- anta/runner.py | 63 ++++++++++------ anta/tests/flow_tracking.py | 24 ++++-- anta/tests/logging.py | 9 ++- anta/tests/routing/bgp.py | 81 ++++++++++++-------- anta/tests/routing/isis.py | 33 +++++---- anta/tests/routing/ospf.py | 18 +++-- anta/tools.py | 31 +++++--- 23 files changed, 669 insertions(+), 357 deletions(-) diff --git a/anta/catalog.py b/anta/catalog.py index 192e17bf1..f747edc33 100644 --- a/anta/catalog.py +++ b/anta/catalog.py @@ -39,8 +39,12 @@ class AntaTestDefinition(BaseModel): """Define a test with its associated inputs. - test: An AntaTest concrete subclass - inputs: The associated AntaTest.Input subclass instance + Attributes + ---------- + test + An AntaTest concrete subclass + inputs + The associated AntaTest.Input subclass instance """ model_config = ConfigDict(frozen=True) @@ -60,6 +64,7 @@ def serialize_model(self) -> dict[str, AntaTest.Input]: Returns ------- + dict A dictionary representing the model. """ return {self.test.__name__: self.inputs} @@ -132,14 +137,14 @@ def check_inputs(self) -> AntaTestDefinition: class AntaCatalogFile(RootModel[dict[ImportString[Any], list[AntaTestDefinition]]]): # pylint: disable=too-few-public-methods """Represents an ANTA Test Catalog File. - Example: + Example ------- - A valid test catalog file must have the following structure: - ``` - : - - : - - ``` + A valid test catalog file must have the following structure: + ``` + : + - : + + ``` """ @@ -149,16 +154,16 @@ class AntaCatalogFile(RootModel[dict[ImportString[Any], list[AntaTestDefinition] def flatten_modules(data: dict[str, Any], package: str | None = None) -> dict[ModuleType, list[Any]]: """Allow the user to provide a data structure with nested Python modules. - Example: + Example ------- - ``` - anta.tests.routing: - generic: - - - bgp: - - - ``` - `anta.tests.routing.generic` and `anta.tests.routing.bgp` are importable Python modules. + ``` + anta.tests.routing: + generic: + - + bgp: + - + ``` + `anta.tests.routing.generic` and `anta.tests.routing.bgp` are importable Python modules. """ modules: dict[ModuleType, list[Any]] = {} @@ -234,6 +239,7 @@ def yaml(self) -> str: Returns ------- + str The YAML representation string of this model. """ # TODO: Pydantic and YAML serialization/deserialization is not supported natively. @@ -247,6 +253,7 @@ def to_json(self) -> str: Returns ------- + str The JSON representation string of this model. """ return self.model_dump_json(serialize_as_any=True, exclude_unset=True, indent=2) @@ -267,8 +274,10 @@ def __init__( Parameters ---------- - tests: A list of AntaTestDefinition instances. - filename: The path from which the catalog is loaded. + tests + A list of AntaTestDefinition instances. + filename + The path from which the catalog is loaded. """ self._tests: list[AntaTestDefinition] = [] @@ -314,8 +323,10 @@ def parse(filename: str | Path, file_format: Literal["yaml", "json"] = "yaml") - Parameters ---------- - filename: Path to test catalog YAML or JSON fil - file_format: Format of the file, either 'yaml' or 'json' + filename + Path to test catalog YAML or JSON fil + file_format + Format of the file, either 'yaml' or 'json' """ if file_format not in ["yaml", "json"]: @@ -343,8 +354,9 @@ def from_dict(data: RawCatalogInput, filename: str | Path | None = None) -> Anta Parameters ---------- - data: Python dictionary used to instantiate the AntaCatalog instance - filename: value to be set as AntaCatalog instance attribute + data + Python dictionary used to instantiate the AntaCatalog instance + filename: value to be set as AntaCatalog instance attribute """ tests: list[AntaTestDefinition] = [] @@ -377,7 +389,8 @@ def from_list(data: ListAntaTestTuples) -> AntaCatalog: Parameters ---------- - data: Python list used to instantiate the AntaCatalog instance + data + Python list used to instantiate the AntaCatalog instance """ tests: list[AntaTestDefinition] = [] @@ -394,10 +407,12 @@ def merge_catalogs(cls, catalogs: list[AntaCatalog]) -> AntaCatalog: Parameters ---------- - catalogs: A list of AntaCatalog instances to merge. + catalogs + A list of AntaCatalog instances to merge. Returns ------- + AntaCatalog A new AntaCatalog instance containing the tests of all the input catalogs. """ combined_tests = list(chain(*(catalog.tests for catalog in catalogs))) @@ -408,10 +423,12 @@ def merge(self, catalog: AntaCatalog) -> AntaCatalog: Parameters ---------- - catalog: AntaCatalog instance to merge to this instance. + catalog + AntaCatalog instance to merge to this instance. Returns ------- + AntaCatalog A new AntaCatalog instance containing the tests of the two instances. """ # TODO: Use a decorator to deprecate this method instead. See https://github.com/aristanetworks/anta/issues/754 @@ -427,6 +444,7 @@ def dump(self) -> AntaCatalogFile: Returns ------- + AntaCatalogFile An AntaCatalogFile instance containing tests of this AntaCatalog instance. """ root: dict[ImportString[Any], list[AntaTestDefinition]] = {} @@ -466,17 +484,21 @@ def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> set[Anta Parameters ---------- - tags: The tags to filter tests by. If empty, return all tests without tags. - strict: If True, returns only tests that contain all specified tags (intersection). - If False, returns tests that contain any of the specified tags (union). + tags + The tags to filter tests by. If empty, return all tests without tags. + strict + If True, returns only tests that contain all specified tags (intersection). + If False, returns tests that contain any of the specified tags (union). Returns ------- - set[AntaTestDefinition]: A set of tests that match the given tags. + set[AntaTestDefinition] + A set of tests that match the given tags. Raises ------ - ValueError: If the indexes have not been built prior to method call. + ValueError + If the indexes have not been built prior to method call. """ if not self.indexes_built: msg = "Indexes have not been built yet. Call build_indexes() first." diff --git a/anta/cli/get/utils.py b/anta/cli/get/utils.py index ba4d886d5..d84104e12 100644 --- a/anta/cli/get/utils.py +++ b/anta/cli/get/utils.py @@ -84,18 +84,24 @@ def get_cv_token(cvp_ip: str, cvp_username: str, cvp_password: str, *, verify_ce Parameters ---------- - cvp_ip: IP address of CloudVision. - cvp_username: Username to connect to CloudVision. - cvp_password: Password to connect to CloudVision. - verify_cert: Enable or disable certificate verification when connecting to CloudVision. + cvp_ip + IP address of CloudVision. + cvp_username + Username to connect to CloudVision. + cvp_password + Password to connect to CloudVision. + verify_cert + Enable or disable certificate verification when connecting to CloudVision. Returns ------- - token(str): The token to use in further API calls to CloudVision. + str + The token to use in further API calls to CloudVision. Raises ------ - requests.ssl.SSLError: If the certificate verification fails + requests.ssl.SSLError + If the certificate verification fails """ # use CVP REST API to generate a token @@ -163,9 +169,12 @@ def create_inventory_from_ansible(inventory: Path, output: Path, ansible_group: Parameters ---------- - inventory: Ansible Inventory file to read - output: ANTA inventory file to generate. - ansible_group: Ansible group from where to extract data. + inventory + Ansible Inventory file to read + output + ANTA inventory file to generate. + ansible_group + Ansible group from where to extract data. """ try: diff --git a/anta/cli/nrfu/utils.py b/anta/cli/nrfu/utils.py index 748578dec..947c08901 100644 --- a/anta/cli/nrfu/utils.py +++ b/anta/cli/nrfu/utils.py @@ -147,8 +147,10 @@ def save_markdown_report(ctx: click.Context, md_output: pathlib.Path) -> None: Parameters ---------- - ctx: Click context containing the result manager. - md_output: Path to save the markdown report. + ctx + Click context containing the result manager. + md_output + Path to save the markdown report. """ try: MDReportGenerator.generate(results=_get_result_manager(ctx), md_filename=md_output) diff --git a/anta/cli/utils.py b/anta/cli/utils.py index 2f6e7d302..bcb373aea 100644 --- a/anta/cli/utils.py +++ b/anta/cli/utils.py @@ -62,7 +62,8 @@ def exit_with_code(ctx: click.Context) -> None: Parameters ---------- - ctx: Click Context + ctx + Click Context """ if ctx.obj.get("ignore_status"): diff --git a/anta/custom_types.py b/anta/custom_types.py index 322fa4aca..6747e7663 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -66,9 +66,9 @@ def interface_case_sensitivity(v: str) -> str: Examples -------- - - ethernet -> Ethernet - - vlan -> Vlan - - loopback -> Loopback + - ethernet -> Ethernet + - vlan -> Vlan + - loopback -> Loopback """ if isinstance(v, str) and v != "" and not v[0].isupper(): @@ -81,10 +81,10 @@ def bgp_multiprotocol_capabilities_abbreviations(value: str) -> str: Examples -------- - - IPv4 Unicast - - L2vpnEVPN - - ipv4 MPLS Labels - - ipv4Mplsvpn + - IPv4 Unicast + - L2vpnEVPN + - ipv4 MPLS Labels + - ipv4Mplsvpn """ patterns = { diff --git a/anta/decorators.py b/anta/decorators.py index c9f8b6d28..f5608ef26 100644 --- a/anta/decorators.py +++ b/anta/decorators.py @@ -22,11 +22,13 @@ def deprecated_test(new_tests: list[str] | None = None) -> Callable[[F], F]: Parameters ---------- - new_tests: A list of new test classes that should replace the deprecated test. + new_tests + A list of new test classes that should replace the deprecated test. Returns ------- - Callable[[F], F]: A decorator that can be used to wrap test functions. + Callable[[F], F] + A decorator that can be used to wrap test functions. """ @@ -35,11 +37,13 @@ def decorator(function: F) -> F: Parameters ---------- - function: The test function to be decorated. + function + The test function to be decorated. Returns ------- - F: The decorated function. + F + The decorated function. """ @@ -66,11 +70,13 @@ def skip_on_platforms(platforms: list[str]) -> Callable[[F], F]: Parameters ---------- - platforms: List of hardware models on which the test should be skipped. + platforms + List of hardware models on which the test should be skipped. Returns ------- - Callable[[F], F]: A decorator that can be used to wrap test functions. + Callable[[F], F] + A decorator that can be used to wrap test functions. """ @@ -79,11 +85,13 @@ def decorator(function: F) -> F: Parameters ---------- - function: The test function to be decorated. + function + The test function to be decorated. Returns ------- - F: The decorated function. + F + The decorated function. """ diff --git a/anta/device.py b/anta/device.py index 5a30de9f2..3e59f95eb 100644 --- a/anta/device.py +++ b/anta/device.py @@ -42,13 +42,20 @@ class AntaDevice(ABC): Attributes ---------- - name: Device name - is_online: True if the device IP is reachable and a port can be open. - established: True if remote command execution succeeds. - hw_model: Hardware model of the device. - tags: Tags for this device. - cache: In-memory cache from aiocache library for this device (None if cache is disabled). - cache_locks: Dictionary mapping keys to asyncio locks to guarantee exclusive access to the cache if not disabled. + name : str + Device name + is_online : bool + True if the device IP is reachable and a port can be open. + established : bool + True if remote command execution succeeds. + hw_model : str + Hardware model of the device. + tags : set[str] + Tags for this device. + cache : Cache | None + In-memory cache from aiocache library for this device (None if cache is disabled). + cache_locks : dict + Dictionary mapping keys to asyncio locks to guarantee exclusive access to the cache if not disabled. """ @@ -57,9 +64,12 @@ def __init__(self, name: str, tags: set[str] | None = None, *, disable_cache: bo Parameters ---------- - name: Device name. - tags: Tags for this device. - disable_cache: Disable caching for all commands for this device. + name + Device name. + tags + Tags for this device. + disable_cache + Disable caching for all commands for this device. """ self.name: str = name @@ -132,8 +142,10 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No Parameters ---------- - command: The command to collect. - collection_id: An identifier used to build the eAPI request ID. + command + The command to collect. + collection_id + An identifier used to build the eAPI request ID. """ async def collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None: @@ -149,8 +161,10 @@ async def collect(self, command: AntaCommand, *, collection_id: str | None = Non Parameters ---------- - command: The command to collect. - collection_id: An identifier used to build the eAPI request ID. + command + The command to collect. + collection_id + An identifier used to build the eAPI request ID. """ # Need to ignore pylint no-member as Cache is a proxy class and pylint is not smart enough # https://github.com/pylint-dev/pylint/issues/7258 @@ -172,8 +186,10 @@ async def collect_commands(self, commands: list[AntaCommand], *, collection_id: Parameters ---------- - commands: The commands to collect. - collection_id: An identifier used to build the eAPI request ID. + commands + The commands to collect. + collection_id + An identifier used to build the eAPI request ID. """ await asyncio.gather(*(self.collect(command=command, collection_id=collection_id) for command in commands)) @@ -194,9 +210,12 @@ async def copy(self, sources: list[Path], destination: Path, direction: Literal[ Parameters ---------- - sources: List of files to copy to or from the device. - destination: Local or remote destination when copying the files. Can be a folder. - direction: Defines if this coroutine copies files to or from the device. + sources + List of files to copy to or from the device. + destination + Local or remote destination when copying the files. Can be a folder. + direction + Defines if this coroutine copies files to or from the device. """ _ = (sources, destination, direction) @@ -209,11 +228,16 @@ class AsyncEOSDevice(AntaDevice): Attributes ---------- - name: Device name - is_online: True if the device IP is reachable and a port can be open - established: True if remote command execution succeeds - hw_model: Hardware model of the device - tags: Tags for this device + name : str + Device name + is_online : bool + True if the device IP is reachable and a port can be open + established : bool + True if remote command execution succeeds + hw_model : str + Hardware model of the device + tags : set[str] + Tags for this device """ @@ -239,19 +263,32 @@ def __init__( Parameters ---------- - host: Device FQDN or IP. - username: Username to connect to eAPI and SSH. - password: Password to connect to eAPI and SSH. - name: Device name. - enable: Collect commands using privileged mode. - enable_password: Password used to gain privileged access on EOS. - port: eAPI port. Defaults to 80 is proto is 'http' or 443 if proto is 'https'. - ssh_port: SSH port. - tags: Tags for this device. - timeout: Timeout value in seconds for outgoing API calls. - insecure: Disable SSH Host Key validation. - proto: eAPI protocol. Value can be 'http' or 'https'. - disable_cache: Disable caching for all commands for this device. + host + Device FQDN or IP. + username + Username to connect to eAPI and SSH. + password + Password to connect to eAPI and SSH. + name + Device name. + enable + Collect commands using privileged mode. + enable_password + Password used to gain privileged access on EOS. + port + eAPI port. Defaults to 80 is proto is 'http' or 443 if proto is 'https'. + ssh_port + SSH port. + tags + Tags for this device. + timeout + Timeout value in seconds for outgoing API calls. + insecure + Disable SSH Host Key validation. + proto + eAPI protocol. Value can be 'http' or 'https'. + disable_cache + Disable caching for all commands for this device. """ if host is None: @@ -315,8 +352,10 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No Parameters ---------- - command: The command to collect. - collection_id: An identifier used to build the eAPI request ID. + command + The command to collect. + collection_id + An identifier used to build the eAPI request ID. """ commands: list[dict[str, str | int]] = [] if self.enable and self._enable_password is not None: @@ -407,9 +446,12 @@ async def copy(self, sources: list[Path], destination: Path, direction: Literal[ Parameters ---------- - sources: List of files to copy to or from the device. - destination: Local or remote destination when copying the files. Can be a folder. - direction: Defines if this coroutine copies files to or from the device. + sources + List of files to copy to or from the device. + destination + Local or remote destination when copying the files. Can be a folder. + direction + Defines if this coroutine copies files to or from the device. """ async with asyncssh.connect( diff --git a/anta/inventory/__init__.py b/anta/inventory/__init__.py index 4aa4f6655..e75b16e30 100644 --- a/anta/inventory/__init__.py +++ b/anta/inventory/__init__.py @@ -46,8 +46,10 @@ def _update_disable_cache(kwargs: dict[str, Any], *, inventory_disable_cache: bo Parameters ---------- - inventory_disable_cache: The value of disable_cache in the inventory - kwargs: The kwargs to instantiate the device + inventory_disable_cache + The value of disable_cache in the inventory + kwargs + The kwargs to instantiate the device """ updated_kwargs = kwargs.copy() @@ -64,9 +66,12 @@ def _parse_hosts( Parameters ---------- - inventory_input: AntaInventoryInput used to parse the devices - inventory: AntaInventory to add the parsed devices to - **kwargs: Additional keyword arguments to pass to the device constructor + inventory_input + AntaInventoryInput used to parse the devices + inventory + AntaInventory to add the parsed devices to + **kwargs + Additional keyword arguments to pass to the device constructor """ if inventory_input.hosts is None: @@ -93,13 +98,17 @@ def _parse_networks( Parameters ---------- - inventory_input: AntaInventoryInput used to parse the devices - inventory: AntaInventory to add the parsed devices to - **kwargs: Additional keyword arguments to pass to the device constructor + inventory_input + AntaInventoryInput used to parse the devices + inventory + AntaInventory to add the parsed devices to + **kwargs + Additional keyword arguments to pass to the device constructor Raises ------ - InventoryIncorrectSchemaError: Inventory file is not following AntaInventory Schema. + InventoryIncorrectSchemaError + Inventory file is not following AntaInventory Schema. """ if inventory_input.networks is None: @@ -126,13 +135,17 @@ def _parse_ranges( Parameters ---------- - inventory_input: AntaInventoryInput used to parse the devices - inventory: AntaInventory to add the parsed devices to - **kwargs: Additional keyword arguments to pass to the device constructor + inventory_input + AntaInventoryInput used to parse the devices + inventory + AntaInventory to add the parsed devices to + **kwargs + Additional keyword arguments to pass to the device constructor Raises ------ - InventoryIncorrectSchemaError: Inventory file is not following AntaInventory Schema. + InventoryIncorrectSchemaError + Inventory file is not following AntaInventory Schema. """ if inventory_input.ranges is None: @@ -177,19 +190,29 @@ def parse( Parameters ---------- - filename: Path to device inventory YAML file. - username: Username to use to connect to devices. - password: Password to use to connect to devices. - enable_password: Enable password to use if required. - timeout: Timeout value in seconds for outgoing API calls. - enable: Whether or not the commands need to be run in enable mode towards the devices. - insecure: Disable SSH Host Key validation. - disable_cache: Disable cache globally. + filename + Path to device inventory YAML file. + username + Username to use to connect to devices. + password + Password to use to connect to devices. + enable_password + Enable password to use if required. + timeout + Timeout value in seconds for outgoing API calls. + enable + Whether or not the commands need to be run in enable mode towards the devices. + insecure + Disable SSH Host Key validation. + disable_cache + Disable cache globally. Raises ------ - InventoryRootKeyError: Root key of inventory is missing. - InventoryIncorrectSchemaError: Inventory file is not following AntaInventory Schema. + InventoryRootKeyError + Root key of inventory is missing. + InventoryIncorrectSchemaError + Inventory file is not following AntaInventory Schema. """ inventory = AntaInventory() @@ -256,13 +279,17 @@ def get_inventory(self, *, established_only: bool = False, tags: set[str] | None Parameters ---------- - established_only: Whether or not to include only established devices. - tags: Tags to filter devices. - devices: Names to filter devices. + established_only + Whether or not to include only established devices. + tags + Tags to filter devices. + devices + Names to filter devices. Returns ------- - An inventory with filtered AntaDevice objects. + AntaInventory + An inventory with filtered AntaDevice objects. """ def _filter_devices(device: AntaDevice) -> bool: @@ -295,7 +322,8 @@ def add_device(self, device: AntaDevice) -> None: Parameters ---------- - device: Device object to be added + device + Device object to be added """ self[device.name] = device diff --git a/anta/inventory/models.py b/anta/inventory/models.py index 5796ef700..2eea701f6 100644 --- a/anta/inventory/models.py +++ b/anta/inventory/models.py @@ -21,11 +21,16 @@ class AntaInventoryHost(BaseModel): Attributes ---------- - host: IP Address or FQDN of the device. - port: Custom eAPI port to use. - name: Custom name of the device. - tags: Tags of the device. - disable_cache: Disable cache for this device. + host : Hostname | IPvAnyAddress + IP Address or FQDN of the device. + port : Port | None + Custom eAPI port to use. + name : str | None + Custom name of the device. + tags : set[str] + Tags of the device. + disable_cache : bool + Disable cache for this device. """ @@ -43,9 +48,12 @@ class AntaInventoryNetwork(BaseModel): Attributes ---------- - network: Subnet to use for scanning. - tags: Tags of the devices in this network. - disable_cache: Disable cache for all devices in this network. + network : IPvAnyNetwork + Subnet to use for scanning. + tags : set[str] + Tags of the devices in this network. + disable_cache : bool + Disable cache for all devices in this network. """ @@ -61,10 +69,14 @@ class AntaInventoryRange(BaseModel): Attributes ---------- - start: IPv4 or IPv6 address for the beginning of the range. - stop: IPv4 or IPv6 address for the end of the range. - tags: Tags of the devices in this IP range. - disable_cache: Disable cache for all devices in this IP range. + start : IPvAnyAddress + IPv4 or IPv6 address for the beginning of the range. + stop : IPvAnyAddress + IPv4 or IPv6 address for the end of the range. + tags : set[str] + Tags of the devices in this IP range. + disable_cache : bool + Disable cache for all devices in this IP range. """ @@ -90,6 +102,7 @@ def yaml(self) -> str: Returns ------- + str The YAML representation string of this model. """ # TODO: Pydantic and YAML serialization/deserialization is not supported natively. diff --git a/anta/logger.py b/anta/logger.py index b64fbe7b4..54733fb73 100644 --- a/anta/logger.py +++ b/anta/logger.py @@ -51,8 +51,10 @@ def setup_logging(level: LogLevel = Log.INFO, file: Path | None = None) -> None: Parameters ---------- - level: ANTA logging level - file: Send logs to a file + level + ANTA logging level + file + Send logs to a file """ # Init root logger @@ -106,9 +108,12 @@ def anta_log_exception(exception: BaseException, message: str | None = None, cal Parameters ---------- - exception: The Exception being logged. - message: An optional message. - calling_logger: A logger to which the exception should be logged. If not present, the logger in this file is used. + exception + The Exception being logged. + message + An optional message. + calling_logger + A logger to which the exception should be logged. If not present, the logger in this file is used. """ if calling_logger is None: diff --git a/anta/models.py b/anta/models.py index ea44205b8..ecc6d61f8 100644 --- a/anta/models.py +++ b/anta/models.py @@ -48,11 +48,16 @@ class AntaTemplate: Attributes ---------- - template: Python f-string. Example: 'show vlan {vlan_id}' - version: eAPI version - valid values are 1 or "latest". - revision: Revision of the command. Valid values are 1 to 99. Revision has precedence over version. - ofmt: eAPI output - json or text. - use_cache: Enable or disable caching for this AntaTemplate if the AntaDevice supports it. + template + Python f-string. Example: 'show vlan {vlan_id}' + version + eAPI version - valid values are 1 or "latest". + revision + Revision of the command. Valid values are 1 to 99. Revision has precedence over version. + ofmt + eAPI output - json or text. + use_cache + Enable or disable caching for this AntaTemplate if the AntaDevice supports it. """ # pylint: disable=too-few-public-methods @@ -97,18 +102,20 @@ def render(self, **params: str | int | bool) -> AntaCommand: Parameters ---------- - params: dictionary of variables with string values to render the Python f-string + params + dictionary of variables with string values to render the Python f-string Returns ------- + AntaCommand The rendered AntaCommand. This AntaCommand instance have a template attribute that references this AntaTemplate instance. Raises ------ - AntaTemplateRenderError - If a parameter is missing to render the AntaTemplate instance. + AntaTemplateRenderError + If a parameter is missing to render the AntaTemplate instance. """ try: command = self.template.format(**params) @@ -141,15 +148,24 @@ class AntaCommand(BaseModel): Attributes ---------- - command: Device command - version: eAPI version - valid values are 1 or "latest". - revision: eAPI revision of the command. Valid values are 1 to 99. Revision has precedence over version. - ofmt: eAPI output - json or text. - output: Output of the command. Only defined if there was no errors. - template: AntaTemplate object used to render this command. - errors: If the command execution fails, eAPI returns a list of strings detailing the error(s). - params: Pydantic Model containing the variables values used to render the template. - use_cache: Enable or disable caching for this AntaCommand if the AntaDevice supports it. + command + Device command + version + eAPI version - valid values are 1 or "latest". + revision + eAPI revision of the command. Valid values are 1 to 99. Revision has precedence over version. + ofmt + eAPI output - json or text. + output + Output of the command. Only defined if there was no errors. + template + AntaTemplate object used to render this command. + errors + If the command execution fails, eAPI returns a list of strings detailing the error(s). + params + Pydantic Model containing the variables values used to render the template. + use_cache + Enable or disable caching for this AntaCommand if the AntaDevice supports it. """ @@ -214,9 +230,9 @@ def requires_privileges(self) -> bool: Raises ------ - RuntimeError - If the command has not been collected and has not returned an error. - AntaDevice.collect() must be called before this property. + RuntimeError + If the command has not been collected and has not returned an error. + AntaDevice.collect() must be called before this property. """ if not self.collected and not self.error: msg = f"Command '{self.command}' has not been collected and has not returned an error. Call AntaDevice.collect()." @@ -229,9 +245,9 @@ def supported(self) -> bool: Raises ------ - RuntimeError - If the command has not been collected and has not returned an error. - AntaDevice.collect() must be called before this property. + RuntimeError + If the command has not been collected and has not returned an error. + AntaDevice.collect() must be called before this property. """ if not self.collected and not self.error: msg = f"Command '{self.command}' has not been collected and has not returned an error. Call AntaDevice.collect()." @@ -247,8 +263,10 @@ def __init__(self, template: AntaTemplate, key: str) -> None: Parameters ---------- - template: The AntaTemplate instance that failed to render - key: Key that has not been provided to render the template + template + The AntaTemplate instance that failed to render + key + Key that has not been provided to render the template """ self.template = template @@ -297,11 +315,16 @@ def test(self) -> None: Attributes ---------- - device: AntaDevice instance on which this test is run - inputs: AntaTest.Input instance carrying the test inputs - instance_commands: List of AntaCommand instances of this test - result: TestResult instance representing the result of this test - logger: Python logger for this test instance + device + AntaDevice instance on which this test is run + inputs + AntaTest.Input instance carrying the test inputs + instance_commands + List of AntaCommand instances of this test + result + TestResult instance representing the result of this test + logger + Python logger for this test instance """ # Mandatory class attributes @@ -332,7 +355,8 @@ class Input(BaseModel): Attributes ---------- - result_overwrite: Define fields to overwrite in the TestResult object + result_overwrite + Define fields to overwrite in the TestResult object """ model_config = ConfigDict(extra="forbid") @@ -351,9 +375,12 @@ class ResultOverwrite(BaseModel): Attributes ---------- - description: overwrite TestResult.description - categories: overwrite TestResult.categories - custom_field: a free string that will be included in the TestResult object + description + overwrite TestResult.description + categories + overwrite TestResult.categories + custom_field + a free string that will be included in the TestResult object """ @@ -367,7 +394,8 @@ class Filters(BaseModel): Attributes ---------- - tags: Tag of devices on which to run the test. + tags + Tag of devices on which to run the test. """ model_config = ConfigDict(extra="forbid") @@ -383,10 +411,13 @@ def __init__( Parameters ---------- - device: AntaDevice instance on which the test will be run - inputs: dictionary of attributes used to instantiate the AntaTest.Input instance - eos_data: Populate outputs of the test commands instead of collecting from devices. - This list must have the same length and order than the `instance_commands` instance attribute. + device + AntaDevice instance on which the test will be run + inputs + dictionary of attributes used to instantiate the AntaTest.Input instance + eos_data + Populate outputs of the test commands instead of collecting from devices. + This list must have the same length and order than the `instance_commands` instance attribute. """ self.logger: logging.Logger = logging.getLogger(f"{self.module}.{self.__class__.__name__}") self.device: AntaDevice = device @@ -558,14 +589,18 @@ async def wrapper( Parameters ---------- - self: The test instance. - eos_data: Populate outputs of the test commands instead of collecting from devices. - This list must have the same length and order than the `instance_commands` instance attribute. - kwargs: Any keyword argument to pass to the test. + self + The test instance. + eos_data + Populate outputs of the test commands instead of collecting from devices. + This list must have the same length and order than the `instance_commands` instance attribute. + kwargs + Any keyword argument to pass to the test. Returns ------- - result: TestResult instance attribute populated with error status if any + TestResult + The TestResult instance attribute populated with error status if any """ if self.result.result != "unset": diff --git a/anta/reporter/__init__.py b/anta/reporter/__init__.py index b5bc381c0..ae71a4430 100644 --- a/anta/reporter/__init__.py +++ b/anta/reporter/__init__.py @@ -45,12 +45,15 @@ def _split_list_to_txt_list(self, usr_list: list[str], delimiter: str | None = N Parameters ---------- - usr_list (list[str]): List of string to concatenate - delimiter (str, optional): A delimiter to use to start string. Defaults to None. + usr_list : list[str] + List of string to concatenate + delimiter : str, optional + A delimiter to use to start string. Defaults to None. Returns ------- - str: Multi-lines string + str + Multi-lines string """ if delimiter is not None: @@ -64,12 +67,15 @@ def _build_headers(self, headers: list[str], table: Table) -> Table: Parameters ---------- - headers: List of headers. - table: A rich Table instance. + headers + List of headers. + table + A rich Table instance. Returns ------- - A rich `Table` instance with headers. + Table + A rich `Table` instance with headers. """ for idx, header in enumerate(headers): @@ -101,12 +107,15 @@ def report_all(self, manager: ResultManager, title: str = "All tests results") - Parameters ---------- - manager: A ResultManager instance. - title: Title for the report. Defaults to 'All tests results'. + manager + A ResultManager instance. + title + Title for the report. Defaults to 'All tests results'. Returns ------- - A fully populated rich `Table` + Table + A fully populated rich `Table` """ table = Table(title=title, show_lines=True) @@ -135,13 +144,17 @@ def report_summary_tests( Parameters ---------- - manager: A ResultManager instance. - tests: List of test names to include. None to select all tests. - title: Title of the report. + manager + A ResultManager instance. + tests + List of test names to include. None to select all tests. + title + Title of the report. Returns ------- - A fully populated rich `Table`. + Table + A fully populated rich `Table`. """ table = Table(title=title, show_lines=True) headers = [ @@ -177,13 +190,17 @@ def report_summary_devices( Parameters ---------- - manager: A ResultManager instance. - devices: List of device names to include. None to select all devices. - title: Title of the report. + manager + A ResultManager instance. + devices + List of device names to include. None to select all devices. + title + Title of the report. Returns ------- - A fully populated rich `Table`. + Table + A fully populated rich `Table`. """ table = Table(title=title, show_lines=True) headers = [ @@ -225,6 +242,9 @@ def render(self, data: list[dict[str, Any]], *, trim_blocks: bool = True, lstrip Report is built based on a J2 template provided by user. Data structure sent to template is: + Example + ------- + ``` >>> print(ResultManager.json) [ { @@ -236,16 +256,21 @@ def render(self, data: list[dict[str, Any]], *, trim_blocks: bool = True, lstrip description: ..., } ] + ``` Parameters ---------- - data: List of results from ResultManager.results - trim_blocks: enable trim_blocks for J2 rendering. - lstrip_blocks: enable lstrip_blocks for J2 rendering. + data + List of results from ResultManager.results + trim_blocks + enable trim_blocks for J2 rendering. + lstrip_blocks + enable lstrip_blocks for J2 rendering. Returns ------- - Rendered template + str + Rendered template """ with self.template_path.open(encoding="utf-8") as file_: diff --git a/anta/reporter/csv_reporter.py b/anta/reporter/csv_reporter.py index 221cbec81..dca02d48f 100644 --- a/anta/reporter/csv_reporter.py +++ b/anta/reporter/csv_reporter.py @@ -42,12 +42,15 @@ def split_list_to_txt_list(cls, usr_list: list[str], delimiter: str = " - ") -> Parameters ---------- - usr_list: List of string to concatenate - delimiter: A delimiter to use to start string. Defaults to None. + usr_list + List of string to concatenate + delimiter + A delimiter to use to start string. Defaults to None. Returns ------- - str: Multi-lines string + str + Multi-lines string """ return f"{delimiter}".join(f"{line}" for line in usr_list) @@ -57,9 +60,14 @@ def convert_to_list(cls, result: TestResult) -> list[str]: """ Convert a TestResult into a list of string for creating file content. - Args: - ---- - results: A TestResult to convert into list. + Parameters + ---------- + results + A TestResult to convert into list. + + Returns + ------- + list[str] """ message = cls.split_list_to_txt_list(result.messages) if len(result.messages) > 0 else "" categories = cls.split_list_to_txt_list(result.categories) if len(result.categories) > 0 else "None" @@ -76,14 +84,17 @@ def convert_to_list(cls, result: TestResult) -> list[str]: def generate(cls, results: ResultManager, csv_filename: pathlib.Path) -> None: """Build CSV flle with tests results. - Parameter - --------- - results: A ResultManager instance. - csv_filename: File path where to save CSV data. - - Raise - ----- - OSError if any is raised while writing the CSV file. + Parameters + ---------- + results + A ResultManager instance. + csv_filename + File path where to save CSV data. + + Raises + ------ + OSError + if any is raised while writing the CSV file. """ headers = [ cls.Headers.device, diff --git a/anta/reporter/md_reporter.py b/anta/reporter/md_reporter.py index 7b97fb176..f4eadb2b5 100644 --- a/anta/reporter/md_reporter.py +++ b/anta/reporter/md_reporter.py @@ -41,8 +41,10 @@ def generate(cls, results: ResultManager, md_filename: Path) -> None: Parameters ---------- - results: The ResultsManager instance containing all test results. - md_filename: The path to the markdown file to write the report into. + results + The ResultsManager instance containing all test results. + md_filename + The path to the markdown file to write the report into. """ try: with md_filename.open("w", encoding="utf-8") as mdfile: @@ -74,8 +76,10 @@ def __init__(self, mdfile: TextIOWrapper, results: ResultManager) -> None: Parameters ---------- - mdfile: An open file object to write the markdown data into. - results: The ResultsManager instance containing all test results. + mdfile + An open file object to write the markdown data into. + results + The ResultsManager instance containing all test results. """ self.mdfile = mdfile self.results = results @@ -102,12 +106,13 @@ def generate_heading_name(self) -> str: Returns ------- - str: Formatted header name. + str + Formatted header name. Example ------- - - `ANTAReport` will become ANTA Report. - - `TestResultsSummary` will become Test Results Summary. + - `ANTAReport` will become ANTA Report. + - `TestResultsSummary` will become Test Results Summary. """ class_name = self.__class__.__name__ @@ -124,8 +129,10 @@ def write_table(self, table_heading: list[str], *, last_table: bool = False) -> Parameters ---------- - table_heading: List of strings to join for the table heading. - last_table: Flag to determine if it's the last table of the markdown file to avoid unnecessary new line. Defaults to False. + table_heading + List of strings to join for the table heading. + last_table + Flag to determine if it's the last table of the markdown file to avoid unnecessary new line. Defaults to False. """ self.mdfile.write("\n".join(table_heading) + "\n") for row in self.generate_rows(): @@ -140,11 +147,12 @@ def write_heading(self, heading_level: int) -> None: Parameters ---------- - heading_level: The level of the heading (1-6). + heading_level + The level of the heading (1-6). Example ------- - ## Test Results Summary + ## Test Results Summary """ # Ensure the heading level is within the valid range of 1 to 6 heading_level = max(1, min(heading_level, 6)) @@ -157,11 +165,13 @@ def safe_markdown(self, text: str | None) -> str: Parameters ---------- - text: The text to escape markdown characters from. + text + The text to escape markdown characters from. Returns ------- - str: The text with escaped markdown characters. + str + The text with escaped markdown characters. """ # Custom field from a TestResult object can be None if text is None: diff --git a/anta/result_manager/__init__.py b/anta/result_manager/__init__.py index 9702689ad..03e7c259c 100644 --- a/anta/result_manager/__init__.py +++ b/anta/result_manager/__init__.py @@ -143,7 +143,8 @@ def _update_status(self, test_status: AntaTestStatus) -> None: Parameters ---------- - test_status: AntaTestStatus to update the ResultManager status. + test_status + AntaTestStatus to update the ResultManager status. """ if test_status == "error": self.error_status = True @@ -158,7 +159,8 @@ def _update_stats(self, result: TestResult) -> None: Parameters ---------- - result: TestResult to update the statistics. + result + TestResult to update the statistics. """ result.categories = [ " ".join(word.upper() if word.lower() in ACRONYM_CATEGORIES else word.title() for word in category.split()) for category in result.categories @@ -194,7 +196,8 @@ def add(self, result: TestResult) -> None: Parameters ---------- - result: TestResult to add to the ResultManager instance. + result + TestResult to add to the ResultManager instance. """ self._result_entries.append(result) self._update_status(result.result) @@ -210,12 +213,14 @@ def get_results(self, status: set[AntaTestStatus] | None = None, sort_by: list[s Parameters ---------- - status: Optional set of AntaTestStatus enum members to filter the results. - sort_by: Optional list of TestResult fields to sort the results. + status + Optional set of AntaTestStatus enum members to filter the results. + sort_by + Optional list of TestResult fields to sort the results. Returns ------- - List of TestResult. + List[TestResult] """ # Return all results if no status is provided, otherwise return results for multiple statuses results = self._result_entries if status is None else list(chain.from_iterable(self.results_by_status.get(status, []) for status in status)) @@ -236,11 +241,13 @@ def get_total_results(self, status: set[AntaTestStatus] | None = None) -> int: Parameters ---------- - status: Optional set of AntaTestStatus enum members to filter the results. + status + Optional set of AntaTestStatus enum members to filter the results. Returns ------- - Total number of results. + int + Total number of results. """ if status is None: # Return the total number of results @@ -258,11 +265,13 @@ def filter(self, hide: set[AntaTestStatus]) -> ResultManager: Parameters ---------- - hide: Set of AntaTestStatus enum members to select tests to hide based on their status. + hide + Set of AntaTestStatus enum members to select tests to hide based on their status. Returns ------- - A filtered `ResultManager`. + ResultManager + A filtered `ResultManager`. """ possible_statuses = set(AntaTestStatus) manager = ResultManager() @@ -274,11 +283,13 @@ def filter_by_tests(self, tests: set[str]) -> ResultManager: Parameters ---------- - tests: Set of test names to filter the results. + tests + Set of test names to filter the results. Returns ------- - A filtered `ResultManager`. + ResultManager + A filtered `ResultManager`. """ manager = ResultManager() manager.results = [result for result in self._result_entries if result.test in tests] @@ -304,7 +315,8 @@ def get_tests(self) -> set[str]: Returns ------- - Set of test names. + set[str] + Set of test names. """ return {str(result.test) for result in self._result_entries} @@ -313,6 +325,7 @@ def get_devices(self) -> set[str]: Returns ------- - Set of device names. + set[str] + Set of device names. """ return {str(result.name) for result in self._result_entries} diff --git a/anta/result_manager/models.py b/anta/result_manager/models.py index e94c464ef..328fccb58 100644 --- a/anta/result_manager/models.py +++ b/anta/result_manager/models.py @@ -33,13 +33,20 @@ class TestResult(BaseModel): Attributes ---------- - name: Name of the device where the test was run. - test: Name of the test run on the device. - categories: List of categories the TestResult belongs to. Defaults to the AntaTest categories. - description: Description of the TestResult. Defaults to the AntaTest description. - result: Result of the test. Must be one of the AntaTestStatus Enum values: unset, success, failure, error or skipped. - messages: Messages to report after the test, if any. - custom_field: Custom field to store a string for flexibility in integrating with ANTA. + name : str + Name of the device where the test was run. + test : str + Name of the test run on the device. + categories : list[str] + List of categories the TestResult belongs to. Defaults to the AntaTest categories. + description : str + Description of the TestResult. Defaults to the AntaTest description. + result : AntaTestStatus + Result of the test. Must be one of the AntaTestStatus Enum values: unset, success, failure, error or skipped. + messages : list[str] + Messages to report after the test, if any. + custom_field : str | None + Custom field to store a string for flexibility in integrating with ANTA. """ @@ -56,7 +63,8 @@ def is_success(self, message: str | None = None) -> None: Parameters ---------- - message: Optional message related to the test + message + Optional message related to the test """ self._set_status(AntaTestStatus.SUCCESS, message) @@ -66,7 +74,8 @@ def is_failure(self, message: str | None = None) -> None: Parameters ---------- - message: Optional message related to the test + message + Optional message related to the test """ self._set_status(AntaTestStatus.FAILURE, message) @@ -76,7 +85,8 @@ def is_skipped(self, message: str | None = None) -> None: Parameters ---------- - message: Optional message related to the test + message + Optional message related to the test """ self._set_status(AntaTestStatus.SKIPPED, message) @@ -86,7 +96,8 @@ def is_error(self, message: str | None = None) -> None: Parameters ---------- - message: Optional message related to the test + message + Optional message related to the test """ self._set_status(AntaTestStatus.ERROR, message) @@ -96,8 +107,10 @@ def _set_status(self, status: AntaTestStatus, message: str | None = None) -> Non Parameters ---------- - status: status of the test - message: optional message + status + status of the test + message + optional message """ self.result = status diff --git a/anta/runner.py b/anta/runner.py index 5fe988ce8..e07cba94f 100644 --- a/anta/runner.py +++ b/anta/runner.py @@ -40,7 +40,8 @@ def adjust_rlimit_nofile() -> tuple[int, int]: Returns ------- - tuple[int, int]: The new soft and hard limits for open file descriptors. + tuple[int, int] + The new soft and hard limits for open file descriptors. """ try: nofile = int(os.environ.get("ANTA_NOFILE", DEFAULT_NOFILE)) @@ -61,7 +62,8 @@ def log_cache_statistics(devices: list[AntaDevice]) -> None: Parameters ---------- - devices: List of devices in the inventory. + devices + List of devices in the inventory. """ for device in devices: if device.cache_statistics is not None: @@ -80,13 +82,17 @@ async def setup_inventory(inventory: AntaInventory, tags: set[str] | None, devic Parameters ---------- - inventory: AntaInventory object that includes the device(s). - tags: Tags to filter devices from the inventory. - devices: Devices on which to run tests. None means all devices. + inventory + AntaInventory object that includes the device(s). + tags + Tags to filter devices from the inventory. + devices + Devices on which to run tests. None means all devices. Returns ------- - AntaInventory | None: The filtered inventory or None if there are no devices to run tests on. + AntaInventory | None + The filtered inventory or None if there are no devices to run tests on. """ if len(inventory) == 0: logger.info("The inventory is empty, exiting") @@ -118,14 +124,19 @@ def prepare_tests( Parameters ---------- - inventory: AntaInventory object that includes the device(s). - catalog: AntaCatalog object that includes the list of tests. - tests: Tests to run against devices. None means all tests. - tags: Tags to filter devices from the inventory. + inventory + AntaInventory object that includes the device(s). + catalog + AntaCatalog object that includes the list of tests. + tests + Tests to run against devices. None means all tests. + tags + Tags to filter devices from the inventory. Returns ------- - A mapping of devices to the tests to run or None if there are no tests to run. + defaultdict[AntaDevice, set[AntaTestDefinition]] | None + A mapping of devices to the tests to run or None if there are no tests to run. """ # Build indexes for the catalog. If `tests` is set, filter the indexes based on these tests catalog.build_indexes(filtered_tests=tests) @@ -162,11 +173,13 @@ def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinitio Parameters ---------- - selected_tests: A mapping of devices to the tests to run. The selected tests are generated by the `prepare_tests` function. + selected_tests + A mapping of devices to the tests to run. The selected tests are generated by the `prepare_tests` function. Returns ------- - The list of coroutines to run. + list[Coroutine[Any, Any, TestResult]] + The list of coroutines to run. """ coros = [] for device, test_definitions in selected_tests.items(): @@ -207,14 +220,22 @@ async def main( # noqa: PLR0913 Parameters ---------- - manager: ResultManager object to populate with the test results. - inventory: AntaInventory object that includes the device(s). - catalog: AntaCatalog object that includes the list of tests. - devices: Devices on which to run tests. None means all devices. These may come from the `--device / -d` CLI option in NRFU. - tests: Tests to run against devices. None means all tests. These may come from the `--test / -t` CLI option in NRFU. - tags: Tags to filter devices from the inventory. These may come from the `--tags` CLI option in NRFU. - established_only: Include only established device(s). - dry_run: Build the list of coroutine to run and stop before test execution. + manager + ResultManager object to populate with the test results. + inventory + AntaInventory object that includes the device(s). + catalog + AntaCatalog object that includes the list of tests. + devices + Devices on which to run tests. None means all devices. These may come from the `--device / -d` CLI option in NRFU. + tests + Tests to run against devices. None means all tests. These may come from the `--test / -t` CLI option in NRFU. + tags + Tags to filter devices from the inventory. These may come from the `--tags` CLI option in NRFU. + established_only + Include only established device(s). + dry_run + Build the list of coroutine to run and stop before test execution. """ # Adjust the maximum number of open file descriptors for the ANTA process limits = adjust_rlimit_nofile() diff --git a/anta/tests/flow_tracking.py b/anta/tests/flow_tracking.py index bab8860e6..5336cf14d 100644 --- a/anta/tests/flow_tracking.py +++ b/anta/tests/flow_tracking.py @@ -19,13 +19,17 @@ def validate_record_export(record_export: dict[str, str], tracker_info: dict[str """ Validate the record export configuration against the tracker info. - Args: - record_export (dict): The expected record export configuration. - tracker_info (dict): The actual tracker info from the command output. + Parameters + ---------- + record_export + The expected record export configuration. + tracker_info + The actual tracker info from the command output. Returns ------- - str : A failure message if the record export configuration does not match, otherwise blank string. + str + A failure message if the record export configuration does not match, otherwise blank string. """ failed_log = "" actual_export = {"inactive timeout": tracker_info.get("inactiveTimeout"), "interval": tracker_info.get("activeInterval")} @@ -39,13 +43,17 @@ def validate_exporters(exporters: list[dict[str, str]], tracker_info: dict[str, """ Validate the exporter configurations against the tracker info. - Args: - exporters (list[dict]): The list of expected exporter configurations. - tracker_info (dict): The actual tracker info from the command output. + Parameters + ---------- + exporters + The list of expected exporter configurations. + tracker_info + The actual tracker info from the command output. Returns ------- - str: Failure message if any exporter configuration does not match. + str + Failure message if any exporter configuration does not match. """ failed_log = "" for exporter in exporters: diff --git a/anta/tests/logging.py b/anta/tests/logging.py index 4d0d56a87..c5202cce1 100644 --- a/anta/tests/logging.py +++ b/anta/tests/logging.py @@ -27,12 +27,15 @@ def _get_logging_states(logger: logging.Logger, command_output: str) -> str: Parameters ---------- - logger: The logger object. - command_output: The `show logging` output. + logger + The logger object. + command_output + The `show logging` output. Returns ------- - str: The operational logging states. + str + The operational logging states. """ log_states = command_output.partition("\n\nExternal configuration:")[0] diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index d6e970414..3477fc8b2 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -26,31 +26,38 @@ def _add_bgp_failures(failures: dict[tuple[str, str | None], dict[str, Any]], af Parameters ---------- - failures: The dictionary to which the failure will be added. - afi: The address family identifier. - vrf: The VRF name. - safi: The subsequent address family identifier. - issue: A description of the issue. Can be of any type. + failures + The dictionary to which the failure will be added. + afi + The address family identifier. + vrf + The VRF name. + safi + The subsequent address family identifier. + issue + A description of the issue. Can be of any type. Example ------- The `failures` dictionary will have the following structure: - { - ('afi1', 'safi1'): { - 'afi': 'afi1', - 'safi': 'safi1', - 'vrfs': { - 'vrf1': issue1, - 'vrf2': issue2 - } - }, - ('afi2', None): { - 'afi': 'afi2', - 'vrfs': { - 'vrf1': issue3 - } + ``` + { + ('afi1', 'safi1'): { + 'afi': 'afi1', + 'safi': 'safi1', + 'vrfs': { + 'vrf1': issue1, + 'vrf2': issue2 + } + }, + ('afi2', None): { + 'afi': 'afi2', + 'vrfs': { + 'vrf1': issue3 } } + } + ``` """ key = (afi, safi) @@ -65,21 +72,27 @@ def _check_peer_issues(peer_data: dict[str, Any] | None) -> dict[str, Any]: Parameters ---------- - peer_data: The BGP peer data dictionary nested in the `show bgp summary` command. + peer_data + The BGP peer data dictionary nested in the `show bgp summary` command. Returns ------- - dict: Dictionary with keys indicating issues or an empty dictionary if no issues. + dict + Dictionary with keys indicating issues or an empty dictionary if no issues. Raises ------ - ValueError: If any of the required keys ("peerState", "inMsgQueue", "outMsgQueue") are missing in `peer_data`, i.e. invalid BGP peer data. + ValueError + If any of the required keys ("peerState", "inMsgQueue", "outMsgQueue") are missing in `peer_data`, i.e. invalid BGP peer data. Example ------- - {"peerNotFound": True} - {"peerState": "Idle", "inMsgQueue": 2, "outMsgQueue": 0} - {} + This can for instance return + ``` + {"peerNotFound": True} + {"peerState": "Idle", "inMsgQueue": 2, "outMsgQueue": 0} + {} + ``` """ if peer_data is None: @@ -106,15 +119,21 @@ def _add_bgp_routes_failure( Parameters ---------- - bgp_routes: The list of expected routes. - bgp_output: The BGP output from the device. - peer: The IP address of the BGP peer. - vrf: The name of the VRF for which the routes need to be verified. - route_type: The type of BGP routes. Defaults to 'advertised_routes'. + bgp_routes + The list of expected routes. + bgp_output + The BGP output from the device. + peer + The IP address of the BGP peer. + vrf + The name of the VRF for which the routes need to be verified. + route_type + The type of BGP routes. Defaults to 'advertised_routes'. Returns ------- - dict[str, dict[str, dict[str, dict[str, list[str]]]]]: A dictionary containing the missing routes and invalid or inactive routes. + dict[str, dict[str, dict[str, dict[str, list[str]]]]] + A dictionary containing the missing routes and invalid or inactive routes. """ # Prepare the failure routes dictionary diff --git a/anta/tests/routing/isis.py b/anta/tests/routing/isis.py index dee472571..5154d6714 100644 --- a/anta/tests/routing/isis.py +++ b/anta/tests/routing/isis.py @@ -20,13 +20,15 @@ def _count_isis_neighbor(isis_neighbor_json: dict[str, Any]) -> int: """Count the number of isis neighbors. - Args - ---- - isis_neighbor_json: The JSON output of the `show isis neighbors` command. + Parameters + ---------- + isis_neighbor_json + The JSON output of the `show isis neighbors` command. Returns ------- - int: The number of isis neighbors. + int + The number of isis neighbors. """ count = 0 @@ -39,13 +41,15 @@ def _count_isis_neighbor(isis_neighbor_json: dict[str, Any]) -> int: def _get_not_full_isis_neighbors(isis_neighbor_json: dict[str, Any]) -> list[dict[str, Any]]: """Return the isis neighbors whose adjacency state is not `up`. - Args - ---- - isis_neighbor_json: The JSON output of the `show isis neighbors` command. + Parameters + ---------- + isis_neighbor_json + The JSON output of the `show isis neighbors` command. Returns ------- - list[dict[str, Any]]: A list of isis neighbors whose adjacency state is not `UP`. + list[dict[str, Any]] + A list of isis neighbors whose adjacency state is not `UP`. """ return [ @@ -66,14 +70,17 @@ def _get_not_full_isis_neighbors(isis_neighbor_json: dict[str, Any]) -> list[dic def _get_full_isis_neighbors(isis_neighbor_json: dict[str, Any], neighbor_state: Literal["up", "down"] = "up") -> list[dict[str, Any]]: """Return the isis neighbors whose adjacency state is `up`. - Args - ---- - isis_neighbor_json: The JSON output of the `show isis neighbors` command. - neighbor_state: Value of the neihbor state we are looking for. Default up + Parameters + ---------- + isis_neighbor_json + The JSON output of the `show isis neighbors` command. + neighbor_state + Value of the neihbor state we are looking for. Default up Returns ------- - list[dict[str, Any]]: A list of isis neighbors whose adjacency state is not `UP`. + list[dict[str, Any]] + A list of isis neighbors whose adjacency state is not `UP`. """ return [ diff --git a/anta/tests/routing/ospf.py b/anta/tests/routing/ospf.py index cb325d6bd..3ffd81d53 100644 --- a/anta/tests/routing/ospf.py +++ b/anta/tests/routing/ospf.py @@ -20,11 +20,13 @@ def _count_ospf_neighbor(ospf_neighbor_json: dict[str, Any]) -> int: Parameters ---------- - ospf_neighbor_json: The JSON output of the `show ip ospf neighbor` command. + ospf_neighbor_json + The JSON output of the `show ip ospf neighbor` command. Returns ------- - int: The number of OSPF neighbors. + int + The number of OSPF neighbors. """ count = 0 @@ -39,11 +41,13 @@ def _get_not_full_ospf_neighbors(ospf_neighbor_json: dict[str, Any]) -> list[dic Parameters ---------- - ospf_neighbor_json: The JSON output of the `show ip ospf neighbor` command. + ospf_neighbor_json + The JSON output of the `show ip ospf neighbor` command. Returns ------- - list[dict[str, Any]]: A list of OSPF neighbors whose adjacency state is not `full`. + list[dict[str, Any]] + A list of OSPF neighbors whose adjacency state is not `full`. """ return [ @@ -65,11 +69,13 @@ def _get_ospf_max_lsa_info(ospf_process_json: dict[str, Any]) -> list[dict[str, Parameters ---------- - ospf_process_json: OSPF process information in JSON format. + ospf_process_json + OSPF process information in JSON format. Returns ------- - list[dict[str, Any]]: A list of dictionaries containing OSPF LSAs information. + list[dict[str, Any]] + A list of dictionaries containing OSPF LSAs information. """ return [ diff --git a/anta/tools.py b/anta/tools.py index 55748b492..00aad5afe 100644 --- a/anta/tools.py +++ b/anta/tools.py @@ -34,12 +34,15 @@ def get_failed_logs(expected_output: dict[Any, Any], actual_output: dict[Any, An Parameters ---------- - expected_output (dict): Expected output of a test. - actual_output (dict): Actual output of a test + expected_output + Expected output of a test. + actual_output + Actual output of a test Returns ------- - str: Failed log of a test. + str + Failed log of a test. """ failed_logs = [] @@ -65,12 +68,15 @@ def custom_division(numerator: float, denominator: float) -> int | float: Parameters ---------- - numerator: The numerator. - denominator: The denominator. + numerator + The numerator. + denominator + The denominator. Returns ------- - Union[int, float]: The result of the division. + Union[int, float] + The result of the division. """ result = numerator / denominator return int(result) if result.is_integer() else result @@ -304,11 +310,13 @@ def cprofile(sort_by: str = "cumtime") -> Callable[[F], F]: Parameters ---------- - sort_by (str): The criterion to sort the profiling results. Default is 'cumtime'. + sort_by + The criterion to sort the profiling results. Default is 'cumtime'. Returns ------- - Callable: The decorated function with conditional profiling. + Callable + The decorated function with conditional profiling. """ def decorator(func: F) -> F: @@ -320,11 +328,14 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any: Parameters ---------- - *args: Arbitrary positional arguments. - **kwargs: Arbitrary keyword arguments. + *args + Arbitrary positional arguments. + **kwargs + Arbitrary keyword arguments. Returns ------- + Any The result of the function call. """ cprofile_file = os.environ.get("ANTA_CPROFILE")