Add Multi ASIC GCU test cases for IDF and LinkCRC.

.azure-pipelines/pr_test_scripts.yaml

@@ -458,6 +458,9 @@ multi-asic-t1-lag:
   - http/test_http_copy.py
   - telemetry/test_telemetry_cert_rotation.py
   - telemetry/test_telemetry.py
+  - generic_config_updater/test_multiasic_idf.py
+  - generic_config_updater/test_multiasic_linkcrc.py
+
 
 dpu:
   - dash/test_dash_vnet.py

tests/common/gu_utils.py

@@ -66,17 +66,18 @@ def format_json_patch_for_multiasic(duthost, json_data, is_asic_specific=False):
     return json_data
 
 
-def apply_patch(duthost, json_data, dest_file):
+def apply_patch(duthost, json_data, dest_file, ignore_tables=None):
     """Run apply-patch on target duthost
 
     Args:
         duthost: Device Under Test (DUT)
         json_data: Source json patch to apply
         dest_file: Destination file on duthost
+        ignore_tables: to be ignored tables, "-i table_name"
     """
     duthost.copy(content=json.dumps(json_data, indent=4), dest=dest_file)
 
-    cmds = 'config apply-patch {}'.format(dest_file)
+    cmds = 'config apply-patch {} {}'.format(dest_file, ignore_tables if ignore_tables else "")
 
     logger.info("Commands: {}".format(cmds))
     start_time = time.time()

tests/generic_config_updater/test_multiasic_idf.py

@@ -0,0 +1,210 @@
+import logging
+import pytest
+from tests.common.helpers.assertions import pytest_assert
+from tests.common.gu_utils import apply_patch
+from tests.common.gu_utils import generate_tmpfile, delete_tmpfile
+from tests.common.gu_utils import (create_checkpoint, delete_checkpoint, rollback_or_reload)
+
+pytestmark = [
+    pytest.mark.topology('any'),
+]
+
+logger = logging.getLogger(__name__)
+
+
+def apply_patch_and_verify(duthost, json_patch, tmpfile):
+    output = apply_patch(duthost, json_data=json_patch, dest_file=tmpfile)
+    if output['rc'] or "Patch applied successfully" not in output['stdout']:
+        err_msg = f"Patching failed: {output['stdout']}"
+        logger.info(err_msg)
+        pytest_assert(False, err_msg)
+    return output
+
+
+def verify_asic_state(duthost, asic_id, expected_state):
+    cmds = (f'sonic-db-cli -n asic{asic_id} CONFIG_DB hget "BGP_DEVICE_GLOBAL|STATE" idf_isolation_state')
+    redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout']
+    pytest_assert(redis_value == expected_state, f"Config IDF ISOLATION failed for asic{asic_id}")
+
+
+def verify_idf_status(duthost, expected_states):
+    if duthost.facts['router_type'] != 'spinerouter':
+        return
+    status_output = duthost.shell("sudo idf_isolation status", module_ignore_errors=False)['stdout']
+
+    if isinstance(expected_states, dict):
+        expected_lines = [
+            f"BGP{asic_id}: IDF isolation state: {state}"
+            for asic_id, state in expected_states.items()
+        ]
+    else:
+        expected_lines = expected_states
+
+    for line in expected_lines:
+        pytest_assert(
+            line in status_output,
+            f"IDF isolation status check failed: {line} not found"
+        )
+
+
+@pytest.fixture(autouse=True)
+def setup_env(duthosts, rand_one_dut_hostname):
+    """Setup/teardown fixture for each test"""
+    duthost = duthosts[rand_one_dut_hostname]
+    create_checkpoint(duthost)
+    yield
+    try:
+        logger.info("Rolled back to original checkpoint")
+        rollback_or_reload(duthost)
+    finally:
+        delete_checkpoint(duthost)
+
+
+@pytest.fixture
+def setup_tmpfile(duthost):
+    """Fixture to handle tmpfile creation/cleanup"""
+    tmpfile = generate_tmpfile(duthost)
+    yield tmpfile
+    delete_tmpfile(duthost, tmpfile)
+
+
+test_params = [
+    pytest.param(
+        [],
+        None,
+        id="empty_patch"
+    ),
+    pytest.param(
+        [
+            {
+                "op": "add",
+                "path": "/asic0/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+                "value": "isolated_no_export"
+            },
+            {
+                "op": "add",
+                "path": "/asic1/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+                "value": "isolated_withdraw_all"
+            }
+        ],
+        {
+            0: "isolated_no_export",
+            1: "isolated_withdraw_all"
+        },
+        id="basic_isolation"
+    ),
+    pytest.param(
+        [
+            {
+                "op": "add",
+                "path": f"/asic{i}/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+                "value": "unisolated"
+            }
+            for i in [0, 1]
+        ],
+        {
+            0: "unisolated",
+            1: "unisolated"
+        },
+        id="unisolation"
+    ),
+    pytest.param(
+        [
+            {
+                "op": "add",
+                "path": f"/asic{i}/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+                "value": "isolated_no_export"
+            }
+            for i in [0, 1]
+        ],
+        {
+            0: "isolated_no_export",
+            1: "isolated_no_export"
+        },
+        id="no_export_all"
+    )
+]
+
+
+@pytest.mark.parametrize("json_patch,expected_states", test_params)
+def test_idf_isolation_states(duthost, setup_tmpfile, json_patch, expected_states):
+    """Parameterized test for various IDF isolation states"""
+    tmpfile = setup_tmpfile
+
+    # For basic isolation test, show current config
+    if expected_states and 0 in expected_states:
+        if (expected_states[0] == "isolated_no_export" and
+                expected_states[1] == "isolated_withdraw_all"):
+            logger.info("The current running config is:")
+            logger.info(
+                duthost.shell("show run all", module_ignore_errors=False)['stdout']
+            )
+
+    apply_patch_and_verify(duthost, json_patch, tmpfile)
+
+    if expected_states:
+        # Verify states for each ASIC
+        for asic_id, expected_state in expected_states.items():
+            verify_asic_state(duthost, asic_id, expected_state)
+        # Verify type of expected_states
+        pytest_assert(isinstance(expected_states, dict), "expected_states must be a dictionary")
+        # Verify status output
+        verify_idf_status(duthost, expected_states)
+
+
+# Mixed states test cases
+mixed_states_params = [
+    # asic0: no_export, asic1: withdraw_all
+    {
+        "patch": [
+            {
+                "op": "add",
+                "path": "/asic0/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+                "value": "isolated_no_export"
+            },
+            {
+                "op": "add",
+                "path": "/asic1/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+                "value": "isolated_withdraw_all"
+            }
+        ],
+        "expected_states": {
+            0: "isolated_no_export",
+            1: "isolated_withdraw_all"
+        }
+    },
+    # asic0: withdraw_all, asic1: no_export
+    {
+        "patch": [
+            {
+                "op": "add",
+                "path": "/asic0/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+                "value": "isolated_withdraw_all"
+            },
+            {
+                "op": "add",
+                "path": "/asic1/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+                "value": "isolated_no_export"
+            }
+        ],
+        "expected_states": {
+            0: "isolated_withdraw_all",
+            1: "isolated_no_export"
+        }
+    }
+]
+
+
+@pytest.mark.parametrize("test_case", mixed_states_params)
+def test_idf_isolation_mixed_states(duthost, setup_tmpfile, test_case):
+    """Test different isolation states on different ASICs"""
+    tmpfile = setup_tmpfile
+
+    apply_patch_and_verify(duthost, test_case["patch"], tmpfile)
+
+    # Verify states
+    for asic_id, expected_state in test_case["expected_states"].items():
+        verify_asic_state(duthost, asic_id, expected_state)
+
+    # Verify status output
+    verify_idf_status(duthost, test_case["expected_states"])

tests/generic_config_updater/test_multiasic_linkcrc.py

@@ -0,0 +1,129 @@
+import json
+import logging
+import pytest
+import re
+
+from tests.common.helpers.assertions import pytest_assert
+from tests.common.gu_utils import apply_patch
+from tests.common.gu_utils import generate_tmpfile, delete_tmpfile
+from tests.common.gu_utils import create_checkpoint, delete_checkpoint, rollback_or_reload
+
+pytestmark = [
+    pytest.mark.topology('any'),
+]
+
+logger = logging.getLogger(__name__)
+
+LINK_CRC_MITIGATION_ADD_TEMPLATE = '[{{"op": "add", "path": "/asic0/PORTCHANNEL_MEMBER/{}|{}", "value": {}}}]'
+LINK_CRC_MITIGATION_REMOVE_TEMPLATE = '[{{"op": "remove", "path": "/asic0/PORTCHANNEL_MEMBER/{}|{}"}}]'
+
+
+def extract_up_interface(output):
+    """Extract portchannel and port from interface output."""
+    pattern = re.compile(
+        r"^\s*(\d+)\s+(PortChannel\d+)\s+LACP\(\w+\)\(Up\)\s+(Ethernet\d+)\([US]\)",
+        re.MULTILINE
+    )
+    match = pattern.search(output)
+    if match:
+        return match.group(2), match.group(3)
+    return None, None
+
+
+def apply_patch_and_verify(duthost, json_patch, tmpfile):
+    """Apply patch and verify success."""
+    output = apply_patch(duthost, json_data=json_patch, dest_file=tmpfile)
+    if output['rc'] or "Patch applied successfully" not in output['stdout']:
+        err_msg = f"Patching failed: {output['stdout']}"
+        logger.info(err_msg)
+        pytest_assert(False, err_msg)
+    return output
+
+
+def verify_portchannel_member(duthost, portchannel, port, member_exists):
+    """Verify portchannel member state in CONFIG_DB.
+
+    Args:
+        duthost: DUT host object
+        portchannel: Name of the portchannel
+        port: Name of the member port
+        member_exists: Boolean indicating if member should exist
+    """
+    cmds = f'sonic-db-cli -n asic0 CONFIG_DB keys "PORTCHANNEL_MEMBER|{portchannel}|{port}"'
+    redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout'].strip()
+    expected_value = f"PORTCHANNEL_MEMBER|{portchannel}|{port}" if member_exists else ""
+    pytest_assert(redis_value == expected_value,
+                  f"Config Link CRC Mitigation action failed. Expected: {expected_value}, Got: {redis_value}")
+
+
+def show_current_config(duthost):
+    """Show current running config."""
+    logger.info("The current running config is:")
+    logger.info(duthost.shell("show run all", module_ignore_errors=False)['stdout'])
+
+
+@pytest.fixture(autouse=True)
+def setup_env(duthosts, rand_one_dut_hostname):
+    """Setup/teardown fixture for each multi asic test."""
+    duthost = duthosts[rand_one_dut_hostname]
+    create_checkpoint(duthost)
+    yield
+    try:
+        logger.info("Rolled back to original checkpoint")
+        rollback_or_reload(duthost)
+    finally:
+        delete_checkpoint(duthost)
+
+
+def test_check_empty_apply_patch(duthost):
+    """Test applying empty patch."""
+    json_patch = []
+    tmpfile = generate_tmpfile(duthost)
+
+    try:
+        apply_patch_and_verify(duthost, json_patch, tmpfile)
+    finally:
+        delete_tmpfile(duthost, tmpfile)
+
+
+def test_check_link_crc_mitigation_remove_and_add_apply_patch(duthost):
+    """Test removing and adding link CRC mitigation."""
+    tmpfile = generate_tmpfile(duthost)
+    try:
+        show_current_config(duthost)
+
+        result = duthost.shell("show interfaces portchannel -n asic0", module_ignore_errors=False)['stdout']
+        portchannel, port = extract_up_interface(result)
+
+        # Verify initial state
+        verify_portchannel_member(duthost, portchannel, port, True)
+
+        # Remove member
+        json_patch = LINK_CRC_MITIGATION_REMOVE_TEMPLATE.format(portchannel, port)
+        apply_patch_and_verify(duthost, json.loads(json_patch), tmpfile)
+        verify_portchannel_member(duthost, portchannel, port, False)
+
+        # Add member back
+        json_patch = LINK_CRC_MITIGATION_ADD_TEMPLATE.format(portchannel, port, "{}")
+        apply_patch_and_verify(duthost, json.loads(json_patch), tmpfile)
+        verify_portchannel_member(duthost, portchannel, port, True)
+
+    finally:
+        delete_tmpfile(duthost, tmpfile)
+
+
+def test_check_apply_patch_negative_case(duthost):
+    """Test patch failure case."""
+    json_patch = '[{"op": "replace", "path": "/x"}]'
+    tmpfile = generate_tmpfile(duthost)
+
+    try:
+        show_current_config(duthost)
+        output = apply_patch(duthost, json_data=json.loads(json_patch), dest_file=tmpfile)
+    finally:
+        delete_tmpfile(duthost, tmpfile)
+
+    pytest_assert(
+        output["rc"] != 0 and "Failed to apply patch" in output["stderr"],
+        f"Expected failure did not occur as expected. Output: {output['stderr']}"
+    )