Add a test case to validate FEC histogram

tests/common/platform/interface_utils.py

@@ -226,3 +226,44 @@ def get_dpu_npu_ports_from_hwsku(duthost):
             dpu_npu_port_list.append(intf)
     logging.info(f"DPU NPU ports in hwsku.json are {dpu_npu_port_list}")
     return dpu_npu_port_list
+
+
+def get_fec_eligible_interfaces(duthost, supported_speeds):
+    """
+    Get interfaces that are operationally up, SFP present and have supported speeds.
+
+    Args:
+        duthost: The device under test.
+        supported_speeds (list): A list of supported speeds for validation.
+
+    Returns:
+        interfaces (list): A list of interface names with SFP present, oper status up
+        and speed in supported_speeds.
+    """
+    logging.info("Get output of 'show interface status'")
+    intf_status = duthost.show_and_parse("show interface status")
+    logging.info("Interface status: {intf_status}")
+
+    logging.info("Get output of 'sudo sfpshow presence'")
+    sfp_presence_output = duthost.show_and_parse("sudo sfpshow presence")
+    logging.info("SFP presence: {sfp_presence_output}")
+
+    sfp_presence_dict = {entry['port']: entry.get('presence', '').lower() for entry in sfp_presence_output}
+
+    interfaces = []
+    for intf in intf_status:
+        intf_name = intf['interface']
+        presence = sfp_presence_dict.get(intf_name, '')
+
+        if presence != "present":
+            continue
+
+        oper = intf.get('oper', '').lower()
+        speed = intf.get('speed', '')
+
+        if oper == "up" and speed in supported_speeds:
+            interfaces.append(intf_name)
+        else:
+            logging.info(f"Skip for {intf_name}: oper_state:{oper} speed:{speed}")
+
+    return interfaces

tests/platform_tests/test_intf_fec.py

@@ -1,8 +1,10 @@
 import logging
 import pytest
+import time
 
 from tests.common.helpers.assertions import pytest_assert
 from tests.common.utilities import skip_release, wait_until
+from tests.common.platform.interface_utils import get_fec_eligible_interfaces
 
 pytestmark = [
     pytest.mark.disable_loganalyzer,  # disable automatic loganalyzer
@@ -49,22 +51,14 @@ def test_verify_fec_oper_mode(duthosts, enum_rand_one_per_hwsku_frontend_hostnam
     if "broadcom" in duthost.facts.get('platform_asic'):
         pytest.skip("Skipping this test on platforms with Broadcom ASICs")
 
-    logging.info("Get output of '{}'".format("show interface status"))
-    intf_status = duthost.show_and_parse("show interface status")
+    # Get interfaces that are operationally up and have supported speeds.
+    interfaces = get_fec_eligible_interfaces(duthost, SUPPORTED_SPEEDS)
 
-    for intf in intf_status:
-        sfp_presence = duthost.show_and_parse("sudo sfpshow presence -p {}"
-                                              .format(intf['interface']))
-        if sfp_presence:
-            presence = sfp_presence[0].get('presence', '').lower()
-            oper = intf.get('oper', '').lower()
-            speed = intf.get('speed', '')
-
-            if presence == "present" and oper == "up" and speed in SUPPORTED_SPEEDS:
-                # Verify the FEC operational mode is valid
-                fec = get_fec_oper_mode(duthost, intf['interface'])
-                if fec == "n/a":
-                    pytest.fail("FEC status is N/A for interface {}".format(intf['interface']))
+    for intf in interfaces:
+        # Verify the FEC operational mode is valid
+        fec = get_fec_oper_mode(duthost, intf)
+        if fec == "n/a":
+            pytest.fail("FEC status is N/A for interface {}".format(intf['interface']))
 
 
 def test_config_fec_oper_mode(duthosts, enum_rand_one_per_hwsku_frontend_hostname,
@@ -78,21 +72,11 @@ def test_config_fec_oper_mode(duthosts, enum_rand_one_per_hwsku_frontend_hostnam
     if "broadcom" in duthost.facts.get('platform_asic'):
         pytest.skip("Skipping this test on platforms with Broadcom ASICs")
 
-    logging.info("Get output of '{}'".format("show interface status"))
-    intf_status = duthost.show_and_parse("show interface status")
+    # Get interfaces that are operationally up and have supported speeds.
+    interfaces = get_fec_eligible_interfaces(duthost, SUPPORTED_SPEEDS)
 
-    for intf in intf_status:
-        sfp_presence = duthost.show_and_parse("sudo sfpshow presence -p {}"
-                                              .format(intf['interface']))
-        if sfp_presence:
-            presence = sfp_presence[0].get('presence', '').lower()
-            oper = intf.get('oper', '').lower()
-            speed = intf.get('speed', '')
-
-            if presence == "not present" or oper != "up" or speed not in SUPPORTED_SPEEDS:
-                continue
-
-        fec_mode = get_fec_oper_mode(duthost, intf['interface'])
+    for intf in interfaces:
+        fec_mode = get_fec_oper_mode(duthost, intf)
         if fec_mode == "n/a":
             pytest.fail("FEC status is N/A for interface {}".format(intf['interface']))
 
@@ -102,7 +86,7 @@ def test_config_fec_oper_mode(duthosts, enum_rand_one_per_hwsku_frontend_hostnam
             pytest_assert(wait_until(30, 2, 0, duthost.is_interface_status_up, intf["interface"]),
                           "Interface {} did not come up after configuring FEC mode".format(intf["interface"]))
             # Verify the FEC operational mode is restored
-            post_fec = get_fec_oper_mode(duthost, intf['interface'])
+            post_fec = get_fec_oper_mode(duthost, intf)
             if not (post_fec == fec_mode):
                 pytest.fail("FEC status is not restored for interface {}".format(intf['interface']))
 
@@ -149,7 +133,7 @@ def skip_ber_counters_test(intf_status: dict) -> bool:
 
     for intf in intf_status:
         intf_name = intf['iface']
-        speed = get_interface_speed(duthost, intf_name)
+        speed = duthost.get_speed(intf_name)
         if speed not in SUPPORTED_SPEEDS:
             continue
 
@@ -174,7 +158,7 @@ def skip_ber_counters_test(intf_status: dict) -> bool:
         # Check for valid FEC correctable codeword errors > FEC symbol errors
         if fec_symbol_err_int > fec_corr_int:
             pytest.fail("FEC symbol errors:{} are higher than FEC correctable errors:{} for interface {}"
-                        .format(intf_name, fec_symbol_err_int, fec_corr_int))
+                        .format(fec_symbol_err_int, fec_corr_int, intf_name))
 
         if skip_ber_counters_test(intf):
             continue
@@ -189,3 +173,66 @@ def skip_ber_counters_test(intf_status: dict) -> bool:
             pytest.fail("Pre-FEC and Post-FEC BER are not valid floats for interface {}, \
                     fec_pre_ber: {} fec_post_ber: {}"
                         .format(intf_name, fec_pre_ber, fec_post_ber))
+
+
+def get_fec_histogram(duthost, intf_name):
+    """
+    @Summary: Fetch FEC histogram for a given interface.
+    """
+    try:
+        logging.info("Get output of 'show interfaces counters fec-histogram {}'".format(intf_name))
+        fec_hist = duthost.show_and_parse("show interfaces counters fec-histogram {}".format(intf_name))
+    except Exception as e:
+        logging.error("Failed to execute 'show interfaces counters fec-histogram {}': {}".format(intf_name, e))
+        pytest.skip("Command 'show interfaces counters fec-histogram {}' not found \
+                or failed: {}".format(intf_name, str(e)))
+        return None
+
+    logging.info("FEC histogram for interface {}: {}".format(intf_name, fec_hist))
+    return fec_hist
+
+
+def validate_fec_histogram(duthost, intf_name):
+    """
+    @Summary: Validate FEC histogram critical bins for any errors. Fail the test if bin value > 0
+    """
+
+    fec_hist = get_fec_histogram(duthost, intf_name)
+    if not fec_hist:
+        pytest.fail("FEC histogram data not found or incomplete for interface {}".format(intf_name))
+
+    critical_bins = range(7, 16)
+    error_bins = []
+    for bin_index in critical_bins:
+        bin_value = int(fec_hist[bin_index].get('codewords', 0))
+        if bin_value > 0:
+            error_bins.append((bin_index, bin_value))
+
+    if error_bins:
+        error_messages = ["FEC histogram bin {} has errors for interface {}: {}".format(bin_index, intf_name, bin_value)
+                          for bin_index, bin_value in error_bins]
+        logging.error("\n".join(error_messages))
+        return False
+
+    return True
+
+
+def test_verify_fec_histogram(duthosts, enum_rand_one_per_hwsku_frontend_hostname,
+                              enum_frontend_asic_index, conn_graph_facts):
+    """
+    @Summary: Verify the FEC histogram is valid and check for errors
+    """
+    duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+
+    if "broadcom" in duthost.facts.get('platform_asic'):
+        if "7060x6" not in duthost.facts['platform']:
+            pytest.skip("Skipping this test on platforms with Broadcom ASICs")
+
+    # Get operationally up and interfaces with supported speeds
+    interfaces = get_fec_eligible_interfaces(duthost, SUPPORTED_SPEEDS)
+
+    for intf_name in interfaces:
+        for _ in range(3):
+            if not validate_fec_histogram(duthost, intf_name):
+                pytest.fail("FEC histogram validation failed for interface {}".format(intf_name))
+            time.sleep(10)