[202503] Fix test_xon_not_counted.py fanout port selection

tests/common/platform/device_utils.py

@@ -120,6 +120,41 @@ def list_dut_fanout_connections(dut, fanouthosts):
     return candidates
 
 
+def eos_to_linux_intf(eos_intf_name, hwsku=None):
+    """
+    @Summary: Map EOS's interface name to Linux's interface name
+    @param eos_intf_name: Interface name in EOS
+    @return: Return the interface name in Linux
+    """
+    if hwsku == "MLNX-OS":
+        linux_intf_name = eos_intf_name.replace(
+            "ernet 1/", "sl1p").replace("/", "sp")
+    elif hwsku and "Nokia" in hwsku:
+        linux_intf_name = eos_intf_name
+    else:
+        linux_intf_name = eos_intf_name.replace(
+            'Ethernet', 'et').replace('/', '_')
+    return linux_intf_name
+
+
+def nxos_to_linux_intf(nxos_intf_name):
+    """
+        @Summary: Map NxOS's interface name to Linux's interface name
+        @param nxos_intf_name: Interface name in NXOS
+        @return: Return the interface name in Linux
+    """
+    return nxos_intf_name.replace('Ethernet', 'Eth').replace('/', '-')
+
+
+def sonic_to_linux_intf(sonic_intf_name):
+    """
+    @Summary: Map SONiC's interface name to Linux's interface name
+    @param sonic_intf_name: Interface name in SONiC
+    @return: Return the interface name in Linux
+    """
+    return sonic_intf_name
+
+
 def watch_system_status(dut):
     """
     Watch DUT's system status

tests/pfcwd/test_xon_not_counted.py

@@ -1,24 +1,54 @@
 # tests/counter/test_xon_xoff.py
 import time
 import pytest
+from tests.common.helpers.assertions import pytest_assert
+from tests.common.fixtures.conn_graph_facts import fanout_graph_facts  # noqa F401
+from tests.common.platform.device_utils import eos_to_linux_intf, nxos_to_linux_intf, sonic_to_linux_intf
 
 pytestmark = [
     pytest.mark.topology("lt2", "ft2")
 ]
 
 
 @pytest.fixture(scope="module")
-def setup_fanouthost(duthosts, rand_one_dut_hostname, fanouthosts, conn_graph_facts, tbinfo):
+def setup_fanouthost(duthosts, rand_one_dut_hostname, fanouthosts, conn_graph_facts,
+                     fanout_graph_facts, tbinfo):  # noqa F811
     """
     A module level fixture to setup fanout host to send XON frames.
     """
     duthost = duthosts[rand_one_dut_hostname]
-    mg_facts = duthost.get_extended_minigraph_facts(tbinfo)
-    dut_port = list(mg_facts['minigraph_ptf_indices'].keys())[0]
+    int_status = duthost.show_interface(command="status")['ansible_facts']['int_status']
+    active_phy_intfs = [
+        intf for intf in int_status
+        if intf.startswith('Ethernet') and
+        int_status[intf]['admin_state'] == 'up' and
+        int_status[intf]['oper_state'] == 'up'
+    ]
     neighbors = conn_graph_facts["device_conn"].get(duthost.hostname, {})
-    peer_device = neighbors.get(dut_port, {}).get("peerdevice", "")
-    fanout_port = neighbors.get(dut_port, {}).get("peerport", "")
-    fanouthost = fanouthosts[peer_device]
+
+    dut_port = None
+    fanout_port = None
+    fanouthost = None
+    for intf in active_phy_intfs:
+        peer_device = neighbors.get(intf, {}).get("peerdevice", "")
+        peer_port = neighbors.get(intf, {}).get("peerport", "")
+        if peer_device in fanouthosts and peer_port:
+            fanouthost = fanouthosts[peer_device]
+            fanout_os = fanouthost.get_fanout_os()
+            fanout_hwsku = fanout_graph_facts[fanouthost.hostname]["device_info"]["HwSku"]
+            if fanout_os == "nxos":
+                fanout_port = nxos_to_linux_intf(peer_port)
+            elif fanout_os == "sonic":
+                fanout_port = sonic_to_linux_intf(peer_port)
+            else:
+                fanout_port = eos_to_linux_intf(peer_port, hwsku=fanout_hwsku)
+            dut_port = intf
+            break
+
+    pytest_assert(
+        dut_port is not None and fanout_port is not None and fanouthost is not None,
+        "No active DUT port with valid fanout connection found"
+    )
 
     # Copy pfc_gen.py to fanout host /tmp
     # Since the test is to verify XON frames are not counted, we do not need to

tests/qos/qos_helpers.py

@@ -34,41 +34,6 @@ def ansible_stdout_to_str(ansible_stdout):
     return result
 
 
-def eos_to_linux_intf(eos_intf_name, hwsku=None):
-    """
-    @Summary: Map EOS's interface name to Linux's interface name
-    @param eos_intf_name: Interface name in EOS
-    @return: Return the interface name in Linux
-    """
-    if hwsku == "MLNX-OS":
-        linux_intf_name = eos_intf_name.replace(
-            "ernet 1/", "sl1p").replace("/", "sp")
-    elif hwsku and "Nokia" in hwsku:
-        linux_intf_name = eos_intf_name
-    else:
-        linux_intf_name = eos_intf_name.replace(
-            'Ethernet', 'et').replace('/', '_')
-    return linux_intf_name
-
-
-def nxos_to_linux_intf(nxos_intf_name):
-    """
-        @Summary: Map NxOS's interface name to Linux's interface name
-        @param nxos_intf_name: Interface name in NXOS
-        @return: Return the interface name in Linux
-    """
-    return nxos_intf_name.replace('Ethernet', 'Eth').replace('/', '-')
-
-
-def sonic_to_linux_intf(sonic_intf_name):
-    """
-    @Summary: Map SONiC's interface name to Linux's interface name
-    @param sonic_intf_name: Interface name in SONiC
-    @return: Return the interface name in Linux
-    """
-    return sonic_intf_name
-
-
 def get_phy_intfs(host_ans):
     """
     @Summary: Get the physical interfaces (e.g., EthernetX) of a DUT

tests/qos/test_pfc_counters.py

@@ -1,6 +1,6 @@
 from tests.common.fixtures.conn_graph_facts import conn_graph_facts, fanout_graph_facts     # noqa F401
+from tests.common.platform.device_utils import eos_to_linux_intf, nxos_to_linux_intf, sonic_to_linux_intf
 from .qos_fixtures import leaf_fanouts      # noqa F401
-from .qos_helpers import eos_to_linux_intf, nxos_to_linux_intf, sonic_to_linux_intf
 import os
 import time
 import pytest