two chagnes:

sonic-net#1 Deprecated dynamic subclass instantiation based on attributes implement in find_subclass().
Switched to static lookup table to avoid syntax errors in one platform affecting others.

Signed-off-by: Xu Chen <xuchen3@microsoft.com>
sonic-net#2 remove QosHelper() class, since not observe benefit for this abstraction

tests/saitests/refactor/counter_collector.py

@@ -11,7 +11,7 @@
 
 from platform_qos_base import PlatformQosBase
 from topology_qos_base import TopologyQosBase
-from qos_helper import QosHelper, instantiate_helper, log_message
+from qos_helper import instantiate_helper, log_message
 import texttable
 
 

tests/saitests/refactor/platform_qos_base.py

@@ -56,7 +56,7 @@ def construct_ip_pkt(pkt_len, dst_mac, src_mac, src_ip, dst_ip, dscp, src_vlan,
         return pkt
 
 
-from qos_helper import get_case, get_platform, get_topology, log_message
+from qos_helper import log_message
 
 
 class PlatformQosBase():
@@ -89,8 +89,7 @@ def fill_leakout(self, src_port_id, dst_port_id, packet, pg, asic_type, pkts_num
 
 
     def send_packet(self, port, packet, packet_number):
-        case = get_case(self)
-        send_packet(case, port, packet, packet_number)
+        send_packet(self.testcase, port, packet, packet_number)
 
 
     def compensate_leakout(self):

tests/saitests/refactor/platform_qos_cisco.py

@@ -102,7 +102,6 @@ def overflow_egress(test_case, src_port_id, pkt, queue, asic_type):
 
 
 from platform_qos_base import PlatformQosBase
-from qos_helper import get_case, get_platform, get_topology
 
 
 class PlatformQosCisco(PlatformQosBase):
@@ -114,22 +113,20 @@ class PlatformQosCisco(PlatformQosBase):
     #
 
     def disable_port_transmit(self, client, asic_type, port_list):
-        case = get_case(self)
         # generate pkts_num_egr_mem in runtime
-        if hasattr(case, 'src_dst_asic_diff') and case.src_dst_asic_diff:
-            case.sai_thrift_port_tx_disable(client, asic_type, port_list)
-            pkts_num_egr_mem, extra_bytes_occupied = overflow_egress(case, src_port_id, pkt,
-                                                                     int(case.test_params['pg']),
+        if hasattr(self.testcase, 'src_dst_asic_diff') and self.testcase.src_dst_asic_diff:
+            self.testcase.sai_thrift_port_tx_disable(client, asic_type, port_list)
+            pkts_num_egr_mem, extra_bytes_occupied = overflow_egress(self.testcase, src_port_id, pkt,
+                                                                     int(self.testcase.test_params['pg']),
                                                                      asic_type)
-            case.sai_thrift_port_tx_enable(client, asic_type, port_list)
+            self.testcase.sai_thrift_port_tx_enable(client, asic_type, port_list)
             time.sleep(2)
         super().disable_port_transmit(client, asic_type, port_list)
 
 
     def fill_leakout(self, src_port_id, dst_port_id, packet, pg, asic_type, pkts_num_egr_mem):
         # For some platform, prefer handle leakout before send_packet
-        case = get_case(self)
-        fill_leakout_plus_one(case, src_port_id, dst_port_id, packet, pg, asic_type, pkts_num_egr_mem)
+        fill_leakout_plus_one(self.testcase, src_port_id, dst_port_id, packet, pg, asic_type, pkts_num_egr_mem)
         # One extra packet is sent to fill the leakout, and number of extra packet is returned to the caller, 
         # so that the caller knows to send one less packet next time.
         return 1

tests/saitests/refactor/qos_helper.py

@@ -18,73 +18,94 @@ def log_message(message, level='info', to_stderr=False):
     log_fn(message)
 
 
-
-
-
-
-
 def find_subclass(base_class, target_name, attr_name="platform_name"):
     # Dynamically find a subclass of base_class where the attr_name matches target_name
     for subclass in base_class.__subclasses__():
         if target_name in getattr(subclass, attr_name, []):
             return subclass
     return base_class
 
-
-def associate_helper(instance, helper):
-    # Associate helper with an instance
-    instance.helper = helper
-
-
-def instantiate_helper(case_instance, hwsku_name, topology_name):
-    from platform_qos_base import PlatformQosBase
-    from topology_qos_base import TopologyQosBase
-    # Factory method to instantiate QosHelper with platform and topology
-    platform_class = find_subclass(PlatformQosBase, hwsku_name, attr_name="hwsku_name")
-    topology_class = find_subclass(TopologyQosBase, topology_name, attr_name="topology_name")
-    platform_instance = platform_class()
-    topology_instance = topology_class()
-    helper = QosHelper(case_instance, platform_instance, topology_instance)
-    associate_helper(platform_instance, helper)
-    associate_helper(topology_instance, helper)
-    return helper
-
-
 #
-# wrappers
+# Deprecated dynamic subclass instantiation based on attributes implement in find_subclass().
+# Switched to static lookup table to avoid syntax errors in one platform affecting others.
 #
 
-def get_case(instance):
-    helper = instance.helper if instance and hasattr(instance, 'helper') else None
-    return helper.case if helper and hasattr(helper, 'case') else None
-
+PLATFORM_MAPPING = {
+    'PlatformQosCisco': {
+        'file': 'platform_qos_cisco',
+        'supported_skus': ['Cisco-8102-C64'],
+    },
+    'PlatformQosBroadcom': {
+        'file': 'platform_qos_broadcom',
+        'supported_skus': [],
+    },
+    'PlatformQosTomhawk': {
+        'file': 'platform_qos_tomhawk',
+        'supported_skus': [],
+    },
+    'PlatformQosTh3': {
+        'file': 'platform_qos_th3',
+        'supported_skus': [],
+    },
+    'PlatformQosKvm': {
+        'file': 'platform_qos_kvm',
+        'supported_skus': ['kvm'],
+    },
+}
+
+
+TOPOLOGY_MAPPING = {
+    'TopologyQosTierOne': {
+        'file': 'topology_qos_tier_one',
+        'supported_topologies': ['t1'],
+    },
+    'TopologyQosTierOne64Lag': {
+        'file': 'topology_qos_tier_one_64lag',
+        'supported_topologies': ['t1-64-lag'],
+    },
+}
+
+
+def instantiate_helper(testcase_instance, hwsku_name, topology_name):
+    from platform_qos_base import PlatformQosBase
+    from topology_qos_base import TopologyQosBase
 
-def get_platform(instance):
-    helper = instance.helper if instance and hasattr(instance, 'helper') else None
-    return helper.platform if helper and hasattr(helper, 'platform') else None
+    # Factory method to instantiate platform and topology
+    # Previously, we used a dynamic approach to find and instantiate subclasses based on attributes.
+    # However, this approach had a significant drawback: if any module contained syntax errors or other issues,
+    # it could break the entire dynamic lookup process, causing failures even for unrelated platforms.
+    # To mitigate this issue, we switched to a static lookup table (mapping) approach.
 
+    platform_class_name = None
+    topology_class_name = None
 
-def get_topology(instance):
-    helper = instance.helper if instance and hasattr(instance, 'helper') else None
-    return helper.topology if helper and hasattr(helper, 'topology') else None
+    for class_name, info in PLATFORM_MAPPING.items():
+        if hwsku_name in info['supported_skus']:
+            platform_class_name = class_name
+            platform_file = info['file']
+            break
 
+    for class_name, info in TOPOLOGY_MAPPING.items():
+        if topology_name in info['supported_topologies']:
+            topology_class_name = class_name
+            topology_file = info['file']
+            break
 
-class QosHelper():
+    if not platform_class_name or not topology_class_name:
+        raise ValueError(f"Unsupported hwsku_name: {hwsku_name} or topology_name: {topology_name}")
 
-    def __init__(self, case, platform, topology):
-        self.case = case
-        self.platform = platform
-        self.topology = topology
+    platform_module = __import__(platform_file, fromlist=[platform_class_name])
+    topology_module = __import__(topology_file, fromlist=[topology_class_name])
 
+    platform_class = getattr(platform_module, platform_class_name)
+    topology_class = getattr(topology_module, topology_class_name)
 
-    #
-    # dynamic proxy
-    #
+    platform_instance = platform_class()
+    topology_instance = topology_class()
 
-    def __getattr__(self, name):
-        # Dynamic proxy to delegate calls to platform or topology based on method availability
-        if hasattr(self.platform, name):
-            return getattr(self.platform, name)
-        elif hasattr(self.topology, name):
-            return getattr(self.topology, name)
-        raise AttributeError(f"'{name}' not found in either platform or topology")
+    testcase_instance.platform = platform_instance
+    testcase_instance.topology = topology_instance
+    platform_instance.testcase = testcase_instance
+    platform_instance.topology = topology_instance
+    topology_instance.testcase = testcase_instance
+    topology_instance.platform = platform_instance

tests/saitests/refactor/sai_qos_tests.py

@@ -293,7 +293,7 @@ def step_short_of_pfc(self, port, packet, packet_number):
         # In previous step, we have already sent packets to fill leakout in some platform,
         # so in this step, we need to send ${leakout_overflow} less packet to trigger pfc
         #
-        self.helper.send_packet(port, packet, packet_number)
+        self.platform.send_packet(port, packet, packet_number)
         # allow enough time for the dut to sync up the counter values in counters_db
         time.sleep(8)
 
@@ -327,7 +327,7 @@ def step_check_short_of_pfc(self):
     @saitests_decorator(func=diag_counter, param='capture', enter=False, exit=True)
     def step_trigger_pfc(self):
         # send 1 packet to trigger pfc
-        self.helper.send_packet(self.src_port_id, self.pkt, 1 + 2 * self.pkts_num_margin)
+        self.platform.send_packet(self.src_port_id, self.pkt, 1 + 2 * self.pkts_num_margin)
         # allow enough time for the dut to sync up the counter values in counters_db
         time.sleep(8)
 
@@ -361,7 +361,7 @@ def step_check_trigger_pfc(self):
     @saitests_decorator(func=diag_counter, param='capture', enter=False, exit=True)
     def step_short_of_ingress_drop(self):
         # send packets short of ingress drop
-        self.helper.send_packet(self.src_port_id, self.pkt, (self.pkts_num_trig_ingr_drp -
+        self.platform.send_packet(self.src_port_id, self.pkt, (self.pkts_num_trig_ingr_drp -
                                 self.pkts_num_trig_pfc) // self.cell_occupancy - 1 - 2 * self.pkts_num_margin)
         # allow enough time for the dut to sync up the counter values in counters_db
         time.sleep(8)
@@ -396,7 +396,7 @@ def step_check_short_of_ingress_drop(self):
     @saitests_decorator(func=diag_counter, param='capture', enter=False, exit=True)
     def step_trigger_ingress_drop(self):
         # send 1 packet to trigger pfc
-        self.helper.send_packet(self.src_port_id, self.pkt, 1 + 2 * self.pkts_num_margin)
+        self.platform.send_packet(self.src_port_id, self.pkt, 1 + 2 * self.pkts_num_margin)
         # allow enough time for the dut to sync up the counter values in counters_db
         time.sleep(8)
         capture_diag_counter(self, 'TrigPfc')

tests/saitests/refactor/testcase_qos_base.py

@@ -8,7 +8,7 @@
 
 from platform_qos_base import PlatformQosBase
 from topology_qos_base import TopologyQosBase
-from qos_helper import QosHelper, instantiate_helper, log_message
+from qos_helper import instantiate_helper, log_message
 from counter_collector import CounterCollector, initialize_diag_counter, capture_diag_counter, summarize_diag_counter
 from saitests_decorators import saitests_decorator, diag_counter, step_result, step_banner
 
@@ -26,7 +26,7 @@ def setUp(self):
         switch_init(self.clients)
         hwsku_name = self.test_params.get('hwsku', 'Unknown HwSKU')
         topology_name = self.test_params.get('testbed_type', 'Unknown Topolog')
-        self.helper = instantiate_helper(self, hwsku_name, topology_name)
+        instantiate_helper(self, hwsku_name, topology_name)
 
 
     def tearDown(self):
@@ -52,32 +52,32 @@ def step_build_param(self):
         # to make testcase specific class see less platform related activity.
         # But, in some corner case, we also can invoke platform related function in 
         # testcase specific class, to make code simple and easy to understand.
-        self.helper.build_param()
+        self.platform.build_param()
 
         # todo: to move to platform specific class
         self.counter_margin = 0  # COUNTER_MARGIN = 2  # Margin for counter check
 
 
     def step_build_packet(self, pkt_len, dst_mac, src_mac, src_ip, dst_ip, dscp, src_vlan, **kwargs):
-        return self.helper.build_packet(pkt_len, dst_mac, src_mac, src_ip, dst_ip, dscp, src_vlan, **kwargs)
+        return self.platform.build_packet(pkt_len, dst_mac, src_mac, src_ip, dst_ip, dscp, src_vlan, **kwargs)
 
 
     @saitests_decorator(func=step_banner, param='Banner', enter=True, exit=False)
     @saitests_decorator(func=step_result, param='Result', enter=False, exit=True)
     @saitests_decorator(func=diag_counter, param='capture', enter=False, exit=True)
     def step_detect_rx_port(self):
-        return self.helper.detect_rx_port()
+        return self.topology.detect_rx_port()
 
 
     def step_disable_port_transmit(self, client, asic_type, port_list):
-        self.helper.disable_port_transmit(client, asic_type, port_list)
+        self.platform.disable_port_transmit(client, asic_type, port_list)
 
 
     @saitests_decorator(func=step_banner, param='Banner', enter=True, exit=False)
     @saitests_decorator(func=step_result, param='Result', enter=False, exit=True)
     @saitests_decorator(func=diag_counter, param='summarize', enter=True, exit=False)
     def step_enable_port_transmit(self, client, asic_type, port_list):
-        self.helper.enable_port_transmit(client, asic_type, port_list)
+        self.platform.enable_port_transmit(client, asic_type, port_list)
 
 
     @saitests_decorator(func=step_banner, param='Banner', enter=True, exit=False)
@@ -88,7 +88,7 @@ def step_fill_leakout(self, src_port_id, dst_port_id, packet, pg, asic_type, pkt
         # before sending packets short of triggering pfc
         # For some platform, prefer handle leakout before send_packet
         #
-        return self.helper.fill_leakout(src_port_id, dst_port_id, packet, pg, asic_type, pkts_num_egr_mem)
+        return self.platform.fill_leakout(src_port_id, dst_port_id, packet, pg, asic_type, pkts_num_egr_mem)
 
 
     @saitests_decorator(func=step_banner, param='Banner', enter=True, exit=False)
@@ -98,4 +98,4 @@ def step_compensate_leakout(self):
         #
         # For some platform, prefer to handler leackout after send_packet
         #
-        self.helper.compensate_leakout()
+        self.platform.compensate_leakout()

tests/saitests/refactor/topology_qos_base.py

@@ -7,7 +7,6 @@
 from ptf.testutils import send_packet
 
 from platform_qos_base import PlatformQosBase
-from qos_helper import get_case, get_platform, get_topology
 
 
 #
@@ -18,7 +17,7 @@ def get_rx_port(dp, device_number, src_port_id, dst_mac, dst_ip, src_ip, src_vla
     ip_id = 0xBABE
     src_port_mac = dp.dataplane.get_mac(device_number, src_port_id)
 
-    pkt = dp.helper.build_packet(64, dst_mac, src_port_mac, src_ip, dst_ip, 0, src_vlan, ip_id=ip_id)
+    pkt = dp.platform.build_packet(64, dst_mac, src_port_mac, src_ip, dst_ip, 0, src_vlan, ip_id=ip_id)
     # Send initial packet for any potential ARP resolution, which may cause the LAG
     # destination to change. Can occur especially when running tests in isolation on a
     # first test attempt.
@@ -27,7 +26,7 @@ def get_rx_port(dp, device_number, src_port_id, dst_mac, dst_ip, src_ip, src_vla
     time.sleep(1)
     send_packet(dp, src_port_id, pkt, 1)
 
-    masked_exp_pkt = dp.helper.build_packet(48, dst_mac, src_port_mac, src_ip, dst_ip, 0, src_vlan, ip_id=ip_id, exp_pkt=True)
+    masked_exp_pkt = dp.platform.build_packet(48, dst_mac, src_port_mac, src_ip, dst_ip, 0, src_vlan, ip_id=ip_id, exp_pkt=True)
 
     pre_result = dp.dataplane.poll(device_number=0, exp_pkt=masked_exp_pkt, timeout=3)
     result = dp.dataplane.poll(device_number=0, exp_pkt=masked_exp_pkt, timeout=3)
@@ -52,6 +51,6 @@ def populate_arp(self):
 
 
     def detect_rx_port(self):
-        case = get_case(self)
-        return get_rx_port(case, 0, case.src_port_id, case.pkt_dst_mac, case.dst_port_ip, case.src_port_ip, case.src_port_vlan)
+        return get_rx_port(self.testcase, 0, self.testcase.src_port_id, self.testcase.pkt_dst_mac,
+                           self.testcase.dst_port_ip, self.testcase.src_port_ip, self.testcase.src_port_vlan)