Skip to content

ANTA Result API

ResultManager

ResultManager()

Manager of ANTA Results.

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 attribute is set to True.

Attributes:

Name Type Description
results list[TestResult]
dump list[dict[str, Any]]
status AntaTestStatus

Status rerpesenting all the results.

error_status bool

Will be True if a test returned an error.

results_by_status dict[AntaTestStatus, list[TestResult]]
dump list[dict[str, Any]]
json str
device_stats dict[str, DeviceStats]
category_stats dict[str, CategoryStats]
test_stats dict[str, TestStats]

category_stats property

category_stats: dict[str, CategoryStats]

Get the category statistics.

device_stats property

device_stats: dict[str, DeviceStats]

Get the device statistics.

dump property

dump: list[dict[str, Any]]

Get a list of dictionary of the results.

json property

json: str

Get a JSON representation of the results.

results property writable

results: list[TestResult]

Get the list of TestResult.

results_by_category cached property

results_by_category: list[TestResult]

A cached property that returns the list of results sorted by categories.

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.

test_stats property

test_stats: dict[str, TestStats]

Get the test statistics.

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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
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._results.append(result)
    self._update_status(result.result)
    self._stats_in_sync = False

    # Every time a new result is added, we need to clear the cached properties
    for name in ["results_by_status", "results_by_category"]:
        self.__dict__.pop(name, 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
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
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 deprecated

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

This method is deprecated. This will be removed in ANTA v2.0.0.

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
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
@deprecated("This method is deprecated. This will be removed in ANTA v2.0.0.", category=DeprecationWarning)
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._results if result.name in devices]
    return manager

filter_by_tests deprecated

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

This method is deprecated. This will be removed in ANTA v2.0.0.

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
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
@deprecated("This method is deprecated. This will be removed in ANTA v2.0.0.", category=DeprecationWarning)
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._results if result.test in tests]
    return manager

get_devices deprecated

get_devices() -> set[str]
Deprecated

This method is deprecated. This will be removed in ANTA v2.0.0.

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
398
399
400
401
402
403
404
405
406
407
@deprecated("This method is deprecated. This will be removed in ANTA v2.0.0.", category=DeprecationWarning)
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._results}

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
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
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._results 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) or "" 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
295
296
297
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 deprecated

get_tests() -> set[str]
Deprecated

This method is deprecated. This will be removed in ANTA v2.0.0.

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
387
388
389
390
391
392
393
394
395
396
@deprecated("This method is deprecated. This will be removed in ANTA v2.0.0.", category=DeprecationWarning)
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._results}

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
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
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)

merge_results classmethod

merge_results(
    results_managers: list[ResultManager],
) -> ResultManager

Merge multiple ResultManager instances.

Parameters:

Name Type Description Default
results_managers list[ResultManager]

A list of ResultManager instances to merge.

required

Returns:

Type Description
ResultManager

A new ResultManager instance containing the results of all the input ResultManagers.

Source code in anta/result_manager/__init__.py
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
@classmethod
def merge_results(cls, results_managers: list[ResultManager]) -> ResultManager:
    """Merge multiple ResultManager instances.

    Parameters
    ----------
    results_managers
        A list of ResultManager instances to merge.

    Returns
    -------
    ResultManager
        A new ResultManager instance containing the results of all the input ResultManagers.
    """
    combined_results = list(chain(*(rm.results for rm in results_managers)))
    merged_manager = cls()
    merged_manager.results = combined_results
    return merged_manager

reset

reset() -> None

Create or reset the attributes of the ResultManager instance.

Source code in anta/result_manager/__init__.py
82
83
84
85
86
87
88
89
def reset(self) -> None:
    """Create or reset the attributes of the ResultManager instance."""
    self._results = []
    self.status = AntaTestStatus.UNSET
    self.error_status = False

    # Initialize the statistics attributes
    self._reset_stats()

sort

sort(sort_by: list[str]) -> ResultManager

Sort the ResultManager results based on TestResult fields.

Parameters:

Name Type Description Default
sort_by list[str]

List of TestResult fields to sort the results.

required
Source code in anta/result_manager/__init__.py
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def sort(self, sort_by: list[str]) -> ResultManager:
    """Sort the ResultManager results based on TestResult fields.

    Parameters
    ----------
    sort_by
        List of TestResult fields to sort the results.
    """
    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)
    self._results.sort(key=lambda result: [getattr(result, field) or "" for field in sort_by])
    return self

TestResult

Bases: BaseTestResult

Describe the result of a test from a single device.

Attributes:

Name Type Description
name str

Name of the device on which the test was run.

test str

Name of the AntaTest subclass.

categories list[str]

List of categories the TestResult belongs to. Defaults to the AntaTest subclass categories.

description str

Description of the TestResult. Defaults to the AntaTest subclass description.

result AntaTestStatus

Result of the test.

messages list[str]

Messages reported by the test.

custom_field str | None

Custom field to store a string for flexibility in integrating with ANTA.