Add q3d and portchannel support for strict priority.

Signed-off-by: Abhishek <abhishek@nexthop.ai>

Author: abhishek-nexthop

tests/common/devices/sonic.py

@@ -1838,6 +1838,7 @@ def _try_get_brcm_asic_name(self, output):
             "th3": {"b98", "BCM5698"},
             "th4": {"b99", "BCM5699"},
             "th5": {"f90", "BCM7890"},
+            "q3d": {"8870", "8872"},
         }
         for asic in search_sets.keys():
             for search_term in search_sets[asic]:

tests/common/helpers/ptf_tests_helper.py

@@ -9,25 +9,168 @@
 
 from ipaddress import ip_address, IPv4Address
 from tests.common.config_reload import config_reload
+from tests.common.utilities import wait_until
 
 logger = logging.getLogger(__name__)
 
 
+# ============================================================================
+# Helper Functions for PTF Port Mapping and Interface Selection
+# ============================================================================
+
 def get_dut_to_ptf_port_mapping(duthost, tbinfo):
     """
-    Get mapping of DUT interfaces to PTF ports.
+    Get mapping of DUT interfaces/PortChannels to PTF ports.
+    Only includes interfaces or PortChannels that have IP addresses configured.
 
     Args:
         duthost: DUT host object
         tbinfo: Testbed info
 
     Returns:
-        dict: Mapping of DUT interface names to PTF port indices
+        dict: {interface_name: ptf_index} for interfaces with IP
+              {portchannel_name: -1} for PortChannels with IP
     """
     mg_facts = duthost.get_extended_minigraph_facts(tbinfo)
-    return mg_facts.get('minigraph_ptf_indices', {})
+    all_indices = mg_facts.get('minigraph_ptf_indices', {})
+    mapping = {}
+
+    # Add interfaces with IP addresses
+    for intf in mg_facts.get('minigraph_interfaces', []):
+        interface_name = intf.get('attachto')
+        if interface_name and intf.get('addr'):
+            ptf_index = all_indices.get(interface_name)
+            if ptf_index is not None:
+                mapping[interface_name] = ptf_index
+
+    # Add PortChannels with IP addresses (PTF index = -1)
+    for intf in mg_facts.get('minigraph_portchannel_interfaces', []):
+        portchannel_name = intf.get('attachto')
+        if portchannel_name and intf.get('addr'):
+            mapping[portchannel_name] = -1
+
+    return mapping
+
+
+def get_interface_ip_address(interface_name, mg_facts):
+    """
+    Get IP address for an interface from minigraph facts.
+
+    Args:
+        interface_name: Interface or PortChannel name
+        mg_facts: Minigraph facts dictionary
+
+    Returns:
+        str: IP address or None if not found
+    """
+    # Check in minigraph_interfaces
+    for intf in mg_facts.get('minigraph_interfaces', []):
+        if intf.get('attachto') == interface_name and intf.get('addr'):
+            return str(intf['addr'])
+
+    # Check in minigraph_portchannel_interfaces
+    for intf in mg_facts.get('minigraph_portchannel_interfaces', []):
+        if intf.get('attachto') == interface_name and intf.get('addr'):
+            return str(intf['addr'])
+
+    return None
 
 
+def select_test_interface_and_ptf_port(duthost, tbinfo):
+    """
+    Select a random test interface/PortChannel and its corresponding PTF port.
+
+    Args:
+        duthost: DUT host object
+        tbinfo: Testbed info
+
+    Returns:
+        tuple: (interface_name, ptf_index) for interfaces
+               (portchannel_name, -1) for PortChannels
+               (None, None) if not found
+    """
+    dut_to_ptf_mapping = get_dut_to_ptf_port_mapping(duthost, tbinfo)
+    if not dut_to_ptf_mapping:
+        return None, None
+
+    interface_name = random.choice(list(dut_to_ptf_mapping.keys()))
+    ptf_port_index = dut_to_ptf_mapping[interface_name]
+
+    logger.info("Selected: {} (PTF port: {})".format(interface_name, ptf_port_index))
+    return interface_name, ptf_port_index
+
+
+def detect_portchannel_egress_member(duthost, tbinfo, ptf_adapter, portchannel_name, test_packet):
+    """
+    Detect which PortChannel member interface is actually used for egress traffic.
+
+    Args:
+        duthost: DUT host object
+        tbinfo: Testbed info
+        ptf_adapter: PTF adapter object
+        portchannel_name: PortChannel name
+        test_packet: Test packet to send
+
+    Returns:
+        tuple: (member_interface, ptf_port) or (None, None)
+    """
+    import ptf.testutils as testutils
+
+    logger.info("Detecting egress member for {}".format(portchannel_name))
+
+    # Get PortChannel members
+    mg_facts = duthost.get_extended_minigraph_facts(tbinfo)
+    portchannels = mg_facts.get('minigraph_portchannels', {})
+
+    if portchannel_name not in portchannels:
+        return None, None
+
+    members = portchannels[portchannel_name].get('members', [])
+    if not members:
+        return None, None
+
+    # Get PTF ports for members
+    ptf_indices = mg_facts.get('minigraph_ptf_indices', {})
+    member_ptf_ports = [(m, ptf_indices[m]) for m in members if m in ptf_indices]
+
+    if not member_ptf_ports:
+        return None, None
+
+    # Try each member port
+    num_test_packets = 100
+    timeout = 10.0
+
+    for member_name, member_ptf_port in member_ptf_ports:
+        logger.info("Trying {} (PTF port {})".format(member_name, member_ptf_port))
+        ptf_adapter.dataplane.flush()
+
+        # Send test packets
+        for i in range(num_test_packets):
+            testutils.send_packet(ptf_adapter, member_ptf_port, test_packet)
+
+        # Check if packets received
+        packets_received = 0
+
+        def check_packets_received():
+            nonlocal packets_received
+            while True:
+                result = testutils.dp_poll(ptf_adapter, device_number=0, timeout=0.1)
+                if isinstance(result, ptf_adapter.dataplane.PollSuccess):
+                    if result.port == member_ptf_port:
+                        packets_received += 1
+                else:
+                    break
+            return packets_received >= num_test_packets
+
+        if wait_until(timeout=timeout, interval=1, delay=1, condition=check_packets_received):
+            logger.info("Found egress member: {} (PTF port {})".format(member_name, member_ptf_port))
+            ptf_adapter.dataplane.flush()
+            return member_name, member_ptf_port
+
+    # No working member found
+    ptf_adapter.dataplane.flush()
+    return None, None
+
 @pytest.fixture(scope="module")
 def downstream_links(rand_selected_dut, tbinfo, nbrhosts):
     """

tests/qos/files/strict_priority_params.json

@@ -3,13 +3,14 @@
         "queue": 0,
         "cir_bytes_per_sec": 500000,
         "pir_bytes_per_sec": 750000,
-        "traffic_duration": 100,
+        "traffic_duration": 120,
         "low_traffic_bytes_per_sec": 500000,
         "high_traffic_bytes_per_sec": 1500000,
         "drop_threshold_low": 1000,
         "drop_threshold_high": 10000,
         "bandwidth_tolerance_min": 90,
         "scheduler_policy": "strict_priority_rate_limit_policy",
-        "packet_size": 1000
+        "packet_size": 1000,
+        "dnx_credit_worth": 10000
     }
 }

tests/qos/qos_helpers.py

@@ -135,34 +135,6 @@ def get_dscp_to_queue_mapping(duthost):
         return None
 
 
-def find_dscp_for_queue(duthost, target_queue):
-    """
-    Find a DSCP value that maps to the target queue
-
-    Args:
-        duthost: DUT host object
-        target_queue: Target queue number
-
-    Returns:
-        int or None: DSCP value that maps to target queue, or None if not found
-    """
-    # Get DSCP to queue mapping from DUT
-    dscp_to_queue_map = get_dscp_to_queue_mapping(duthost)
-    if dscp_to_queue_map is None:
-        logger.error("Could not get DSCP to queue mapping from DUT")
-        return None
-
-    for dscp, queue in dscp_to_queue_map.items():
-        if queue == target_queue:
-            logger.info(f"Found DSCP {dscp} maps to target queue {target_queue}")
-            return dscp
-
-    # If no exact match found, log available mappings and return None
-    available_mappings = {f"DSCP {dscp}": f"Queue {queue}" for dscp, queue in dscp_to_queue_map.items()}
-    logger.error(f"No DSCP found that maps to queue {target_queue}. Available mappings: {available_mappings}")
-    return None
-
-
 def get_phy_intfs(host_ans):
     """
     @Summary: Get the physical interfaces (e.g., EthernetX) of a DUT