Fixed the failure of test_valid_pfc_frame_with_snappi.py in Cisco DUT

tests/common/snappi_tests/read_pcap.py

@@ -3,6 +3,7 @@
 from dpkt.utils import mac_to_str
 
 from tests.common.snappi_tests.pfc_packet import PFCPacket
+from tests.snappi_tests.pfc.files.cisco_pfc_packet import CiscoPFCPacket
 
 logger = logging.getLogger(__name__)
 
@@ -62,6 +63,63 @@ def validate_pfc_frame(pfc_pcap_file, SAMPLE_SIZE=15000, UTIL_THRESHOLD=0.8):
     return True, None
 
 
+def validate_pfc_frame_cisco(pfc_pcap_file, SAMPLE_SIZE=15000, UTIL_THRESHOLD=0.8, peer_mac_addr=None):
+    """
+    Validate PFC frame by checking the CBFC opcode, class enable vector and class pause times.
+
+    Args:
+        pfc_cap: PFC pcap file
+        SAMPLE_SIZE: number of packets to sample
+        UTIL_THRESHOLD: threshold for PFC utilization to check if enough PFC frames were sent
+
+    Returns:
+        True if valid PFC frame, False otherwise
+    """
+    f = open(pfc_pcap_file, "rb")
+    pcap = dpkt.pcapng.Reader(f)
+    seen_non_zero_cev = False  # Flag for checking if any PFC frame has non-zero class enable vector
+
+    curPktCount = 0
+    curPFCXoffPktCount = 0
+    for _, buf in pcap:
+        if curPFCXoffPktCount >= SAMPLE_SIZE:
+            break
+        eth = dpkt.ethernet.Ethernet(buf)
+        if eth.type == PFC_MAC_CONTROL_CODE:
+            dest_mac = mac_to_str(eth.dst)
+            if dest_mac.lower() != PFC_DEST_MAC:
+                return False, "Destination MAC address is not 01:80:c2:00:00:01"
+            if peer_mac_addr:
+                src_mac = mac_to_str(eth.src)
+                if src_mac.lower() != peer_mac_addr:
+                    return False, "Source MAC address is not the peer's mac address"
+            pfc_packet = CiscoPFCPacket(pfc_frame_bytes=bytes(eth.data))
+            if not pfc_packet.is_valid():
+                logger.info("PFC frame {} is not valid. Please check the capture file.".format(curPktCount))
+                return False, "PFC frame is not valid"
+            cev = [int(i) for i in pfc_packet.class_enable_vec]
+            seen_non_zero_cev = True if sum(cev) > 0 else seen_non_zero_cev
+            if seen_non_zero_cev:
+                curPFCXoffPktCount += 1
+        curPktCount += 1
+
+    if not seen_non_zero_cev:
+        logger.info("No PFC frames with non-zero class enable vector found in the capture file.")
+        return False, "No PFC frames with non-zero class enable vector found"
+
+    f.close()
+    pfc_util = curPktCount / SAMPLE_SIZE
+
+    if curPktCount == 0:
+        logger.info("No PFC frames found in the capture file.")
+        return False, "No PFC frames found in the capture file"
+    elif pfc_util < UTIL_THRESHOLD:
+        logger.info("PFC utilization is too low. Please check the capture file.")
+        return False, "PFC utilization is too low"
+
+    return True, None
+
+
 def get_ipv4_pkts(pcap_file_name, protocol_num=61):
     """
     Get IPv4 packets from the pcap/pcapng file

tests/snappi_tests/pfc/files/cisco_pfc_packet.py

@@ -0,0 +1,32 @@
+"""
+The CiscoPFCPacket module  handles Cisco specific PFC frame checks
+"""
+from tests.common.snappi_tests.pfc_packet import PFCPacket, PRIO_DEFAULT_LEN
+
+
+class CiscoPFCPacket(PFCPacket):
+    def __init__(self, pfc_frame_bytes=None, cbfc_opcode=None, class_enable_vec=None, class_pause_times=None):
+        """
+        Initialize the PFCPacket base class
+
+        """
+        super().__init__(pfc_frame_bytes, cbfc_opcode, class_enable_vec, class_pause_times)
+
+    def _check_class_pause_times(self):
+        """
+        Check if class pause times are valid. Both conditions must be met:
+        1) class pause times are between 0x0 and 0xFFFF
+        2) class pause times are 0 if the corresponding bit in the class enable vector is 0, and vice versa
+
+        Args:
+
+        Returns:
+            True if valid class pause times, False otherwise
+        """
+        for i in range(len(self.class_pause_times)):
+            if self.class_pause_times[i] < 0x0 or self.class_pause_times[i] > 0xFFFF:
+                return False
+            elif self.class_pause_times[i] == 0x0 and self.class_enable_vec[PRIO_DEFAULT_LEN - i - 1] == "1":
+                return False
+
+        return True

tests/snappi_tests/pfc/files/helper.py

@@ -18,7 +18,8 @@
     verify_in_flight_buffer_pkts, verify_unset_cev_pause_frame_count, verify_tx_frame_count_dut, \
     verify_rx_frame_count_dut
 from tests.common.snappi_tests.snappi_test_params import SnappiTestParams
-from tests.common.snappi_tests.read_pcap import validate_pfc_frame
+from tests.common.cisco_data import is_cisco_device
+from tests.common.snappi_tests.read_pcap import validate_pfc_frame, validate_pfc_frame_cisco
 
 
 logger = logging.getLogger(__name__)
@@ -191,7 +192,7 @@ def run_pfc_test(api,
         # PFC pause frame capture is not requested
         valid_pfc_frame_test = False
 
-    if valid_pfc_frame_test:
+    if valid_pfc_frame_test and not is_cisco_device(duthost):
         snappi_extra_params.traffic_flow_config.pause_flow_config["flow_dur_sec"] = DATA_FLOW_DURATION_SEC + \
             data_flow_delay_sec + SNAPPI_POLL_DELAY_SEC + PAUSE_FLOW_DUR_BASE_SEC
         snappi_extra_params.traffic_flow_config.pause_flow_config["flow_traffic_type"] = \
@@ -245,7 +246,11 @@ def run_pfc_test(api,
 
     # Verify PFC pause frames
     if valid_pfc_frame_test:
-        is_valid_pfc_frame, error_msg = validate_pfc_frame(snappi_extra_params.packet_capture_file + ".pcapng")
+        if not is_cisco_device(duthost):
+            is_valid_pfc_frame, error_msg = validate_pfc_frame(snappi_extra_params.packet_capture_file + ".pcapng")
+        else:
+            is_valid_pfc_frame, error_msg = validate_pfc_frame_cisco(
+                                                              snappi_extra_params.packet_capture_file + ".pcapng")
         pytest_assert(is_valid_pfc_frame, error_msg)
         return