subscription tests and support

confdgnmi/requirements.txt

@@ -0,0 +1,3 @@
+grpcio-tools
+robotframework
+pyyaml

confdgnmi/src/confd_gnmi_client.py

@@ -170,22 +170,26 @@ def read_subscribe_responses(responses, read_count=-1):
         log.info("<==")
 
     # TODO this API would change with more subscription support
-    def subscribe(self, subscription_list, read_fun=None,
+    def subscribe(self, requests, read_fun=None,
                   poll_interval=0.0, poll_count=0, read_count=-1,
                   subscription_end_delay=0.0):
         log.info("==>")
-        request = ConfDgNMIClient.generate_subscriptions(subscription_list, poll_interval,
-                                                         poll_count, subscription_end_delay)
+        if isinstance(requests, list):
+            requests = ConfDgNMIClient.generate_subscriptions(requests,
+                                                              poll_interval,
+                                                              poll_count,
+                                                              subscription_end_delay)
         responses = logged_rpc_call("Subscribe", request,
-                                    lambda: self.stub.Subscribe(request, metadata=self.metadata))
+                                    lambda: self.stub.Subscribe(requests, metadata=self.metadata))
         if read_fun is not None:
             read_fun(responses, read_count)
         log.info("<== responses=%s", responses)
         return responses
 
     def get_public(self,
-                   prefix: Optional[str], paths: list[str],
-                   get_type: Optional[int], encoding: Optional[int]) -> gnmi_pb2.GetResponse:
+                   prefix: Optional[str] = None, paths: list[str] = [],
+                   get_type: Optional[int] = None, encoding: Optional[int] = None) \
+            -> gnmi_pb2.GetResponse:
         sanitized_params = {
             'prefix': None if prefix is None else make_gnmi_path(prefix),
             'paths': [make_gnmi_path(p) for p in paths],

confdgnmi/testtool/defaults.yaml

@@ -0,0 +1,2 @@
+get-path: /ietf-interfaces:interfaces
+secondary-path: /route-status:route-status/route

confdgnmi/testtool/gNMI_Interface/04__Subscribe.robot

@@ -5,90 +5,105 @@ Test Tags        subscribe
 Resource         Subscribe.resource
 
 Resource         gNMIClient.resource
-Suite Setup      Setup gNMI Client
-Suite Teardown   Close gNMI Client
+Test Setup       Setup gNMI Client
+Test Teardown    Teardown gNMI state
 
 
 *** Test Cases ***
 
 Basic subscription with "mode" parameter
     [Tags]    sanity
-    [Documentation]  Test that the device correctly responds to all three
-    ...    "Subscribe" request modes.  No further functionality (such as actually
+    [Documentation]    Test that the device correctly responds to all three
+    ...    "Subscribe" request modes.    No further functionality (such as actually
     ...    polling the device) is tested.
-    [Template]    Verify Subscribe
+    [Template]    Subscribe ${mode} to default path with encoding JSON_IETF
     STREAM
     ONCE
     POLL
 
-Subscribe to non-existent "prefix"
-
-Subscribe to non-existent "path"
-
-# message Subscription {
-#   Path path = 1;                    // The data tree path.
-#   SubscriptionMode mode = 2;        // Subscription mode to be used.
-#   uint64 sample_interval = 3;       // ns between samples in SAMPLE mode.
-#   bool suppress_redundant = 4;
-#   uint64 heartbeat_interval = 5;
-# }
-
 Subscribe ONCE with supported "encoding" values
-
-Two Subscriptions in single request - with same "path"
-
-Two Subscriptions in single request - with different "path"
+    [Tags]    sanity
+    [Documentation]    Verify that the device is able to respond correctly for
+    ...    all declared encodings.
+    Given device capabilities
+    and subscription paths    ${GET-PATH}
+    Then subscribe ONCE with supported encodings
+
+Two subscriptions in single request with the same "path"
+    [Documentation]    Verify that the device can handle ONCE subscription
+    ...    with two identical paths.
+    Given subscription paths    ${GET-PATH}    ${GET-PATH}
+    And subscription ONCE with encoding JSON_IETF
+    Then device responds
+
+Two subscriptions in single request with different "path"
+    [Documentation]    Verify that the device can handle ONCE subscription
+    ...    with two different paths.
+    Given subscription paths    ${GET-PATH}    ${SECONDARY-PATH}
+    And subscription ONCE with encoding JSON_IETF
+    Then device responds
+
+Subscribe ONCE sends final message with "sync_response"
+    [Documentation]    When a ONCE subscription is created, the device must
+    ...    respond with a series of responses terminated by an empty
+    ...    response with "sync_response".
+    Given Subscription paths    ${GET-PATH}
+    And Subscription ONCE with encoding JSON_IETF
+    Then Device sends terminated response series and closes the stream
+
+Subscribe POLL sends final message with "sync_response"
+    [Documentation]    When a POLL subscription is created, the device must
+    ...    send an initial set of responses terminated by an empty
+    ...    response with "sync_response".
+    Given Subscription paths    ${GET-PATH}
+    And Subscription POLL with encoding JSON_IETF
+    Then Device sends terminated response series
+
+Subscribe STREAM sends final message with "sync_response"
+    [Documentation]    When a STREAM subscription is created, the device must
+    ...    send an initial set of responses terminated by an empty
+    ...    response with "sync_response".
+    Given Subscription paths    ${GET-PATH}
+    And Subscription POLL with encoding JSON_IETF
+    Then Device sends terminated response series
 
 QOS marked subscriptions
+    [Tags]    unimplemented
+    # idk., what to test here?
 
 STREAM with TARGET_DEFINED mode
+    [Tags]    unimplemented
+    # same - nothing to test, really
 
 STREAM with ON_CHANGE mode
+    [Tags]    unimplemented
 
 STREAM with SAMPLE mode
-
-
-
-# TODO:
-
-Basic functionality of the Subscribe ONCE
-    [Documentation]    TODO
-    ...  Failure: Device does not respond, responds with an error, responds with
-    ...  an empty notification set or with a notification without updates,
-    ...  responds with incorrect encoding.
-    ...
-    ...  Tags: Rel-1, active, intf; Short: gNMI Subscribe ONCE Parent: GNMI-HLT-010
-    ...
-    ...  Test is passed path to the data model, which contains limited number of elements.
-    ...   ONCE Subscription operation is invoked.
-    ...  SubscriptionRequest does not have set updates_only field.
-    ...  Test verifies for each subscription
-    ...    data is received as a stream of SubscribeResponses
-    ...  Last SubscribeResponse is with sync_response set to true.
-    # TODO
-    # Can SubscribeResponse with sync_response contain also something else, e.g. updates?
+    [Tags]    unimplemented
 
 
 Subscribe ONCE with updates_only in the SubscriptionList
     # Failure: Device does not respond, responds with an error, responds with an empty notification set or with a notification without updates, responds with incorrect encoding.
     # Test is passed path to the data model, which contains limited number of elements. ONCE Subscription operation is invoked with updates_only field set in SubscriptionRequest. Test verifies only one SubscribeResponse with only sync_response set to true.
+    [Tags]    unimplemented
 
 Basic functionality of the Subscribe POLL RPC .
     # Parameters: common connection parameters, path, poll count, poll interval.
     # Failure: Device does not respond, responds with an error, responds with an empty notification set or with a notification without updates, responds with incorrect encoding.
     # Test is passed path to the data model, which contains limited number of elements. Subscription operation is invoked. First SubscriptionRequest is ONCE. After that POLL subscription requests are invoked with poll interval delay. Test verifies for each subscription data is received. Optionally, it can verify the data is the same as the one received in response for ONCE subscription. Test issues WARNING, if SubscriptionResponse stream is closed prematurely. Test issues WARNING if Updates in SubscriptionResponse are aggregated.
+    [Tags]    unimplemented
 
 Subscribe POLL RPC with updates_only in the SubscriptionList.
     # Parameters: common connection parameters, path, poll count, poll interval.
     # Failure: Device does not respond, responds with an error, responds with an empty notification set or with a notification without updates, responds with incorrect encoding.
     # Test is passed path to the data model, which contains limited number of elements. Subscription` operation is invoked. First SubscriptionRequest is ONCE with filled SubscriptionList containing updates_only set to true. This subscription is handled as ONCE subscription. After that, empty POLL subscription requests are invoked with poll interval delay. Test verifies for each subscription (also for initial ONCE) a SubscriptionResponse is received with only sync_response set to true (without other fields). Test issues WARNING, if SubscriptionResponse stream is closed prematurely. Test issues WARNING if Updates in SubscriptionResponse are aggregated.
-
+    [Tags]    unimplemented
 
 Basic functionality of the Subscribe STREAM RPC with ON_CHANGE mode.
     # Parameters: common connection parameters, path, read count
     # Failure: Device does not respond, responds with an error, responds with an empty notification set or with a notification without updates, responds with incorrect encoding.
     # Test is passed path to the data model, which contains limited number of elements, where some of them periodically change (e.g. packet count on interface). STREAM` Subscription operation is invoked. First SubscriptionResponse contains all elements, next responses contains only changed elements. After read count parameter test ends (and subscription stream is closed). Test issues WARNING, if SubscriptionResponse stream is closed prematurely. Test issues WARNING if Updates in SubscriptionResponse are aggregated.
-
+    [TAGS]    unimplemented
 
 # TODO - Michal's backlog:
 
@@ -101,20 +116,4 @@ Other STREAM variants (and combinations):
     # STREAM SAMPLE mode, with heartbeat_interval
     # STREAM SAMPLE mode, with updates_only
     # STREAM SAMPLE mode, with suppress_redundant
-
-Subscription test for all device supported encodings, for unsupported encoding, for non existing encoding.
-
-Subscribe for not existing path
-
-Subscribe for not existing prefix
-
-We will not test TARGET_DEFINED mode
-# TODO - why?
-
-
-#   Path prefix = 1;                          // Prefix used for paths.
-#   repeated Subscription subscription = 2;   // Set of subscriptions to create.
-#   QOSMarking qos = 4;                       // DSCP marking to be used.
-#   bool allow_aggregation = 6;
-#   repeated ModelData use_models = 7;
-#   bool updates_only = 9;
+    [Tags]    unimplemented

confdgnmi/testtool/gNMI_Interface/Subscribe.resource

@@ -6,6 +6,36 @@ Resource         gNMIClient.resource
 
 *** Keywords ***
 
-Verify Subscribe
-    [Arguments]  ${mode}
-    Test Subscribe  ${mode}  ${GET-PATH}
+Subscribe ${mode} to default path with encoding ${encoding}
+    Given subscription paths    ${GET-PATH}
+    And Subscription ${mode} with encoding ${encoding}
+    Then device responds
+
+Subscription ${mode} with encoding ${encoding}
+    Subscribe    ${mode}    ${encoding}
+
+Device responds
+    ${updates}=    Check updates
+    Should be true    ${updates}
+
+Subscribe ONCE with supported encodings
+    @{encodings}=    last supported encodings
+    FOR    ${encoding}    IN    @{encodings}
+        Given subscription ONCE with encoding ${encoding}
+        Then device responds
+    END
+
+Device capabilities
+    Get capabilities from device
+
+Device sends terminated response series and closes the stream
+    Device terminates    True
+
+Device sends terminated response series
+    Device terminates    False
+
+Device terminates
+    [Arguments]    ${should_close}
+    [Timeout]    5 seconds
+    ${terminated}=    Check responses terminated    ${should_close}
+    Should be true    ${terminated}    "The server did not send sync_response"

confdgnmi/testtool/gNMI_Interface/SubscribeLibrary.py

@@ -1,18 +1,95 @@
 from __future__ import annotations
 
-from confd_gnmi_common import make_gnmi_path
+import typing as t
+
+from confd_gnmi_common import make_gnmi_path, encoding_str_to_int
 from confd_gnmi_client import ConfDgNMIClient
-from gNMIRobotLibrary import gNMIRobotLibrary
+from CapabilitiesLibrary import CapabilitiesLibrary
+
+import gnmi_pb2 as gnmi
+
+import threading
+import queue
+
+SlistType = t.Optional[t.Union[gnmi.Poll, gnmi.SubscriptionList]]
+
+
+class Requester(threading.Thread):
+    def __init__(self, client: ConfDgNMIClient) -> None:
+        self.client: ConfDgNMIClient = client
+        self.slist_queue: queue.Queue[SlistType] = queue.Queue()
+        self.response_queue: queue.Queue[gnmi.SubscribeResponse] = queue.Queue()
+        super().__init__()
+
+    def run(self) -> None:
+        # TODO: this dumps the MultiThreaded thing exception for XR
+        for response in self.client.stub.Subscribe(self.requests(),
+                                                   metadata=self.client.metadata):
+            self.response_queue.put(response)
+
+    def requests(self) -> t.Iterator[gnmi.SubscribeRequest]:
+        while (slitem := self.slist_queue.get()) is not None:
+            if isinstance(slitem, gnmi.SubscriptionList):
+                yield gnmi.SubscribeRequest(subscribe=slitem)
+            elif isinstance(slitem, gnmi.Poll):
+                yield gnmi.SubscribeRequest(poll=slitem)
 
+    def enqueue(self, item: SlistType) -> None:
+        self.slist_queue.put(item)
 
-class SubscribeLibrary(gNMIRobotLibrary):
+    def responses(self, wait: bool = True) -> t.Iterator[gnmi.SubscribeResponse]:
+        try:
+            while (response := self.response_queue.get(block=wait)) is not None:
+                yield response
+        except queue.Empty:
+            pass
+
+
+class SubscribeLibrary(CapabilitiesLibrary):
     "ROBOT test suite library for servicing the gNMI SubscribeRequest tests."
     ROBOT_LIBRARY_SCOPE = 'SUITE'
 
-    def test_subscribe(self, mode, path):
-        slist = ConfDgNMIClient.make_subscription_list(make_gnmi_path(''),
-                                                       [make_gnmi_path(path)],
-                                                       mode,
-                                                       encoding_str_to_int('JSON_IETF'))
-        responses = self._client.subscribe(slist)
-        return next(responses)
+    def __init__(self) -> None:
+        self.paths: t.List[str] = []
+        self.requester: t.Optional[Requester] = None
+
+    def test_teardown(self) -> None:
+        self.paths = []
+        if self.requester is not None:
+            self.requester.enqueue(None)
+        super().test_teardown()
+        if self.requester is not None:
+            self.requester.join(2)
+        self.requester = None
+
+    def subscribe(self, mode: str, encoding: str) -> None:
+        paths = [make_gnmi_path(path) for path in self.paths]
+        iencoding = encoding_str_to_int(encoding)
+        slist = ConfDgNMIClient.make_subscription_list(make_gnmi_path(''), paths, mode, iencoding)
+        self.requester = Requester(self._client)
+        self.requester.start()
+        self.requester.enqueue(slist)
+
+    def check_updates(self) -> bool:
+        # return the first notification update in the first nonempty response
+        try:
+            next(update
+                 for response in self.requester.responses()
+                 for update in response.update.update)
+            return True
+        except StopIteration:
+            return False
+
+    def subscription_paths(self, *paths: str) -> None:
+        self.paths = paths
+
+    def check_responses_terminated(self, should_close: bool) -> bool:
+        terminated = False
+        for response in self.requester.responses():
+            assert not terminated, 'The server did not close the stream'
+            if response.sync_response:
+                terminated = True
+                if not should_close:
+                    # the stream does not need to be closed, exit now
+                    break
+        return terminated

confdgnmi/testtool/xr.yaml

@@ -10,3 +10,6 @@ enable_extra_logs: False # configuration of gNMI library - whether to enable int
 
 # test-case specific settings
 oc_interface: MgmtEth0/RP0/CPU0/0
+
+get-path: /interfaces
+secondary-path: /components