Skip to content

ANTA catalog for flow tracking tests

Tests

Module related to the flow tracking tests.

VerifyHardwareFlowTrackerStatus

Verifies the hardware flow tracking state.

This test performs the following checks:

  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 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 
anta.tests.flow_tracking:
  - VerifyHardwareFlowTrackerStatus:
      trackers:
        - name: FLOW-TRACKER
          record_export:
            on_inactive_timeout: 70000
            on_interval: 300000
          exporters:
            - name: CV-TELEMETRY
              local_interface: Loopback0
              template_interval: 3600000

Inputs

Name Type Description Default
trackers list[FlowTracker]
List of flow trackers to verify.
-
Source code in anta/tests/flow_tracking.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class VerifyHardwareFlowTrackerStatus(AntaTest):
    """Verifies the hardware flow tracking state.

    This test performs the following checks:

      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 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
    --------
    ```yaml
    anta.tests.flow_tracking:
      - VerifyHardwareFlowTrackerStatus:
          trackers:
            - name: FLOW-TRACKER
              record_export:
                on_inactive_timeout: 70000
                on_interval: 300000
              exporters:
                - name: CV-TELEMETRY
                  local_interface: Loopback0
                  template_interval: 3600000
    ```
    """

    categories: ClassVar[list[str]] = ["flow tracking"]
    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."""

    @skip_on_platforms(["cEOSLab", "vEOS-lab"])
    @AntaTest.anta_test
    def test(self) -> None:
        """Main test function for VerifyHardwareFlowTrackerStatus."""
        self.result.is_success()

        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
            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"{tracker} - Disabled")
                continue

            # Check the input hardware tracker timeouts
            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}")

Input models

Module containing input models for flow tracking tests.

Exporter

Model representing the exporter used for flow record export.

Name Type Description Default
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.
-
Source code in anta/input_models/flow_tracking.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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}"

FlowTracker

Flow Tracking model representing the tracker details.

Name Type Description Default
name str
The name of the flow tracker.
-
record_export RecordExport | None
Configuration for record export, specifying details about timeouts.
None
exporters list[Exporter] | None
A list of exporters associated with the flow tracker.
None
Source code in anta/input_models/flow_tracking.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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}"

RecordExport

Model representing the record export configuration for a flow tracker.

Name Type Description Default
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.
-
Source code in anta/input_models/flow_tracking.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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}"