diff --git a/tests/gnmi/conftest.py b/tests/gnmi/conftest.py index 6a74fc580b9..859ba20aa6f 100644 --- a/tests/gnmi/conftest.py +++ b/tests/gnmi/conftest.py @@ -210,7 +210,13 @@ def setup_gnmi_server(duthosts, rand_one_dut_hostname, localhost, ptfhost): # Rollback configuration rollback(duthost, SETUP_ENV_CP) - recover_cert_config(duthost) + # Get the skip_gnmi_check flag from duthost options + skip_gnmi_check = duthost.host.options.get('skip_gnmi_check', False) + # Skip GNMI restart if the reboot flag is set + if not skip_gnmi_check: + recover_cert_config(duthost) + else: + logging.info("Skipping GNMI restart due to skip_gnmi_check flag") @pytest.fixture(scope="module", autouse=True) diff --git a/tests/gnmi/test_gnoi_system.py b/tests/gnmi/test_gnoi_system.py index a4ed6355369..b914aa726df 100644 --- a/tests/gnmi/test_gnoi_system.py +++ b/tests/gnmi/test_gnoi_system.py @@ -4,12 +4,13 @@ from .helper import gnoi_request from tests.common.helpers.assertions import pytest_assert -import re +from tests.common.reboot import wait_for_startup pytestmark = [ pytest.mark.topology('any') ] +MAX_TIME_TO_REBOOT = 300 """ This module contains tests for the gNOI System API. @@ -35,6 +36,118 @@ def test_gnoi_system_time(duthosts, rand_one_dut_hostname, localhost): pytest_assert("time" in msg_json, "System.Time API did not return time") +def test_gnoi_system_cold_reboot(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI System Reboot API triggers a reboot and the device comes back online. + """ + duthost = duthosts[rand_one_dut_hostname] + + # Set flag to indicate that this test involves reboot + duthost.host.options['skip_gnmi_reboot'] = True + + # Trigger reboot + ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 1,"delay":0,"message":"Cold Reboot"}') + pytest_assert(ret == 0, "System.Reboot API reported failure (rc = {}) with message: {}".format(ret, msg)) + logging.info("System.Reboot API returned msg: {}".format(msg)) + + +@pytest.mark.disable_loganalyzer +def test_gnoi_system_cold_reboot_fail_invalid_method(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI System Reboot API fails with invalid method. + """ + duthost = duthosts[rand_one_dut_hostname] + + # Set flag to indicate that this test involves reboot + duthost.host.options['skip_gnmi_reboot'] = True + + # Trigger reboot with invalid method + ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 99}') + pytest_assert(ret != 0, "System.Reboot API did not report failure with invalid method") + + +@pytest.mark.disable_loganalyzer +def test_gnoi_system_cold_reboot_when_reboot_active(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI System Reboot API fails if a reboot is already active. + """ + duthost = duthosts[rand_one_dut_hostname] + + # Set flag to indicate that this test involves reboot + duthost.host.options['skip_gnmi_reboot'] = True + + # Trigger first reboot + ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 1,"delay":0,"message":"Cold Reboot"}') + pytest_assert(ret == 0, "System.Reboot API reported failure (rc = {}) with message: {}".format(ret, msg)) + logging.info("System.Reboot API returned msg: {}".format(msg)) + + # Trigger second reboot while the first one is still active + ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 1,"delay":0,"message":"Cold Reboot"}') + pytest_assert(ret != 0, "System.Reboot API did not report failure when reboot is already active") + + +@pytest.mark.disable_loganalyzer +def test_gnoi_system_cold_reboot_status_immediately(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI System RebootStatus API returns the correct status immediately after reboot. + """ + duthost = duthosts[rand_one_dut_hostname] + + # Set flag to indicate that this test involves reboot + duthost.host.options['skip_gnmi_reboot'] = True + + # Trigger reboot + ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 1, "message": "test"}') + pytest_assert(ret == 0, "System.Reboot API reported failure (rc = {}) with message: {}".format(ret, msg)) + logging.info("System.Reboot API returned msg: {}".format(msg)) + + # Get reboot status + ret, msg = gnoi_request(duthost, localhost, "RebootStatus", "") + pytest_assert(ret == 0, "System.RebootStatus API reported failure (rc = {}) with message: {}".format(ret, msg)) + logging.info("System.RebootStatus API returned msg: {}".format(msg)) + # Message should contain a json substring like this + # {"active":true,"wait":0,"when":0,"reason":"test","count":1,"method":1,"status":1} + # Extract JSON part from the message + msg_json = extract_first_json_substring(msg) + if not msg_json: + pytest.fail("Failed to extract JSON from System.RebootStatus API response") + logging.info("Extracted JSON: {}".format(msg_json)) + pytest_assert("active" in msg_json, "System.RebootStatus API did not return active") + pytest_assert(msg_json["active"] is True, "System.RebootStatus API did not return active = true") + + +def gnoi_system_reboot_status_after_startup(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI System RebootStatus API returns the correct status after the device has started up. + """ + duthost = duthosts[rand_one_dut_hostname] + + # Set flag to indicate that this test involves reboot + duthost.host.options['skip_gnmi_reboot'] = True + + # Trigger reboot + ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 1, "message": "test"}') + pytest_assert(ret == 0, "System.Reboot API reported failure (rc = {}) with message: {}".format(ret, msg)) + logging.info("System.Reboot API returned msg: {}".format(msg)) + + # Wait for device to come back online + wait_for_startup(duthost) + + # Get reboot status + ret, msg = gnoi_request(duthost, localhost, "RebootStatus", "") + pytest_assert(ret == 0, "System.RebootStatus API reported failure (rc = {}) with message: {}".format(ret, msg)) + logging.info("System.RebootStatus API returned msg: {}".format(msg)) + # Message should contain a json substring like this + # {"active":false,"wait":0,"when":0,"reason":"test","count":1,"method":1,"status":1} + # Extract JSON part from the message + msg_json = extract_first_json_substring(msg) + if not msg_json: + pytest.fail("Failed to extract JSON from System.RebootStatus API response") + logging.info("Extracted JSON: {}".format(msg_json)) + pytest_assert("active" in msg_json, "System.RebootStatus API did not return active") + pytest_assert(msg_json["active"] is False, "System.RebootStatus API did not return active = false") + + def extract_first_json_substring(s): """ Extract the first JSON substring from a given string. @@ -43,12 +156,18 @@ def extract_first_json_substring(s): :return: The first JSON substring if found, otherwise None. """ - json_pattern = re.compile(r'\{.*?\}') - match = json_pattern.search(s) - if match: - try: - return json.loads(match.group()) - except json.JSONDecodeError: - logging.error("Failed to parse JSON: {}".format(match.group())) - return None - return None + start_index = s.find('{') # Find the first '{' in the string + if start_index == -1: + logging.error("No JSON found in response: {}".format(s)) + return None + json_str = s[start_index:] # Extract substring starting from '{' + try: + parsed_json = json.loads(json_str) # Attempt to parse the JSON + # Handle cases where "status": {} is empty + if "status" in parsed_json and parsed_json["status"] == {}: + logging.warning("Replacing empty 'status' field with a default value.") + parsed_json["status"] = {"unknown": "empty_status"} + return parsed_json + except json.JSONDecodeError as e: + logging.error("Failed to parse JSON: {} | Error: {}".format(json_str, e)) + return None