GCU loopback interface test

tests/generic_config_updater/gu_utils.py

@@ -1,12 +1,15 @@
 import json
 import logging
+import pytest
 from tests.common.helpers.assertions import pytest_assert
 from tests.common.utilities import wait_until
-
+from tests.common.config_reload import config_reload
 
 logger = logging.getLogger(__name__)
 
 CONTAINER_SERVICES_LIST = ["swss", "syncd", "radv", "lldp", "dhcp_relay", "teamd", "bgp", "pmon", "telemetry", "acms"]
+DEFAULT_CHECKPOINT_NAME = "test"
+YANG_IGNORED_OPTIONS    = "-i /FEATURE -i /QUEUE -i /SCHEDULER"
 
 def generate_tmpfile(duthost):
     """Generate temp file
@@ -28,7 +31,7 @@ def apply_patch(duthost, json_data, dest_file):
     """
     duthost.copy(content=json.dumps(json_data, indent=4), dest=dest_file)
 
-    cmds = 'config apply-patch {}'.format(dest_file)
+    cmds = 'config apply-patch {} {}'.format(YANG_IGNORED_OPTIONS, dest_file)
 
     logger.info("Commands: {}".format(cmds))
     output = duthost.shell(cmds, module_ignore_errors=True)
@@ -95,7 +98,7 @@ def start_limit_hit(duthost, service_name):
     Args:
         service_name: Service to reset
     """
-    service_status = duthost.shell("sudo systemctl status {}.service | grep 'Active'".format(service_name))
+    service_status = duthost.shell("systemctl status {}.service | grep 'Active'".format(service_name))
     pytest_assert(
         not service_status['rc'],
         "{} service status cannot be found".format(service_name)
@@ -119,13 +122,13 @@ def reset_start_limit_hit(duthost, service_name, timeout, interval, delay):
     """
     logger.info("Reset service '{}' due to start-limit-hit".format(service_name))
 
-    service_reset_failed = duthost.shell("sudo systemctl reset-failed {}.service".format(service_name))
+    service_reset_failed = duthost.shell("systemctl reset-failed {}.service".format(service_name))
     pytest_assert(
         not service_reset_failed['rc'],
         "{} systemctl reset-failed service fails"
     )
 
-    service_start = duthost.shell("sudo systemctl start {}.service".format(service_name))
+    service_start = duthost.shell("systemctl start {}.service".format(service_name))
     pytest_assert(
         not service_start['rc'],
         "{} systemctl start service fails"
@@ -143,3 +146,95 @@ def reset_start_limit_hit(duthost, service_name, timeout, interval, delay):
         reset_service,
         "Failed to reset service '{}' due to start-limit-hit".format(service_name)
     )
+
+def list_checkpoints(duthost):
+    """List checkpoint on target duthost
+
+    Args:
+        duthost: Device Under Test (DUT)
+        cp: checkpoint filename
+    """
+    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 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 delete_checkpoint(duthost, cp=DEFAULT_CHECKPOINT_NAME):
+    """Run checkpoint on target duthost
+
+    Args:
+        duthost: Device Under Test (DUT)
+        cp: checkpoint filename
+    """
+    pytest_assert(
+        verify_checkpoints_exist(duthost, cp),
+        "Failed to find the checkpoint file: {}".format(cp)
+    )
+
+    cmds = 'config delete-checkpoint {}'.format(cp)
+
+    logger.info("Commands: {}".format(cmds))
+    output = duthost.shell(cmds, module_ignore_errors=True)
+
+    pytest_assert(
+        not output['rc'] and "Checkpoint deleted successfully" in output['stdout'],
+        "Failed to delete a checkpoint file: {}".format(cp)
+    )
+
+def rollback(duthost, cp=DEFAULT_CHECKPOINT_NAME):
+    """Run rollback on target duthost
+
+    Args:
+        duthost: Device Under Test (DUT)
+        rb: rollback filename
+    """
+    cmds = 'config rollback {} {}'.format(YANG_IGNORED_OPTIONS, cp)
+
+    logger.info("Commands: {}".format(cmds))
+    output = duthost.shell(cmds, module_ignore_errors=True)
+
+    return output
+
+def rollback_or_reload(duthost, cp=DEFAULT_CHECKPOINT_NAME):
+    """Run rollback on target duthost. config_reload if rollback failed.
+
+    Args:
+        duthost: Device Under Test (DUT)
+    """
+    output = rollback(duthost, cp)
+
+    if output['rc'] or "Config rolled back successfull" not in output['stdout']:
+        config_reload(duthost)
+        pytest.fail("config rollback failed. Restored by config_reload")