Fix for parse_show_copp_configuration for multi-asic dut

tests/copp/copp_utils.py

@@ -492,16 +492,18 @@ def get_copp_trap_capabilities(duthost):
     return trap_ids.split(",")
 
 
-def parse_show_copp_configuration(duthost):
+def parse_show_copp_configuration(duthost, namespace):
     """
     Parses the output of the `show copp configuration` command into a structured dictionary.
     Args:
         duthost (SonicHost): The target device.
     Returns:
         dict: A dictionary mapping trap IDs to their configuration details.
     """
-
-    copp_config_output = duthost.shell("show copp configuration")["stdout"]
+    command = "show copp configuration"
+    if namespace is not None:
+        command = namespace.ns_arg + ' ' + command
+    copp_config_output = duthost.shell(command)["stdout"]
     copp_config_lines = copp_config_output.splitlines()
 
     # Parse the command output into a structured format
@@ -523,7 +525,7 @@ def parse_show_copp_configuration(duthost):
     return copp_config_data
 
 
-def is_trap_installed(duthost, trap_id):
+def is_trap_installed(duthost, trap_id, namespace=None):
     """
     Checks if a specific trap is installed by parsing the output of `show copp configuration`.
     Args:
@@ -533,7 +535,7 @@ def is_trap_installed(duthost, trap_id):
         bool: True if the trap is installed, False otherwise.
     """
 
-    output = parse_show_copp_configuration(duthost)
+    output = parse_show_copp_configuration(duthost, namespace)
     assert trap_id in output, f"Trap {trap_id} not found in the configuration"
     assert "hw_status" in output[trap_id], f"hw_status not found for trap {trap_id}"
 
@@ -552,22 +554,27 @@ def is_trap_uninstalled(duthost, trap_id):
     return not is_trap_installed(duthost, trap_id)
 
 
-def get_trap_hw_status(duthost):
+def get_trap_hw_status(duthost, namespace):
     """
     Retrieves the hw_status for traps from the STATE_DB.
     Args:
         dut (SonicHost): The target device
     Returns:
         dict: A dictionary mapping trap IDs to their hw_status.
     """
-
-    state_db_data = duthost.shell("sonic-db-cli STATE_DB KEYS 'COPP_TRAP_TABLE|*'")["stdout"]
+    if namespace is None:
+        state_db_cmd = "sonic-db-cli STATE_DB KEYS 'COPP_TRAP_TABLE|*'"
+        trap_data_cmd = "sonic-db-cli STATE_DB HGETALL "
+    else:
+        state_db_cmd = "sonic-db-cli -n {} STATE_DB KEYS 'COPP_TRAP_TABLE|*'".format(namespace.namespace)
+        trap_data_cmd = "sonic-db-cli -n {} STATE_DB HGETALL ".format(namespace.namespace)
+    state_db_data = duthost.shell(state_db_cmd)["stdout"]
     state_db_data = state_db_data.splitlines()
     hw_status = {}
 
     for key in state_db_data:
         trap_id = key.split("|")[-1]
-        trap_data = duthost.shell(f"sonic-db-cli STATE_DB HGETALL '{key}'")["stdout"]
+        trap_data = duthost.shell(trap_data_cmd + f"'{key}'")["stdout"]
         trap_data_dict = ast.literal_eval(trap_data)
         hw_status[trap_id] = trap_data_dict.get("hw_status", "not-installed")
 

tests/copp/test_copp.py

@@ -35,6 +35,7 @@
 from tests.common.utilities import skip_release
 from tests.common.utilities import wait_until
 from tests.common.helpers.assertions import pytest_assert
+from tests.common.helpers.constants import DEFAULT_NAMESPACE
 from tests.common.utilities import find_duthost_on_role
 from tests.common.utilities import get_upstream_neigh_type
 
@@ -119,17 +120,20 @@ def test_policer(self, protocol, duthosts, enum_rand_one_per_hwsku_frontend_host
                     pytest.skip("Skip UDLD test for Arista-7060x6 fanout without UDLD forward support")
 
         duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+        namespace = DEFAULT_NAMESPACE
+        if duthost.is_multi_asic:
+            namespace = random.choice(duthost.asics)
 
         # Skip the check if the protocol is "Default"
         if protocol != "Default":
             trap_ids = PROTOCOL_TO_TRAP_ID.get(protocol)
             is_always_enabled, feature_name = copp_utils.get_feature_name_from_trap_id(duthost, trap_ids[0])
             if is_always_enabled:
-                pytest_assert(copp_utils.is_trap_installed(duthost, trap_ids[0]),
+                pytest_assert(copp_utils.is_trap_installed(duthost, trap_ids[0], namespace),
                               f"Trap {trap_ids[0]} for protocol {protocol} is not installed")
             else:
                 feature_list, _ = duthost.get_feature_status()
-                trap_installed = copp_utils.is_trap_installed(duthost, trap_ids[0])
+                trap_installed = copp_utils.is_trap_installed(duthost, trap_ids[0], namespace)
                 if feature_name in feature_list and feature_list[feature_name] == "enabled":
                     pytest_assert(trap_installed,
                                   f"Trap {trap_ids[0]} for protocol {protocol} is not installed")
@@ -303,10 +307,13 @@ def test_verify_copp_configuration_cli(duthosts, enum_rand_one_per_hwsku_fronten
     """
 
     duthost = duthosts[enum_rand_one_per_hwsku_frontend_hostname]
+    namespace = DEFAULT_NAMESPACE
+    if duthost.is_multi_asic:
+        namespace = random.choice(duthost.asics)
 
     trap, trap_group, copp_group_cfg = copp_utils.get_random_copp_trap_config(duthost)
-    hw_status = copp_utils.get_trap_hw_status(duthost)
-    show_copp_config = copp_utils.parse_show_copp_configuration(duthost)
+    hw_status = copp_utils.get_trap_hw_status(duthost, namespace)
+    show_copp_config = copp_utils.parse_show_copp_configuration(duthost, namespace)
 
     pytest_assert(trap in show_copp_config,
                   f"Trap {trap} not found in show copp configuration output")