Add on change update test case to test_telemetry (sonic-net#9073) (so…

…nic-net#11700)

Cherry pick of Add on change update test case to test_telemetry (sonic-net#9073)

tests/telemetry/telemetry_utils.py

@@ -16,6 +16,9 @@
 SUBMODE_SAMPLE = 2
 SUBMODE_ONCHANGE = 1
 
+EVENT_REGEX = "json_ietf_val: \"(.*)\""
+ON_CHANGE_REGEX = "json_ietf_val:\"({.*?})\""
+
 
 def assert_equal(actual, expected, message):
     """Helper method to compare an expected value vs the actual value.
@@ -74,6 +77,27 @@ def restore_telemetry_forpyclient(duthost, default_client_auth):
         duthost.service(name=env.gnmi_container, state="restarted")
 
 
+def check_gnmi_cli_running(ptfhost):
+    program_list = ptfhost.shell("pgrep -f 'python /root/gnxi/gnmi_cli_py/py_gnmicli.py'")["stdout"]
+    return len(program_list) > 0
+
+
+def parse_gnmi_output(gnmi_output, match_no, find_data):
+    gnmi_str = str(gnmi_output)
+    gnmi_str = gnmi_str.replace('\\', '')
+    gnmi_str = gnmi_str.replace(' ', '')
+    if find_data != "":
+        result = fetch_json_ptf_output(ON_CHANGE_REGEX, gnmi_str, match_no)
+        return find_data in result
+
+
+def fetch_json_ptf_output(regex, output, match_no):
+    match = re.findall(regex, output)
+    assert len(match) > match_no, "Not able to parse json from output"
+    event_str = match[match_no]
+    return event_str
+
+
 def listen_for_event(ptfhost, cmd, results):
     ret = ptfhost.shell(cmd)
     assert ret["rc"] == 0, "PTF docker was not able to query EVENTS path"
@@ -91,9 +115,7 @@ def listen_for_events(duthost, gnxi_path, ptfhost, filter_event_regex, op_file,
     assert results[0] != "", "No output from PTF docker, thread timed out after {} seconds".format(thread_timeout)
     # regex logic and then to write to file
     result = results[0]
-    match = re.findall('json_ietf_val: \"(.*)\"', result)
-    assert len(match) > 0, "Not able to parse json from output"
-    event_str = match[0]
+    event_str = fetch_json_ptf_output(EVENT_REGEX, result, 0)
     event_str = event_str.replace('\\', '')
     event_json = json.loads(event_str)
     with open(op_file, "w") as f:

tests/telemetry/test_telemetry.py

@@ -3,11 +3,13 @@
 import logging
 import re
 import pytest
+import random
 
 from tests.common.helpers.assertions import pytest_assert
 from tests.common.helpers.gnmi_utils import GNMIEnvironment
+from tests.common.utilities import wait_until
 from telemetry_utils import assert_equal, get_list_stdout, get_dict_stdout, skip_201911_and_older
-from telemetry_utils import generate_client_cli
+from telemetry_utils import generate_client_cli, parse_gnmi_output, check_gnmi_cli_running
 
 pytestmark = [
     pytest.mark.topology('any')
@@ -19,6 +21,7 @@
 METHOD_GET = "get"
 MEMORY_CHECKER_WAIT = 1
 MEMORY_CHECKER_CYCLES = 60
+SUBMODE_ONCHANGE = 1
 
 
 def test_config_db_parameters(duthosts, enum_rand_one_per_hwsku_hostname):
@@ -187,8 +190,45 @@ def test_virtualdb_table_streaming(duthosts, enum_rand_one_per_hwsku_hostname, p
                  "Timestamp markers for each update message in:\n{0}".format(result))
 
 
-def invoke_py_cli_from_ptf(ptfhost, cmd):
-    ptfhost.shell(cmd)
+def invoke_py_cli_from_ptf(ptfhost, cmd, results):
+    ret = ptfhost.shell(cmd)
+    assert ret["rc"] == 0, "PTF docker did not get a response"
+    if results is not None and len(results) > 0:
+        results[0] = ret["stdout"]
+
+
+def test_on_change_updates(duthosts, enum_rand_one_per_hwsku_hostname, ptfhost, localhost, gnxi_path):
+    logger.info("Testing on change update notifications")
+
+    duthost = duthosts[enum_rand_one_per_hwsku_hostname]
+    skip_201911_and_older(duthost)
+    cmd = generate_client_cli(duthost=duthost, gnxi_path=gnxi_path, method=METHOD_SUBSCRIBE,
+                              submode=SUBMODE_ONCHANGE, update_count=2, xpath="NEIGH_STATE_TABLE",
+                              target="STATE_DB")
+    results = [""]
+
+    bgp_nbrs = list(duthost.get_bgp_neighbors().keys())
+    bgp_neighbor = random.choice(bgp_nbrs)
+    bgp_info = duthost.get_bgp_neighbor_info(bgp_neighbor)
+    original_state = bgp_info["bgpState"]
+    new_state = "Established" if original_state.lower() == "active" else "Active"
+
+    client_thread = threading.Thread(target=invoke_py_cli_from_ptf, args=(ptfhost, cmd, results,))
+    client_thread.start()
+
+    wait_until(5, 1, 0, check_gnmi_cli_running, ptfhost)
+    duthost.shell("sonic-db-cli STATE_DB HSET \"NEIGH_STATE_TABLE|{}\" \"state\" {}".format(bgp_neighbor,
+                                                                                            new_state))
+
+    client_thread.join(30)
+
+    try:
+        assert results[0] != "", "Did not get output from PTF client"
+    finally:
+        duthost.shell("sonic-db-cli STATE_DB HSET \"NEIGH_STATE_TABLE|{}\" \"state\" {}".format(bgp_neighbor,
+                                                                                                original_state))
+    ret = parse_gnmi_output(results[0], 1, bgp_neighbor)
+    assert ret is True, "Did not find key in update"
 
 
 @pytest.mark.disable_loganalyzer
@@ -203,7 +243,7 @@ def test_mem_spike(duthosts, rand_one_dut_hostname, ptfhost, gnxi_path):
 
     cmd = generate_client_cli(duthost=duthost, gnxi_path=gnxi_path, method=METHOD_SUBSCRIBE,
                               xpath="DOCKER_STATS", target="STATE_DB", update_count=1, create_connections=2000)
-    client_thread = threading.Thread(target=invoke_py_cli_from_ptf, args=(ptfhost, cmd,))
+    client_thread = threading.Thread(target=invoke_py_cli_from_ptf, args=(ptfhost, cmd, None))
     client_thread.start()
 
     for i in range(MEMORY_CHECKER_CYCLES):