Add Multi ASIC GCU test cases for IDF and LinkCRC.

.azure-pipelines/pr_test_scripts.yaml

@@ -444,6 +444,8 @@ multi-asic-t1-lag:
   - process_monitoring/test_critical_process_monitoring.py
   - container_checker/test_container_checker.py
   - http/test_http_copy.py
+  - generic_config_updater/test_multiasic_scenarios.py
+
 
 dpu:
   - dash/test_dash_vnet.py

tests/generic_config_updater/gu_utils.py

@@ -1,58 +1,148 @@
-import os
-import logging
 import json
-from tests.common.gu_utils import apply_patch, generate_tmpfile, delete_tmpfile
+import logging
 
+import pytest
 
-BASE_DIR = os.path.dirname(os.path.realpath(__file__))
-TEMPLATES_DIR = os.path.join(BASE_DIR, "../generic_config_updater/templates")
-TMP_DIR = '/tmp'
+from tests.common import config_reload
+from tests.common.helpers.assertions import pytest_assert
 
 logger = logging.getLogger(__name__)
+DEFAULT_CHECKPOINT_NAME = "test"
+
+
+def generate_tmpfile(duthost):
+    """Generate temp file
+    """
+    return duthost.shell('mktemp')['stdout']
+
+
+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, ignore_tables if ignore_tables else "")
+
+    logger.info("Commands: {}".format(cmds))
+    output = duthost.shell(cmds, module_ignore_errors=True)
+
+    return output
+
+
+def delete_tmpfile(duthost, tmpfile):
+    """Delete temp file
+    """
+    duthost.file(path=tmpfile, state='absent')
+
+
+def create_checkpoint(duthost, cp=DEFAULT_CHECKPOINT_NAME):
+    """Run checkpoint on target duthost
+
+    Args:
+        duthost: Device Under Test (DUT)
+        cp: checkpoint filename
+    """
+    cmds = 'config checkpoint {}'.format(cp)
+
+    logger.info("Commands: {}".format(cmds))
+    output = duthost.shell(cmds, module_ignore_errors=True)
+
+    pytest_assert(
+        not output['rc']
+        and "Checkpoint created successfully" in output['stdout']
+        and verify_checkpoints_exist(duthost, cp),
+        "Failed to config a checkpoint file: {}".format(cp)
+    )
+
+
+def list_checkpoints(duthost):
+    """List checkpoint on target duthost
+
+    Args:
+        duthost: Device Under Test (DUT)
+    """
+    cmds = 'config list-checkpoints'
+
+    logger.info("Commands: {}".format(cmds))
+    output = duthost.shell(cmds, module_ignore_errors=True)
+
+    pytest_assert(
+        not output['rc'],
+        "Failed to list all checkpoint file"
+    )
+
+    return output
+
+
+def verify_checkpoints_exist(duthost, cp):
+    """Check if checkpoint file exist in duthost
+    """
+    output = list_checkpoints(duthost)
+    return '"{}"'.format(cp) in output['stdout']
+
+
+def rollback(duthost, cp=DEFAULT_CHECKPOINT_NAME):
+    """Run rollback on target duthost
+
+    Args:
+        duthost: Device Under Test (DUT)
+        cp: rollback filename
+    """
+    cmds = 'config rollback {}'.format(cp)
+
+    logger.info("Commands: {}".format(cmds))
+    output = duthost.shell(cmds, module_ignore_errors=True)
 
+    return output
 
-def format_and_apply_template(duthost, template_name, extra_vars, setup):
-    dest_path = os.path.join(TMP_DIR, template_name)
 
-    duts_to_apply = [duthost]
-    outputs = []
-    if setup["is_dualtor"]:
-        duts_to_apply.append(setup["rand_unselected_dut"])
+def rollback_or_reload(duthost, cp=DEFAULT_CHECKPOINT_NAME):
+    """Run rollback on target duthost. config_reload if rollback failed.
 
-    for dut in duts_to_apply:
-        dut.host.options['variable_manager'].extra_vars.update(extra_vars)
-        dut.file(path=dest_path, state='absent')
-        dut.template(src=os.path.join(TEMPLATES_DIR, template_name), dest=dest_path)
+    Args:
+        duthost: Device Under Test (DUT)
+    """
+    output = rollback(duthost, cp)
 
-        try:
-            # duthost.template uses single quotes, which breaks apply-patch. this replaces them with double quotes
-            dut.shell("sed -i \"s/'/\\\"/g\" " + dest_path)
-            output = dut.shell("config apply-patch {}".format(dest_path))
-            outputs.append(output)
-        finally:
-            dut.file(path=dest_path, state='absent')
+    if output['rc'] or "Config rolled back successfully" not in output['stdout']:
+        config_reload(duthost)
+        pytest.fail("config rollback failed. Restored by config_reload")
 
-    return outputs
 
+def delete_checkpoint(duthost, cp=DEFAULT_CHECKPOINT_NAME):
+    """Run checkpoint on target duthost
 
-def load_and_apply_json_patch(duthost, file_name, setup):
-    with open(os.path.join(TEMPLATES_DIR, file_name)) as file:
-        json_patch = json.load(file)
+    Args:
+        duthost: Device Under Test (DUT)
+        cp: checkpoint filename
+    """
+    pytest_assert(
+        verify_checkpoints_exist(duthost, cp),
+        "Failed to find the checkpoint file: {}".format(cp)
+    )
 
-    duts_to_apply = [duthost]
-    outputs = []
-    if setup["is_dualtor"]:
-        duts_to_apply.append(setup["rand_unselected_dut"])
+    cmds = 'config delete-checkpoint {}'.format(cp)
 
-    for dut in duts_to_apply:
+    logger.info("Commands: {}".format(cmds))
+    output = duthost.shell(cmds, module_ignore_errors=True)
 
-        tmpfile = generate_tmpfile(dut)
-        logger.info("tmpfile {}".format(tmpfile))
+    pytest_assert(
+        not output['rc'] and "Checkpoint deleted successfully" in output['stdout'],
+        "Failed to delete a checkpoint file: {}".format(cp)
+    )
 
-        try:
-            output = apply_patch(dut, json_data=json_patch, dest_file=tmpfile)
-            outputs.append(output)
-        finally:
-            delete_tmpfile(dut, tmpfile)
 
-    return outputs
+def expect_op_success(duthost, output):
+    """Expected success from apply-patch output
+    """
+    pytest_assert(not output['rc'], "Command is not running successfully")
+    pytest_assert(
+        "Patch applied successfully" in output['stdout'],
+        "Please check if json file is validate"
+    )

tests/generic_config_updater/test_multiasic_scenarios.py

@@ -0,0 +1,197 @@
+import json
+import logging
+import pytest
+import re
+
+from tests.common.helpers.assertions import pytest_assert
+from tests.generic_config_updater.gu_utils import apply_patch
+from tests.generic_config_updater.gu_utils import generate_tmpfile, delete_tmpfile
+from tests.generic_config_updater.gu_utils import create_checkpoint, delete_checkpoint, rollback_or_reload
+
+pytestmark = [
+    pytest.mark.topology('any'),
+]
+
+logger = logging.getLogger(__name__)
+
+IDF_ISOLATION = [
+    {
+        "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"
+    },
+]
+
+IDF_UNISOLATION = [
+    {
+        "op": "add",
+        "path": "/asic0/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+        "value": "unisolated"
+    },
+    {
+        "op": "add",
+        "path": "/asic1/BGP_DEVICE_GLOBAL/STATE/idf_isolation_state",
+        "value": "unisolated"
+    },
+]
+
+LINK_CRC_MITIGATION_REMOVE_TEMPLATE = '[{{"op": "remove", "path": "/asic0/PORTCHANNEL_MEMBER/{}|{}"}}]'
+LINK_CRC_MITIGATION_ADD_TEMPLATE = '[{{"op": "add", "path": "/asic0/PORTCHANNEL_MEMBER/{}|{}", "value": {}}}]'
+
+
+def extract_up_interface(output):
+    # Updated regex pattern to match both (U) and (S) status
+    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)
+    else:
+        return None, None
+
+
+@pytest.fixture(autouse=True)
+def setup_env(duthosts, rand_one_dut_hostname):
+    """
+    Setup/teardown fixture for each multi asic test.
+    rollback to check if it goes back to starting config
+    Args:
+        duthosts: list of DUTs.
+        rand_selected_dut: The fixture returns a randomly selected DuT.
+    """
+    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):
+    json_patch = []
+    tmpfile = generate_tmpfile(duthost)
+
+    try:
+        output = apply_patch(duthost, json_data=json_patch, dest_file=tmpfile)
+    finally:
+        delete_tmpfile(duthost, tmpfile)
+
+    if output['rc'] or "Patch applied successfully" not in output['stdout']:
+        logger.info("Patching process broken, the error output is {}").format(output['stdout'])
+        pytest_assert(False, "Patching process broken, the error output is {}").format(output['stdout'])
+
+
+def test_check_idf_isolation_apply_patch(duthost):
+    json_patch = IDF_ISOLATION
+    tmpfile = generate_tmpfile(duthost)
+
+    try:
+        output = apply_patch(duthost, json_data=json_patch, dest_file=tmpfile, ignore_tables="-i PORT")
+
+        if output['rc'] or "Patch applied successfully" not in output['stdout']:
+            logger.info("Patching process broken, the error output is {}".format(output['stdout']))
+            pytest_assert(False, "Patching process broken, the error output is {}").format(output['stdout'])
+
+        cmds = 'sonic-db-cli -n asic0 CONFIG_DB hget "BGP_DEVICE_GLOBAL|STATE" idf_isolation_state'
+        expected_value = "isolated_no_export"
+        redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout']
+        pytest_assert(redis_value == expected_value, "Config IDF ISOLATION failed")
+
+        cmds = 'sonic-db-cli -n asic1 CONFIG_DB hget "BGP_DEVICE_GLOBAL|STATE" "idf_isolation_state"'
+        expected_value = "isolated_withdraw_all"
+        redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout']
+        pytest_assert(redis_value == expected_value, "Config IDF ISOLATION failed")
+    finally:
+        delete_tmpfile(duthost, tmpfile)
+
+
+def test_check_idf_unisolation_apply_patch(duthost):
+    json_patch = IDF_UNISOLATION
+    tmpfile = generate_tmpfile(duthost)
+
+    try:
+        output = apply_patch(duthost, json_data=json_patch, dest_file=tmpfile, ignore_tables="-i PORT")
+
+        if output['rc'] or "Patch applied successfully" not in output['stdout']:
+            logger.info("Patching process broken, the error output is {}".format(output['stdout']))
+            pytest_assert(False, "Patching process broken, the error output is {}").format(output['stdout'])
+
+        cmds = 'sonic-db-cli -n asic0 CONFIG_DB hget "BGP_DEVICE_GLOBAL|STATE" idf_isolation_state'
+        expected_value = "unisolated"
+        redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout']
+        pytest_assert(redis_value == expected_value, "Config IDF ISOLATION failed")
+
+        cmds = 'sonic-db-cli -n asic1 CONFIG_DB hget "BGP_DEVICE_GLOBAL|STATE" idf_isolation_state'
+        expected_value = "unisolated"
+        redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout']
+        pytest_assert(redis_value == expected_value, "Config IDF ISOLATION failed")
+    finally:
+        delete_tmpfile(duthost, tmpfile)
+
+
+def test_check_link_crc_mitigation_remove_and_add_apply_patch(duthost):
+    tmpfile = generate_tmpfile(duthost)
+
+    try:
+        result = duthost.shell("show interfaces portchannel -n asic0", module_ignore_errors=False)['stdout']
+        portchannel, port = extract_up_interface(result)
+
+        # Precheck keys existing
+        cmds = 'sonic-db-cli -n asic0 CONFIG_DB keys "PORTCHANNEL_MEMBER|{}|{}"'.format(portchannel, port)
+        expected_value = "PORTCHANNEL_MEMBER|{}|{}".format(portchannel, port)
+        redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout']
+        pytest_assert(redis_value == expected_value, "Config Link CRC Mitigation add action failed.")
+
+        json_patch = LINK_CRC_MITIGATION_REMOVE_TEMPLATE.format(portchannel, port)
+        output = apply_patch(duthost, json_data=json.loads(json_patch), dest_file=tmpfile, ignore_tables="-i PORT")
+
+        if output['rc'] or "Patch applied successfully" not in output['stdout']:
+            logger.info("Patching process broken, the error output is {}".format(output['stdout']))
+            pytest_assert(False, "Patching process broken, the error output is {}").format(output['stdout'])
+
+        cmds = 'sonic-db-cli -n asic0 CONFIG_DB keys "PORTCHANNEL_MEMBER|{}|{}"'.format(portchannel, port)
+        expected_value = ""
+        redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout']
+        pytest_assert(redis_value.strip() == expected_value, "Config Link CRC Mitigation remove action failed.")
+
+        json_patch = LINK_CRC_MITIGATION_ADD_TEMPLATE.format(portchannel, port, "{}")
+        output = apply_patch(duthost, json_data=json.loads(json_patch), dest_file=tmpfile)
+
+        if output['rc'] or "Patch applied successfully" not in output['stdout']:
+            logger.info("Patching process broken, the error output is {}".format(output['stdout']))
+            pytest_assert(False, "Patching process broken, the error output is {}").format(output['stdout'])
+
+        cmds = 'sonic-db-cli -n asic0 CONFIG_DB keys "PORTCHANNEL_MEMBER|{}|{}"'.format(portchannel, port)
+        expected_value = "PORTCHANNEL_MEMBER|{}|{}".format(portchannel, port)
+        redis_value = duthost.shell(cmds, module_ignore_errors=False)['stdout']
+        pytest_assert(redis_value == expected_value, "Config Link CRC Mitigation add action failed.")
+    finally:
+        delete_tmpfile(duthost, tmpfile)
+
+
+def test_check_apply_patch_negative_case(duthost):
+    json_patch = '[{"op": "replace", "path": "/x"}]'
+    tmpfile = generate_tmpfile(duthost)
+
+    try:
+        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"],
+        "Expected failure did not occur as expected. Output: {}".format(
+            output["stderr"]
+        ),
+    )