fix(anta.tests): Cleaning up flow-tracking tests module

anta/input_models/flow_tracking.py

@@ -0,0 +1,72 @@
+# Copyright (c) 2023-2025 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for flow tracking tests."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict
+
+
+class FlowTracker(BaseModel):
+    """Flow Tracking model representing the tracker details."""
+
+    model_config = ConfigDict(extra="forbid")
+    name: str
+    """The name of the flow tracker."""
+    record_export: RecordExport | None = None
+    """Configuration for record export, specifying details about timeouts."""
+    exporters: list[Exporter] | None = None
+    """A list of exporters associated with the flow tracker."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the FlowTracker for reporting.
+
+        Examples
+        --------
+        Flow Tracker: FLOW-TRACKER
+
+        """
+        return f"Flow Tracker: {self.name}"
+
+
+class RecordExport(BaseModel):
+    """Model representing the record export configuration for a flow tracker."""
+
+    model_config = ConfigDict(extra="forbid")
+    on_inactive_timeout: int
+    """The timeout in milliseconds for exporting flow records when the flow becomes inactive."""
+    on_interval: int
+    """The interval in milliseconds for exporting flow records."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the RecordExport for reporting.
+
+        Examples
+        --------
+        Inactive Timeout: 60000, Active Interval: 300000
+
+        """
+        return f"Inactive Timeout: {self.on_inactive_timeout}, Active Interval: {self.on_interval}"
+
+
+class Exporter(BaseModel):
+    """Model representing the exporter used for flow record export."""
+
+    model_config = ConfigDict(extra="forbid")
+    name: str
+    """The name of the exporter."""
+    local_interface: str
+    """The local interface used by the exporter to send flow records."""
+    template_interval: int
+    """The template interval, in milliseconds, for the exporter to refresh the flow template."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the Exporter for reporting.
+
+        Examples
+        --------
+        Exporter: CVP-TELEMETRY
+
+        """
+        return f"Exporter: {self.name}"

anta/tests/flow_tracking.py

@@ -9,37 +9,13 @@
 
 from typing import ClassVar
 
-from pydantic import BaseModel
-
 from anta.decorators import skip_on_platforms
+from anta.input_models.flow_tracking import FlowTracker
 from anta.models import AntaCommand, AntaTemplate, AntaTest
-from anta.tools import get_failed_logs
-
+from anta.tools import get_value
 
-def validate_record_export(record_export: dict[str, str], tracker_info: dict[str, str]) -> str:
-    """Validate the record export configuration against the tracker info.
-
-    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.
-    """
-    failed_log = ""
-    actual_export = {"inactive timeout": tracker_info.get("inactiveTimeout"), "interval": tracker_info.get("activeInterval")}
-    expected_export = {"inactive timeout": record_export.get("on_inactive_timeout"), "interval": record_export.get("on_interval")}
-    if actual_export != expected_export:
-        failed_log = get_failed_logs(expected_export, actual_export)
-    return failed_log
-
-
-def validate_exporters(exporters: list[dict[str, str]], tracker_info: dict[str, str]) -> str:
+def validate_exporters(exporters: list[dict[str, str]], tracker_info: dict[str, str]) -> list[str]:
     """Validate the exporter configurations against the tracker info.
 
     Parameters
@@ -51,36 +27,52 @@ def validate_exporters(exporters: list[dict[str, str]], tracker_info: dict[str,
 
     Returns
     -------
-    str
-        Failure message if any exporter configuration does not match.
+    list
+        List of failure messages for any exporter configuration that does not match.
     """
-    failed_log = ""
+    failure_messages = []
     for exporter in exporters:
-        exporter_name = exporter["name"]
+        exporter_name = exporter.name
         actual_exporter_info = tracker_info["exporters"].get(exporter_name)
         if not actual_exporter_info:
-            failed_log += f"\nExporter `{exporter_name}` is not configured."
+            failure_messages.append(f"{exporter} - Not configured")
             continue
+        local_interface = actual_exporter_info["localIntf"]
+        template_interval = actual_exporter_info["templateInterval"]
 
-        expected_exporter_data = {"local interface": exporter["local_interface"], "template interval": exporter["template_interval"]}
-        actual_exporter_data = {"local interface": actual_exporter_info["localIntf"], "template interval": actual_exporter_info["templateInterval"]}
+        if local_interface != exporter.local_interface:
+            failure_messages.append(f"{exporter} - Incorrect local interface - Expected: {exporter.local_interface}, Actual: {local_interface}")
 
-        if expected_exporter_data != actual_exporter_data:
-            failed_msg = get_failed_logs(expected_exporter_data, actual_exporter_data)
-            failed_log += f"\nExporter `{exporter_name}`: {failed_msg}"
-    return failed_log
+        if template_interval != exporter.template_interval:
+            failure_messages.append(f"{exporter} - Incorrect template interval - Expected: {exporter.template_interval}, Actual: {template_interval}")
+    return failure_messages
 
 
 class VerifyHardwareFlowTrackerStatus(AntaTest):
-    """Verifies if hardware flow tracking is running and an input tracker is active.
+    """Verifies the hardware flow tracking state.
+
+    This test performs the following checks:
 
-    This test optionally verifies the tracker interval/timeout and exporter configuration.
+      1. Confirms that hardware flow tracking is running.
+      2. For each specified flow tracker:
+        - Confirms that the tracker is active.
+        - Optionally, checks the tracker interval/timeout configuration.
+        - Optionally, verifies the tracker exporter configuration
 
     Expected Results
     ----------------
-    * Success: The test will pass if hardware flow tracking is running and an input tracker is active.
-    * Failure: The test will fail if hardware flow tracking is not running, an input tracker is not active,
-               or the tracker interval/timeout and exporter configuration does not match the expected values.
+    * Success: The test will pass if all of the following conditions are met:
+        - Hardware flow tracking is running.
+        - For each specified flow tracker:
+            - The flow tracker is active.
+            - The tracker interval/timeout matches the expected values, if provided.
+            - The exporter configuration matches the expected values, if provided.
+    * Failure: The test will fail if any of the following conditions are met:
+        - Hardware flow tracking is not running.
+        - For any specified flow tracker:
+            - The flow tracker is not active.
+            - The tracker interval/timeout does not match the expected values, if provided.
+            - The exporter configuration does not match the expected values, if provided.
 
     Examples
     --------
@@ -99,94 +91,51 @@ class VerifyHardwareFlowTrackerStatus(AntaTest):
     ```
     """
 
-    description = (
-        "Verifies if hardware flow tracking is running and an input tracker is active. Optionally verifies the tracker interval/timeout and exporter configuration."
-    )
     categories: ClassVar[list[str]] = ["flow tracking"]
-    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show flow tracking hardware tracker {name}", revision=1)]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show flow tracking hardware", revision=1)]
 
     class Input(AntaTest.Input):
         """Input model for the VerifyHardwareFlowTrackerStatus test."""
 
         trackers: list[FlowTracker]
         """List of flow trackers to verify."""
 
-        class FlowTracker(BaseModel):
-            """Detail of a flow tracker."""
-
-            name: str
-            """Name of the flow tracker."""
-
-            record_export: RecordExport | None = None
-            """Record export configuration for the flow tracker."""
-
-            exporters: list[Exporter] | None = None
-            """List of exporters for the flow tracker."""
-
-            class RecordExport(BaseModel):
-                """Record export configuration."""
-
-                on_inactive_timeout: int
-                """Timeout in milliseconds for exporting records when inactive."""
-
-                on_interval: int
-                """Interval in milliseconds for exporting records."""
-
-            class Exporter(BaseModel):
-                """Detail of an exporter."""
-
-                name: str
-                """Name of the exporter."""
-
-                local_interface: str
-                """Local interface used by the exporter."""
-
-                template_interval: int
-                """Template interval in milliseconds for the exporter."""
-
-    def render(self, template: AntaTemplate) -> list[AntaCommand]:
-        """Render the template for each hardware tracker."""
-        return [template.render(name=tracker.name) for tracker in self.inputs.trackers]
-
     @skip_on_platforms(["cEOSLab", "vEOS-lab"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyHardwareFlowTrackerStatus."""
         self.result.is_success()
-        for command, tracker_input in zip(self.instance_commands, self.inputs.trackers):
-            hardware_tracker_name = command.params.name
-            record_export = tracker_input.record_export.model_dump() if tracker_input.record_export else None
-            exporters = [exporter.model_dump() for exporter in tracker_input.exporters] if tracker_input.exporters else None
-            command_output = command.json_output
 
-            # Check if hardware flow tracking is configured
-            if not command_output.get("running"):
-                self.result.is_failure("Hardware flow tracking is not running.")
-                return
+        command_output = self.instance_commands[0].json_output
+        # Check if hardware flow tracking is configured
+        if not command_output.get("running"):
+            self.result.is_failure("Hardware flow tracking is not running.")
+            return
 
+        for tracker in self.inputs.trackers:
             # Check if the input hardware tracker is configured
-            tracker_info = command_output["trackers"].get(hardware_tracker_name)
-            if not tracker_info:
-                self.result.is_failure(f"Hardware flow tracker `{hardware_tracker_name}` is not configured.")
+            if not (tracker_info := get_value(command_output["trackers"], f"{tracker.name}")):
+                self.result.is_failure(f"{tracker} - Not found")
                 continue
 
             # Check if the input hardware tracker is active
             if not tracker_info.get("active"):
-                self.result.is_failure(f"Hardware flow tracker `{hardware_tracker_name}` is not active.")
+                self.result.is_failure(f"{tracker} - Disabled")
                 continue
 
             # Check the input hardware tracker timeouts
-            failure_msg = ""
-            if record_export:
-                record_export_failure = validate_record_export(record_export, tracker_info)
-                if record_export_failure:
-                    failure_msg += record_export_failure
-
-            # Check the input hardware tracker exporters' configuration
-            if exporters:
-                exporters_failure = validate_exporters(exporters, tracker_info)
-                if exporters_failure:
-                    failure_msg += exporters_failure
-
-            if failure_msg:
-                self.result.is_failure(f"{hardware_tracker_name}: {failure_msg}\n")
+            if tracker.record_export:
+                inactive_interval = tracker.record_export.on_inactive_timeout
+                on_interval = tracker.record_export.on_interval
+                act_inactive = tracker_info.get("inactiveTimeout")
+                act_interval = tracker_info.get("activeInterval")
+                if not all([inactive_interval == act_inactive, on_interval == act_interval]):
+                    self.result.is_failure(
+                        f"{tracker}, {tracker.record_export} - Incorrect timers - Inactive Timeout: {act_inactive}, OnActive Interval: {act_interval}"
+                    )
+
+            # Check the input hardware tracker exporters configuration
+            if tracker.exporters:
+                failure_messages = validate_exporters(tracker.exporters, tracker_info)
+                for message in failure_messages:
+                    self.result.is_failure(f"{tracker}, {message}")

docs/api/tests.flow_tracking.md

@@ -7,6 +7,8 @@ anta_title: ANTA catalog for flow tracking tests
   ~ that can be found in the LICENSE file.
   -->
 
+# Tests
+
 ::: anta.tests.flow_tracking
     options:
       show_root_heading: false
@@ -18,3 +20,19 @@ anta_title: ANTA catalog for flow tracking tests
       filters:
         - "!test"
         - "!render"
+        - "!validate_exporters"
+
+# Input models
+
+::: anta.input_models.flow_tracking
+
+    options:
+      show_root_heading: false
+      show_root_toc_entry: false
+      show_bases: false
+      anta_hide_test_module_description: true
+      merge_init_into_class: false
+      show_labels: true
+      filters:
+        - "!^__init__"
+        - "!^__str__"

examples/tests.yaml

@@ -162,7 +162,7 @@ anta.tests.field_notices:
       # Verifies if the device is exposed to FN0072, and if the issue has been mitigated.
 anta.tests.flow_tracking:
   - VerifyHardwareFlowTrackerStatus:
-      # Verifies if hardware flow tracking is running and an input tracker is active. Optionally verifies the tracker interval/timeout and exporter configuration.
+      # Verifies the hardware flow tracking state.
       trackers:
         - name: FLOW-TRACKER
           record_export: