diff --git a/tests/common/plugins/loganalyzer/__init__.py b/tests/common/plugins/loganalyzer/__init__.py index 9a6f38a4d82..98abad58d08 100644 --- a/tests/common/plugins/loganalyzer/__init__.py +++ b/tests/common/plugins/loganalyzer/__init__.py @@ -68,6 +68,13 @@ def loganalyzer(duthosts, request, log_rotate_modular_chassis): yield return + # Skip LogAnalyzer setup for known failing tests + if "test_gnoi_system_reboot_when_reboot_active" in request.node.nodeid or \ + "test_gnoi_system_reboot_status_immediately" in request.node.nodeid: + logging.info("Skipping LogAnalyzer Setup for gnoi_system_reboot tests") + yield + return + # Analyze all the duts fail_test = not (request.config.getoption("--ignore_la_failure")) store_la_logs = request.config.getoption("--store_la_logs") @@ -87,9 +94,12 @@ def loganalyzer(duthosts, request, log_rotate_modular_chassis): yield analyzers - # Skip LogAnalyzer if case is skipped + # Skip LogAnalyzer if case is skipped or if it's a known negative test if "rep_call" in request.node.__dict__ and request.node.rep_call.skipped or \ - "rep_setup" in request.node.__dict__ and request.node.rep_setup.skipped: + "rep_setup" in request.node.__dict__ and request.node.rep_setup.skipped or \ + "test_gnoi_system_reboot_fail_invalid_method" in request.node.nodeid or \ + "test_gnoi_system_reboot_when_reboot_active" in request.node.nodeid or \ + "test_gnoi_system_reboot_status_immediately" in request.node.nodeid: return logging.info("Starting to analyse on all DUTs") parallel_run(analyze_logs, [analyzers, markers], {'fail_test': fail_test, 'store_la_logs': store_la_logs}, diff --git a/tests/gnmi/test_gnoi_system.py b/tests/gnmi/test_gnoi_system.py index ae6e21650ab..44029fe9e43 100644 --- a/tests/gnmi/test_gnoi_system.py +++ b/tests/gnmi/test_gnoi_system.py @@ -4,12 +4,14 @@ from .helper import gnoi_request from tests.common.helpers.assertions import pytest_assert +from tests.common.reboot import wait_for_startup import re pytestmark = [ pytest.mark.topology('any') ] +MAX_TIME_TO_REBOOT = 300 """ This module contains tests for the gNOI System API. @@ -45,7 +47,7 @@ def test_gnoi_system_reboot(duthosts, rand_one_dut_hostname, localhost): duthost.host.options['skip_gnmi_check'] = True # Trigger reboot - ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 1}') + 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)) @@ -56,17 +58,42 @@ def test_gnoi_system_reboot_fail_invalid_method(duthosts, rand_one_dut_hostname, """ duthost = duthosts[rand_one_dut_hostname] + # Set flag to indicate that this test involves reboot + duthost.host.options['skip_gnmi_check'] = True + # Trigger reboot with invalid method - ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 2}') + ret, msg = gnoi_request(duthost, localhost, "Reboot", '{"method": 99}') pytest_assert(ret != 0, "System.Reboot API did not report failure with invalid method") +def test_gnoi_system_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_check'] = 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") + + def test_gnoi_system_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_check'] = 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)) @@ -87,6 +114,37 @@ def test_gnoi_system_reboot_status_immediately(duthosts, rand_one_dut_hostname, 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_check'] = 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, localhost, 0, MAX_TIME_TO_REBOOT) + + # 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. @@ -95,12 +153,21 @@ 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 +