diff options
Diffstat (limited to 'anta/result_manager/__init__.py')
| -rw-r--r-- | anta/result_manager/__init__.py | 203 |
1 files changed, 96 insertions, 107 deletions
diff --git a/anta/result_manager/__init__.py b/anta/result_manager/__init__.py index 9ce880e..9e1f6ae 100644 --- a/anta/result_manager/__init__.py +++ b/anta/result_manager/__init__.py @@ -1,35 +1,32 @@ # Copyright (c) 2023-2024 Arista Networks, Inc. # Use of this source code is governed by the Apache License 2.0 # that can be found in the LICENSE file. -""" -Result Manager Module for ANTA. -""" +"""Result Manager module for ANTA.""" + from __future__ import annotations import json -import logging +from typing import TYPE_CHECKING from pydantic import TypeAdapter from anta.custom_types import TestStatus -from anta.result_manager.models import TestResult -logger = logging.getLogger(__name__) +if TYPE_CHECKING: + from anta.result_manager.models import TestResult class ResultManager: - """ - Helper to manage Test Results and generate reports. - - Examples: + """Helper to manage Test Results and generate reports. + Examples + -------- Create Inventory: inventory_anta = AntaInventory.parse( filename='examples/inventory.yml', username='ansible', password='ansible', - timeout=0.5 ) Create Result Manager: @@ -38,17 +35,17 @@ class ResultManager: Run tests for all connected devices: - for device in inventory_anta.get_inventory(): - manager.add_test_result( + for device in inventory_anta.get_inventory().devices: + manager.add( VerifyNTP(device=device).test() ) - manager.add_test_result( + manager.add( VerifyEOSVersion(device=device).test(version='4.28.3M') ) Print result in native format: - manager.get_results() + manager.results [ TestResult( host=IPv4Address('192.168.0.10'), @@ -63,11 +60,11 @@ class ResultManager: message=None ), ] + """ def __init__(self) -> None: - """ - Class constructor. + """Class constructor. The status of the class is initialized to "unset" @@ -88,124 +85,116 @@ class ResultManager: error_status is set to True. """ self._result_entries: list[TestResult] = [] - # Initialize status self.status: TestStatus = "unset" self.error_status = False def __len__(self) -> int: - """ - Implement __len__ method to count number of results. - """ + """Implement __len__ method to count number of results.""" return len(self._result_entries) - def _update_status(self, test_status: TestStatus) -> None: - """ - Update ResultManager status based on the table above. - """ - ResultValidator = TypeAdapter(TestStatus) - ResultValidator.validate_python(test_status) - if test_status == "error": - self.error_status = True - return - if self.status == "unset": - self.status = test_status - elif self.status == "skipped" and test_status in {"success", "failure"}: - self.status = test_status - elif self.status == "success" and test_status == "failure": - self.status = "failure" - - def add_test_result(self, entry: TestResult) -> None: - """Add a result to the list + @property + def results(self) -> list[TestResult]: + """Get the list of TestResult.""" + return self._result_entries - Args: - entry (TestResult): TestResult data to add to the report - """ - logger.debug(entry) - self._result_entries.append(entry) - self._update_status(entry.result) + @results.setter + def results(self, value: list[TestResult]) -> None: + self._result_entries = [] + self.status = "unset" + self.error_status = False + for e in value: + self.add(e) - def add_test_results(self, entries: list[TestResult]) -> None: - """Add a list of results to the list + @property + def json(self) -> str: + """Get a JSON representation of the results.""" + return json.dumps([result.model_dump() for result in self._result_entries], indent=4) - Args: - entries (list[TestResult]): List of TestResult data to add to the report - """ - for e in entries: - self.add_test_result(e) + def add(self, result: TestResult) -> None: + """Add a result to the ResultManager instance. - def get_status(self, ignore_error: bool = False) -> str: - """ - Returns the current status including error_status if ignore_error is False - """ + Args: + ---- + result: TestResult to add to the ResultManager instance. + """ + + def _update_status(test_status: TestStatus) -> None: + result_validator = TypeAdapter(TestStatus) + result_validator.validate_python(test_status) + if test_status == "error": + self.error_status = True + return + if self.status == "unset" or self.status == "skipped" and test_status in {"success", "failure"}: + self.status = test_status + elif self.status == "success" and test_status == "failure": + self.status = "failure" + + self._result_entries.append(result) + _update_status(result.result) + + def get_status(self, *, ignore_error: bool = False) -> str: + """Return the current status including error_status if ignore_error is False.""" return "error" if self.error_status and not ignore_error else self.status - def get_results(self) -> list[TestResult]: - """ - Expose list of all test results in different format - - Returns: - any: List of results. - """ - return self._result_entries + def filter(self, hide: set[TestStatus]) -> ResultManager: + """Get a filtered ResultManager based on test status. - def get_json_results(self) -> str: - """ - Expose list of all test results in JSON + Args: + ---- + hide: set of TestStatus literals to select tests to hide based on their status. - Returns: - str: JSON dumps of the list of results + Returns + ------- + A filtered `ResultManager`. """ - result = [result.model_dump() for result in self._result_entries] - return json.dumps(result, indent=4) + manager = ResultManager() + manager.results = [test for test in self._result_entries if test.result not in hide] + return manager - def get_result_by_test(self, test_name: str) -> list[TestResult]: - """ - Get list of test result for a given test. + def filter_by_tests(self, tests: set[str]) -> ResultManager: + """Get a filtered ResultManager that only contains specific tests. Args: - test_name (str): Test name to use to filter results - output_format (str, optional): format selector. Can be either native/list. Defaults to 'native'. + ---- + tests: Set of test names to filter the results. - Returns: - list[TestResult]: List of results related to the test. + Returns + ------- + A filtered `ResultManager`. """ - return [result for result in self._result_entries if str(result.test) == test_name] + manager = ResultManager() + manager.results = [result for result in self._result_entries if result.test in tests] + return manager - def get_result_by_host(self, host_ip: str) -> list[TestResult]: - """ - Get list of test result for a given host. + def filter_by_devices(self, devices: set[str]) -> ResultManager: + """Get a filtered ResultManager that only contains specific devices. Args: - host_ip (str): IP Address of the host to use to filter results. - output_format (str, optional): format selector. Can be either native/list. Defaults to 'native'. + ---- + devices: Set of device names to filter the results. - Returns: - list[TestResult]: List of results related to the host. + Returns + ------- + A filtered `ResultManager`. """ - return [result for result in self._result_entries if str(result.name) == host_ip] + manager = ResultManager() + manager.results = [result for result in self._result_entries if result.name in devices] + return manager - def get_testcases(self) -> list[str]: - """ - Get list of name of all test cases in current manager. + def get_tests(self) -> set[str]: + """Get the set of all the test names. - Returns: - list[str]: List of names for all tests. + Returns + ------- + Set of test names. """ - result_list = [] - for testcase in self._result_entries: - if str(testcase.test) not in result_list: - result_list.append(str(testcase.test)) - return result_list + return {str(result.test) for result in self._result_entries} - def get_hosts(self) -> list[str]: - """ - Get list of IP addresses in current manager. + def get_devices(self) -> set[str]: + """Get the set of all the device names. - Returns: - list[str]: List of IP addresses. + Returns + ------- + Set of device names. """ - result_list = [] - for testcase in self._result_entries: - if str(testcase.name) not in result_list: - result_list.append(str(testcase.name)) - return result_list + return {str(result.name) for result in self._result_entries} |