[PFC] Test if PFC can pause lossless priorities

ansible/roles/test/files/ptftests/pfc_pause_test.py

@@ -0,0 +1,149 @@
+import ipaddress
+import logging
+import random
+import socket
+import sys
+import struct
+import ipaddress
+import re
+
+import ptf
+import ptf.packet as scapy
+import ptf.dataplane as dataplane
+
+from ptf import config
+from ptf.base_tests import BaseTest
+from ptf.mask import Mask
+from ptf.testutils import add_filter, reset_filters, dp_poll, simple_udp_packet, send_packet, test_params_get
+
+def udp_filter(pkt_str):
+    try:
+        pkt = scapy.Ether(pkt_str)
+        return scapy.UDP in pkt
+
+    except:
+        return False
+
+def capture_matched_packets(test, exp_packet, port, device_number=0, timeout=1):
+    """
+    Receive all packets on the port and return all the received packets.
+    As soon as the packets stop arriving, the function waits for the timeout value and returns the received packets. Therefore, this function requires a positive timeout value.
+    """
+    if timeout <= 0:
+        raise Exception("%s() requires positive timeout value." % sys._getframe().f_code.co_name)
+
+    pkts = list()
+    while True:
+        result = dp_poll(test, device_number=device_number, port_number=port, timeout=timeout)
+        if isinstance(result, test.dataplane.PollSuccess):
+            if ptf.dataplane.match_exp_pkt(exp_packet, result.packet):
+                pkts.append(result.packet)
+        else:
+            break
+
+    return pkts 
+
+class PfcPauseTest(BaseTest):
+    def __init__(self):
+        BaseTest.__init__(self)
+        self.test_params = test_params_get()
+
+    def setUp(self):
+        add_filter(udp_filter)
+        self.dataplane = ptf.dataplane_instance
+        self.mac_src = self.test_params['mac_src']
+        self.mac_dst = self.test_params['mac_dst']
+        self.pkt_count = int(self.test_params['pkt_count'])
+        self.pkt_intvl = float(self.test_params['pkt_intvl']) 
+        self.port_src = int(self.test_params['port_src'])
+        self.port_dst = self.test_params['port_dst']
+        self.ip_src = self.test_params['ip_src']
+        self.ip_dst = self.test_params['ip_dst']
+        self.dscp = self.test_params['dscp']
+        """ DSCP for background traffic """
+        self.dscp_bg = self.test_params['dscp_bg']
+        self.queue_paused = self.test_params['queue_paused']
+        """ if DUT has MAC information """
+        self.dut_has_mac = self.test_params['dut_has_mac']
+
+    def runTest(self):
+        pass_cnt = 0
+        tos = self.dscp<<2
+        tos_bg = self.dscp_bg<<2
+
+        """ If DUT needs to learn MAC addresses """
+        if not self.dut_has_mac:            
+            pkt = simple_udp_packet(
+                eth_dst=self.mac_dst,
+                eth_src=self.mac_src,
+                ip_src=self.ip_src,
+                ip_dst=self.ip_dst)
+
+            send_packet(self, self.port_src, pkt, 5)
+
+            pkt = simple_udp_packet(
+                eth_dst=self.mac_src,
+                eth_src=self.mac_dst,
+                ip_src=self.ip_dst,
+                ip_dst=self.ip_src)
+
+            send_packet(self, self.port_dst, pkt, 5)
+
+        for x in range(self.pkt_count):
+            sport = random.randint(0, 65535)
+            dport = random.randint(0, 65535)
+
+            pkt = simple_udp_packet(
+                        eth_dst=self.mac_dst,
+                        eth_src=self.mac_src,
+                        ip_src=self.ip_src,
+                        ip_dst=self.ip_dst,
+                        ip_tos=tos,
+                        udp_sport=sport,
+                        udp_dport=dport,
+                        ip_ttl=64)
+
+            pkt_bg = simple_udp_packet(
+                        eth_dst=self.mac_dst,
+                        eth_src=self.mac_src,
+                        ip_src=self.ip_src,
+                        ip_dst=self.ip_dst,
+                        ip_tos=tos_bg,
+                        udp_sport=sport,
+                        udp_dport=dport,
+                        ip_ttl=64)
+
+            exp_pkt = simple_udp_packet(
+                        ip_src=self.ip_src,
+                        ip_dst=self.ip_dst,
+                        ip_tos=tos_bg,
+                        udp_sport=sport,
+                        udp_dport=dport,
+                        ip_ttl=63)
+
+            masked_exp_pkt = Mask(exp_pkt)
+            masked_exp_pkt.set_do_not_care_scapy(scapy.Ether, "src")
+            masked_exp_pkt.set_do_not_care_scapy(scapy.Ether, "dst")
+            masked_exp_pkt.set_do_not_care_scapy(scapy.IP, "ttl")
+            masked_exp_pkt.set_do_not_care_scapy(scapy.IP, "chksum")
+            masked_exp_pkt.set_do_not_care_scapy(scapy.IP, "tos")
+
+            send_packet(self, self.port_src, pkt, 1)
+            send_packet(self, self.port_src, pkt_bg, 1)
+
+            pkts = capture_matched_packets(self, masked_exp_pkt, self.port_dst)
+
+            time.sleep(self.pkt_intvl)
+
+            """ If the queue is paused, we should only receive the background packet """
+            if self.queue_paused:
+                pass_cnt += int(len(pkts) == 1 and scapy.Ether(pkts[0])[scapy.IP].tos == tos_bg)
+
+            else:
+                pass_cnt += int(len(pkts) == 2)
+
+        print "Passes: %d / %d" % (pass_cnt, self.pkt_count)
+
+    def tearDown(self):   
+        reset_filters()
+        BaseTest.tearDown(self)

tests/qos/qos_fixtures.py

@@ -1,5 +1,37 @@
 import pytest
 import os
+from ansible_host import AnsibleHost
+
+@pytest.fixture(scope = "module")
+def lossless_prio_dscp_map(testbed_devices):
+    dut = testbed_devices["dut"]
+    config_facts = dut.config_facts(host=dut.hostname, source="persistent")['ansible_facts']
+
+    if "PORT_QOS_MAP" not in config_facts.keys():
+        return None 
+
+    port_qos_map = config_facts["PORT_QOS_MAP"]
+    lossless_priorities = list()
+    intf = port_qos_map.keys()[0]
+    if 'pfc_enable' not in port_qos_map[intf]:
+        return None 
+
+    lossless_priorities = [int(x) for x in port_qos_map[intf]['pfc_enable'].split(',')]
+    dscp_to_tc_map = config_facts["DSCP_TO_TC_MAP"]
+
+    result = dict()
+    for prio in lossless_priorities:
+        result[prio] = list()
+
+    profile = dscp_to_tc_map.keys()[0]
+
+    for dscp in dscp_to_tc_map[profile]:
+        tc = dscp_to_tc_map[profile][dscp]
+
+        if int(tc) in lossless_priorities:
+            result[int(tc)].append(int(dscp))
+
+    return result 
 
 @pytest.fixture(scope = "module")
 def conn_graph_facts(testbed_devices):
@@ -33,4 +65,4 @@ def leaf_fanouts(conn_graph_facts):
         if peer_device not in leaf_fanouts:
             leaf_fanouts.append(peer_device)
 
-    return leaf_fanouts
+    return leaf_fanouts