aristanetworks · carl-baillargeon · Mar 24, 2025 · Mar 26, 2025 · Apr 11, 2025 · Apr 11, 2025
diff --git a/anta/models.py b/anta/models.py
@@ -431,6 +431,8 @@ def __init__(
         device: AntaDevice,
         inputs: dict[str, Any] | AntaTest.Input | None = None,
         eos_data: list[dict[Any, Any] | str] | None = None,
+        *,
+        save_evidence: bool = False,
     ) -> None:
        """Initialize an AntaTest instance.
 
@@ -443,6 +445,8 @@ def __init__(
         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.
+        save_evidence
+            Save the test inputs and commands used to run the test in the TestResult object.
         """
         self.logger: logging.Logger = logging.getLogger(f"{self.module}.{self.__class__.__name__}")
         self.device: AntaDevice = device
@@ -458,6 +462,9 @@ def __init__(
         if self.result.result == AntaTestStatus.UNSET:
             self._init_commands(eos_data)
 
+            if save_evidence:
+                self.result.evidence = {"inputs": self.inputs.model_dump(mode="json", exclude_unset=True), "commands": self.instance_commands}
+
     def _init_inputs(self, inputs: dict[str, Any] | AntaTest.Input | None) -> None:
         """Instantiate the `inputs` instance attribute with an `AntaTest.Input` instance to validate test inputs using the model.
 

diff --git a/anta/result_manager/__init__.py b/anta/result_manager/__init__.py
@@ -102,7 +102,7 @@ def results(self, value: list[TestResult]) -> None:
     @property
     def dump(self) -> list[dict[str, Any]]:
         """Get a list of dictionary of the results."""
-        return [result.model_dump() for result in self._result_entries]
+        return [result.model_dump(exclude={"evidence"}) for result in self._result_entries]
 
     @property
     def json(self) -> str:

diff --git a/anta/result_manager/models.py b/anta/result_manager/models.py
@@ -7,6 +7,7 @@
 
 from dataclasses import dataclass, field
 from enum import Enum
+from typing import Any
 
 from pydantic import BaseModel
 
@@ -47,6 +48,8 @@ class TestResult(BaseModel):
         Messages to report after the test, if any.
     custom_field : str | None
         Custom field to store a string for flexibility in integrating with ANTA.
+    evidence : dict[str, Any] | None
+        Optional evidence attached to the result.
 
     """
 
@@ -57,6 +60,7 @@ class TestResult(BaseModel):
     result: AntaTestStatus = AntaTestStatus.UNSET
     messages: list[str] = []
     custom_field: str | None = None
+    evidence: dict[str, Any] | None = None
 
     def is_success(self, message: str | None = None) -> None:
         """Set status to success.

diff --git a/anta/runner.py b/anta/runner.py
@@ -181,7 +181,12 @@ def prepare_tests(
     return device_to_tests
 
 
-def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinition]], manager: ResultManager | None = None) -> list[Coroutine[Any, Any, TestResult]]:
+def get_coroutines(
+    selected_tests: defaultdict[AntaDevice, set[AntaTestDefinition]],
+    manager: ResultManager | None = None,
+    *,
+    save_evidence: bool = False,
+) -> list[Coroutine[Any, Any, TestResult]]:
     """Get the coroutines for the ANTA run.
 
     Parameters
@@ -190,6 +195,8 @@ def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinitio
         A mapping of devices to the tests to run. The selected tests are generated by the `prepare_tests` function.
     manager
         An optional ResultManager object to pre-populate with the test results. Used in dry-run mode.
+    save_evidence
+        Save each test inputs and command outputs to their respective result in the ResultManager.
 
     Returns
     -------
@@ -200,7 +207,7 @@ def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinitio
     for device, test_definitions in selected_tests.items():
         for test in test_definitions:
             try:
-                test_instance = test.test(device=device, inputs=test.inputs)
+                test_instance = test.test(device=device, inputs=test.inputs, save_evidence=save_evidence)
                 if manager is not None:
                     manager.add(test_instance.result)
                 coros.append(test_instance.test())
@@ -228,6 +235,7 @@ async def main(
     *,
     established_only: bool = True,
     dry_run: bool = False,
+    save_evidence: bool = False,
 ) -> None:
     """Run ANTA.
 
@@ -252,6 +260,8 @@ async def main(
         Include only established device(s).
     dry_run
         Build the list of coroutine to run and stop before test execution.
+    save_evidence
+        Save each test inputs and command outputs to their respective result in the ResultManager.
     """
     if not catalog.tests:
         logger.info("The list of tests is empty, exiting")
@@ -295,7 +305,7 @@ async def main(
                 "Please consult the ANTA FAQ."
             )
 
-        coroutines = get_coroutines(selected_tests, manager if dry_run else None)
+        coroutines = get_coroutines(selected_tests, manager if dry_run else None, save_evidence=save_evidence)
 
     if dry_run:
         logger.info("Dry-run mode, exiting before running the tests.")

diff --git a/anta/tests/cvx.py b/anta/tests/cvx.py
@@ -92,8 +92,8 @@ def test(self) -> None:
                 self.result.is_failure("Management CVX status - Not configured")
                 return
             cluster_state = "enabled" if cluster_state else "disabled"
-            self.inputs.enabled = "enabled" if self.inputs.enabled else "disabled"
-            self.result.is_failure(f"Management CVX status is not valid: Expected: {self.inputs.enabled} Actual: {cluster_state}")
+            expected_state = "enabled" if self.inputs.enabled else "disabled"
+            self.result.is_failure(f"Management CVX status is not valid: Expected: {expected_state} Actual: {cluster_state}")
 
 
 class VerifyMcsServerMounts(AntaTest):