Result Manager module

Result Manager definition

ResultManager

ResultManager()

Helper to manage Test Results and generate reports.

Examples

Create Inventory:

inventory_anta = AntaInventory.parse(
    filename='examples/inventory.yml',
    username='ansible',
    password='ansible',
)

Create Result Manager:

manager = ResultManager()

Run tests for all connected devices:

for device in inventory_anta.get_inventory().devices:
    manager.add(
        VerifyNTP(device=device).test()
    )
    manager.add(
        VerifyEOSVersion(device=device).test(version='4.28.3M')
    )

Print result in native format:

manager.results
[
    TestResult(
        name="pf1",
        test="VerifyZeroTouch",
        categories=["configuration"],
        description="Verifies ZeroTouch is disabled",
        result="success",
        messages=[],
        custom_field=None,
    ),
    TestResult(
        name="pf1",
        test='VerifyNTP',
        categories=["software"],
        categories=['system'],
        description='Verifies if NTP is synchronised.',
        result='failure',
        messages=["The device is not synchronized with the configured NTP server(s): 'NTP is disabled.'"],
        custom_field=None,
    ),
]

The status of the class is initialized to “unset”

Then when adding a test with a status that is NOT ‘error’ the following table shows the updated status:

Current Status	Added test Status	Updated Status
unset	Any	Any
skipped	unset, skipped	skipped
skipped	success	success
skipped	failure	failure
success	unset, skipped, success	success
success	failure	failure
failure	unset, skipped success, failure	failure

If the status of the added test is error, the status is untouched and the error_status is set to True.

Source code in anta/result_manager/__init__.py

def __init__(self) -> None:
    """Class constructor.

    The status of the class is initialized to "unset"

    Then when adding a test with a status that is NOT 'error' the following
    table shows the updated status:

    | Current Status |         Added test Status       | Updated Status |
    | -------------- | ------------------------------- | -------------- |
    |      unset     |              Any                |       Any      |
    |     skipped    |         unset, skipped          |     skipped    |
    |     skipped    |            success              |     success    |
    |     skipped    |            failure              |     failure    |
    |     success    |     unset, skipped, success     |     success    |
    |     success    |            failure              |     failure    |
    |     failure    | unset, skipped success, failure |     failure    |

    If the status of the added test is error, the status is untouched and the
    error_status is set to True.
    """
    self._result_entries: list[TestResult] = []
    self.status: AntaTestStatus = AntaTestStatus.UNSET
    self.error_status = False

    self.device_stats: defaultdict[str, DeviceStats] = defaultdict(DeviceStats)
    self.category_stats: defaultdict[str, CategoryStats] = defaultdict(CategoryStats)
    self.test_stats: defaultdict[str, TestStats] = defaultdict(TestStats)

json property

json: str

Get a JSON representation of the results.

results property writable

results: list[TestResult]

Get the list of TestResult.

results_by_status cached property

results_by_status: dict[AntaTestStatus, list[TestResult]]

A cached property that returns the results grouped by status.

sorted_category_stats property

sorted_category_stats: dict[str, CategoryStats]

A property that returns the category_stats dictionary sorted by key name.

__len__

__len__() -> int

Implement len method to count number of results.

Source code in anta/result_manager/__init__.py

def __len__(self) -> int:
    """Implement __len__ method to count number of results."""
    return len(self._result_entries)

add

add(result: TestResult) -> None

Add a result to the ResultManager instance.

The result is added to the internal list of results and the overall status of the ResultManager instance is updated based on the added test status.

Parameters:

Name	Type	Description	Default
result	TestResult	TestResult to add to the ResultManager instance.	required

Source code in anta/result_manager/__init__.py

def add(self, result: TestResult) -> None:
    """Add a result to the ResultManager instance.

    The result is added to the internal list of results and the overall status
    of the ResultManager instance is updated based on the added test status.

    Parameters
    ----------
    result
        TestResult to add to the ResultManager instance.
    """
    self._result_entries.append(result)
    self._update_status(result.result)
    self._update_stats(result)

    # Every time a new result is added, we need to clear the cached property
    self.__dict__.pop("results_by_status", None)

filter

filter(hide: set[AntaTestStatus]) -> ResultManager

Get a filtered ResultManager based on test status.

Parameters:

Name	Type	Description	Default
hide	set[AntaTestStatus]	Set of AntaTestStatus enum members to select tests to hide based on their status.	required

Returns:

Type	Description
ResultManager	A filtered ResultManager.

Source code in anta/result_manager/__init__.py

def filter(self, hide: set[AntaTestStatus]) -> ResultManager:
    """Get a filtered ResultManager based on test status.

    Parameters
    ----------
    hide
        Set of AntaTestStatus enum members to select tests to hide based on their status.

    Returns
    -------
    ResultManager
        A filtered `ResultManager`.
    """
    possible_statuses = set(AntaTestStatus)
    manager = ResultManager()
    manager.results = self.get_results(possible_statuses - hide)
    return manager

filter_by_devices

filter_by_devices(devices: set[str]) -> ResultManager

Get a filtered ResultManager that only contains specific devices.

Parameters:

Name	Type	Description	Default
devices	set[str]	Set of device names to filter the results.	required

Returns:

Type	Description
ResultManager	A filtered ResultManager.

Source code in anta/result_manager/__init__.py

def filter_by_devices(self, devices: set[str]) -> ResultManager:
    """Get a filtered ResultManager that only contains specific devices.

    Parameters
    ----------
    devices
        Set of device names to filter the results.

    Returns
    -------
    ResultManager
        A filtered `ResultManager`.
    """
    manager = ResultManager()
    manager.results = [result for result in self._result_entries if result.name in devices]
    return manager

filter_by_tests

filter_by_tests(tests: set[str]) -> ResultManager

Get a filtered ResultManager that only contains specific tests.

Parameters:

Name	Type	Description	Default
tests	set[str]	Set of test names to filter the results.	required

Returns:

Type	Description
ResultManager	A filtered ResultManager.

Source code in anta/result_manager/__init__.py

def filter_by_tests(self, tests: set[str]) -> ResultManager:
    """Get a filtered ResultManager that only contains specific tests.

    Parameters
    ----------
    tests
        Set of test names to filter the results.

    Returns
    -------
    ResultManager
        A filtered `ResultManager`.
    """
    manager = ResultManager()
    manager.results = [result for result in self._result_entries if result.test in tests]
    return manager

get_devices

get_devices() -> set[str]

Get the set of all the device names.

Returns:

Type	Description
set[str]	Set of device names.

Source code in anta/result_manager/__init__.py

def get_devices(self) -> set[str]:
    """Get the set of all the device names.

    Returns
    -------
    set[str]
        Set of device names.
    """
    return {str(result.name) for result in self._result_entries}

get_results

get_results(
    status: set[AntaTestStatus] | None = None,
    sort_by: list[str] | None = None,
) -> list[TestResult]

Get the results, optionally filtered by status and sorted by TestResult fields.

If no status is provided, all results are returned.

Parameters:

Name	Type	Description	Default
status	set[AntaTestStatus] | None	Optional set of AntaTestStatus enum members to filter the results.	None
sort_by	list[str] | None	Optional list of TestResult fields to sort the results.	None

Returns:

Type	Description
list[TestResult]	List of results.

Source code in anta/result_manager/__init__.py

def get_results(self, status: set[AntaTestStatus] | None = None, sort_by: list[str] | None = None) -> list[TestResult]:
    """Get the results, optionally filtered by status and sorted by TestResult fields.

    If no status is provided, all results are returned.

    Parameters
    ----------
    status
        Optional set of AntaTestStatus enum members to filter the results.
    sort_by
        Optional list of TestResult fields to sort the results.

    Returns
    -------
    list[TestResult]
        List of results.
    """
    # 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))

    if sort_by:
        accepted_fields = TestResult.model_fields.keys()
        if not set(sort_by).issubset(set(accepted_fields)):
            msg = f"Invalid sort_by fields: {sort_by}. Accepted fields are: {list(accepted_fields)}"
            raise ValueError(msg)
        results = sorted(results, key=lambda result: [getattr(result, field) for field in sort_by])

    return results

get_status

get_status(*, ignore_error: bool = False) -> str

Return the current status including error_status if ignore_error is False.

Source code in anta/result_manager/__init__.py

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

get_tests

get_tests() -> set[str]

Get the set of all the test names.

Returns:

Type	Description
set[str]	Set of test names.

Source code in anta/result_manager/__init__.py

def get_tests(self) -> set[str]:
    """Get the set of all the test names.

    Returns
    -------
    set[str]
        Set of test names.
    """
    return {str(result.test) for result in self._result_entries}

get_total_results

get_total_results(
    status: set[AntaTestStatus] | None = None,
) -> int

Get the total number of results, optionally filtered by status.

If no status is provided, the total number of results is returned.

Parameters:

Name	Type	Description	Default
status	set[AntaTestStatus] | None	Optional set of AntaTestStatus enum members to filter the results.	None

Returns:

Type	Description
int	Total number of results.

Source code in anta/result_manager/__init__.py

def get_total_results(self, status: set[AntaTestStatus] | None = None) -> int:
    """Get the total number of results, optionally filtered by status.

    If no status is provided, the total number of results is returned.

    Parameters
    ----------
    status
        Optional set of AntaTestStatus enum members to filter the results.

    Returns
    -------
    int
        Total number of results.
    """
    if status is None:
        # Return the total number of results
        return sum(len(results) for results in self.results_by_status.values())

    # Return the total number of results for multiple statuses
    return sum(len(self.results_by_status.get(status, [])) for status in status)

  options:
    filters: ["!^_[^_]", "!^__len__"]