diff --git a/tests/gnmi/helper.py b/tests/gnmi/helper.py index e20b90a44f6..dd1132ef8d4 100644 --- a/tests/gnmi/helper.py +++ b/tests/gnmi/helper.py @@ -419,7 +419,7 @@ def gnoi_reboot(duthost, method, delay, message): return 0, output['stdout'] -def gnoi_request(duthost, localhost, module, rpc, request_json_data): +def gnoi_request(duthost, localhost, module, rpc, request_json_data, input_data=None): env = GNMIEnvironment(duthost, GNMIEnvironment.GNMI_MODE) ip = duthost.mgmt_ip port = env.gnmi_port @@ -428,7 +428,12 @@ def gnoi_request(duthost, localhost, module, rpc, request_json_data): cmd += "-key /etc/sonic/telemetry/gnmiclient.key " cmd += "-ca /etc/sonic/telemetry/gnmiCA.pem " cmd += "-logtostderr -module {} -rpc {} ".format(module, rpc) - cmd += f'-jsonin \'{request_json_data}\'' + if request_json_data: + cmd += f'-jsonin \'{request_json_data}\'' + + if input_data: + cmd += input_data + output = duthost.shell(cmd, module_ignore_errors=True) if output['stderr']: logger.error(output['stderr']) diff --git a/tests/gnmi/test_gnoi_healthz.py b/tests/gnmi/test_gnoi_healthz.py new file mode 100644 index 00000000000..cb63b8505d0 --- /dev/null +++ b/tests/gnmi/test_gnoi_healthz.py @@ -0,0 +1,193 @@ +import pytest +import logging +import re + +from .helper import gnoi_request +from tests.common.helpers.assertions import pytest_assert + +pytestmark = [pytest.mark.topology("any")] + +""" +This module contains tests for the gNOI Healthz API +""" + +# Below were the current supported containers which supports Healthz RPC's. +containers = ["gnmi", "orch"] + + +@pytest.mark.disable_loganalyzer +def test_gnoi_healthz_get_alert_info(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI Healthz is retrieving the latest health status for a gNMI alert-info path + """ + + duthost = duthosts[rand_one_dut_hostname] + + for container in containers: + + request_json1 = f'{{\"path\":\"/components/component[name={container}]/healthz/alert-info\"}}' + ret, msg = gnoi_request(duthost, localhost, "Healthz", "Get", request_json1) + pytest_assert(ret == 0, f"Healthz.Get-alert-info RPC failed: rc = {ret}, msg = {msg}") + + logging.info("Healthz.Get-alert-info API returned msg: {}".format(msg)) + # Assert that the response contains Status: STATUS_HEALTHY + pytest_assert("STATUS_HEALTHY" in msg, "Healthz.Get-alert_info did not return correct HEALTH status") + + # Now invoking Aritifact RPC + file_pattern_match = re.search(r"File Name:\s*(.*)", msg) + file_name = file_pattern_match.group(1).strip() + id = f' -id "{file_name}"' + + ret1, msg1 = gnoi_request(duthost, localhost, "Healthz", "Artifact", "", id) + pytest_assert(ret1 == 0, f"Healthz.Get-alert-info RPC failed: rc = {ret1}, msg = {msg1}") + + logging.info("Healthz.Get-alert-info API returned msg: {}".format(msg1)) + # Assert that the response contains Status: Response Success + pytest_assert( + "Artifact Response success" in msg1, + "Healthz.Get-alert_info did not return correct Artificat status" + ) + + # Now invoking Acknowledge RPC + request_json2 = ( + f'{{\"path\": \"/components/component[name={container}]/healthz/alert-info\", \"id\": \"{file_name}\"}}' + ) + ret2, msg2 = gnoi_request(duthost, localhost, "Healthz", "Acknowledge", request_json2) + pytest_assert(ret2 == 0, f"Healthz.Get-alert-info RPC failed: rc = {ret2}, msg = {msg2}") + + logging.info("Healthz.Get-alert-info API returned msg: {}".format(msg2)) + # Assert that the response contains Status: Acknowledge resonse nil + pytest_assert( + "Acknowledge response: " in msg2, + "Healthz.Get-alert_info did not return correct Acknowledge status" + ) + + +@pytest.mark.disable_loganalyzer +def test_gnoi_healthz_get_all_info(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI Healthz is retrieving the latest health status for a gNMI all-info path + """ + + duthost = duthosts[rand_one_dut_hostname] + + for container in containers: + + request_json = f'{{\"path\":\"/components/component[name={container}]/healthz/all-info\"}}' + ret, msg = gnoi_request(duthost, localhost, "Healthz", "Get", request_json) + pytest_assert(ret == 0, f"Healthz.Get-all-info RPC failed: rc = {ret}, msg = {msg}") + + logging.info("Healthz.Get-all-info API returned msg: {}".format(msg)) + # Assert that the response contains Status: STATUS_HEALTHY + pytest_assert("STATUS_HEALTHY" in msg, "Healthz.Get-all_info did not return correct HEALTH status") + + # Now invoking Aritifact RPC + file_pattern_match = re.search(r"File Name:\s*(.*)", msg) + file_name = file_pattern_match.group(1).strip() + id = f' -id "{file_name}"' + + ret1, msg1 = gnoi_request(duthost, localhost, "Healthz", "Artifact", "", id) + pytest_assert(ret1 == 0, f"Healthz.Get-all-info RPC failed: rc = {ret1}, msg = {msg1}") + + logging.info("Healthz.Get-all-info API returned msg: {}".format(msg1)) + # Assert that the response contains Status: Response Success + pytest_assert( + "Artifact Response success" in msg1, + "Healthz.Get-all_info did not return correct Artificat status" + ) + + # Now invoking Acknowledge RPC + request_json2 = ( + f'{{\"path\": \"/components/component[name={container}]/healthz/all-info\", \"id\": \"{file_name}\"}}' + ) + ret2, msg2 = gnoi_request(duthost, localhost, "Healthz", "Acknowledge", request_json2) + pytest_assert(ret2 == 0, f"Healthz.Get-all-info RPC failed: rc = {ret2}, msg = {msg2}") + + logging.info("Healthz.Get-all-info API returned msg: {}".format(msg2)) + # Assert that the response contains Status: Acknowledge resonse nil + pytest_assert( + "Acknowledge response: " in msg2, + "Healthz.Get-all_info did not return correct Acknowledge status" + ) + + +@pytest.mark.disable_loganalyzer +def test_gnoi_healthz_get_critical_info(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI Healthz is retrieving the latest health status for a gNMI critical-info path + """ + + duthost = duthosts[rand_one_dut_hostname] + + for container in containers: + + request_json = f'{{\"path\":\"/components/component[name={container}]/healthz/critical-info\"}}' + ret, msg = gnoi_request(duthost, localhost, "Healthz", "Get", request_json) + pytest_assert(ret == 0, f"Healthz.Get-critical-info RPC failed: rc = {ret}, msg = {msg}") + + logging.info("Healthz.Get-critical-info API returned msg: {}".format(msg)) + # Assert that the response contains Status: STATUS_HEALTHY + pytest_assert("STATUS_HEALTHY" in msg, "Healthz.Get-critical_info did not return correct HEALTH status") + + # Now invoking Aritifact RPC + file_pattern_match = re.search(r"File Name:\s*(.*)", msg) + file_name = file_pattern_match.group(1).strip() + id = f' -id "{file_name}"' + + ret1, msg1 = gnoi_request(duthost, localhost, "Healthz", "Artifact", "", id) + pytest_assert(ret1 == 0, f"Healthz.Get-critical-info RPC failed: rc = {ret1}, msg = {msg1}") + + logging.info("Healthz.Get-critical-info API returned msg: {}".format(msg1)) + # Assert that the response contains Status: Response Success + pytest_assert( + "Artifact Response success" in msg1, + "Healthz.Get-critical_info did not return correct Artificat status" + ) + + # Now invoking Acknowledge RPC + request_json2 = ( + f'{{\"path\": \"/components/component[name={container}]/healthz/critical-info\", \"id\": \"{file_name}\"}}' + ) + ret2, msg2 = gnoi_request(duthost, localhost, "Healthz", "Acknowledge", request_json2) + pytest_assert(ret2 == 0, f"Healthz.Get-critical-info RPC failed: rc = {ret2}, msg = {msg2}") + + logging.info("Healthz.Get-critical-info API returned msg: {}".format(msg2)) + # Assert that the response contains Status: Acknowledge resonse nil + pytest_assert( + "Acknowledge response: " in msg2, + "Healthz.Get-critical_info did not return correct Acknowledge status" + ) + + +@pytest.mark.disable_loganalyzer +def test_gnoi_healthz_list(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI Healthz List RPC is returning Method Unimplimented Error + """ + + duthost = duthosts[rand_one_dut_hostname] + + request_json = '{\"path\":\"/components/component[name=healthz]\", \"include_acknowledged\": true}' + ret, msg = gnoi_request(duthost, localhost, "Healthz", "List", request_json) + pytest_assert(ret == 0, f"Healthz.List RPC failed: rc = {ret}, msg = {msg}") + + logging.info("Healtlz.List API returned msg: {}".format(msg)) + # Assert that the response contains "not implimented Error" + pytest_assert("not implemented" in msg, "Expected method unimplemented error") + + +@pytest.mark.disable_loganalyzer +def test_gnoi_healthz_check(duthosts, rand_one_dut_hostname, localhost): + """ + Verify the gNOI Healthz Check RPC is returning Method Unimplimented Error + """ + + duthost = duthosts[rand_one_dut_hostname] + + request_json = '{\"path\":\"/components/component[name=healthz]\", \"event_id\": \"event-abc123\"}' + ret, msg = gnoi_request(duthost, localhost, "Healthz", "Check", request_json) + pytest_assert(ret == 0, f"Healthz.Check RPC failed: rc = {ret}, msg = {msg}") + + logging.info("Healtlz.Check API returned msg: {}".format(msg)) + # Assert that the response contains "not implimented Error" + pytest_assert("not implemented" in msg, "Expected method unimplemented error")