Add tests for service Notifications

[transportpce.git] / tests / transportpce_tests / common / test_utils.py

diff --git a/tests/transportpce_tests/common/test_utils.py b/tests/transportpce_tests/common/test_utils.py
index fc8a4f69cec71e514d4df31e7d3ae6ceb9438bcf..1026df7e0583d20f743b32f57bd84a11fa224be5 100644 (file)
--- a/tests/transportpce_tests/common/test_utils.py
+++ b/tests/transportpce_tests/common/test_utils.py
@@ -45,6 +45,7 @@ URL_CONFIG_CLLI_NET = "{}/config/ietf-network:networks/network/clli-network/"
 URL_CONFIG_ORDM_NET = "{}/config/ietf-network:networks/network/openroadm-network/"
 URL_PORTMAPPING = "{}/config/transportpce-portmapping:network/nodes/"
 URL_OPER_SERV_LIST = "{}/operational/org-openroadm-service:service-list/"
+URL_GET_NBINOTIFICATIONS_SERV = "{}/operations/nbi-notifications:get-notifications-service/"
 URL_SERV_CREATE = "{}/operations/org-openroadm-service:service-create"
 URL_SERV_DELETE = "{}/operations/org-openroadm-service:service-delete"
 URL_SERVICE_PATH = "{}/operations/transportpce-device-renderer:service-path"
@@ -137,7 +138,8 @@ def install_karaf_feature(feature_name: str):
         os.path.dirname(os.path.realpath(__file__)),
         "..", "..", "..", "karaf", "target", "assembly", "bin", "client")
     return subprocess.run([executable],
-                          input='feature:install ' + feature_name + '\n feature:list | grep tapi \n logout \n',
+                          input='feature:install ' + feature_name + '\n feature:list | grep '
+                          + feature_name + ' \n logout \n',
                           universal_newlines=True, check=False)
 
 
@@ -150,6 +152,7 @@ def get_request(url):
 
 def post_request(url, data):
     if data:
+        print(json.dumps(data))
         return requests.request(
             "POST", url.format(RESTCONF_BASE_URL),
             data=json.dumps(data),
@@ -335,6 +338,10 @@ def portmapping_request(suffix: str):
     return get_request(url)
 
 
+def get_notifications_service_request(attr):
+    return post_request(URL_GET_NBINOTIFICATIONS_SERV, attr)
+
+
 def get_service_list_request(suffix: str):
     url = URL_OPER_SERV_LIST + suffix
     return get_request(url)
@@ -359,13 +366,21 @@ def service_delete_request(servicename: str,
     return post_request(URL_SERV_DELETE, attr)
 
 
-def service_path_request(operation: str, servicename: str, wavenumber: str, nodes):
+def service_path_request(operation: str, servicename: str, wavenumber: str, nodes, centerfreq: str,
+                         slotwidth: int, minfreq: float, maxfreq: float, lowerslotnumber: int,
+                         higherslotnumber: int):
     attr = {"renderer:input": {
         "renderer:service-name": servicename,
         "renderer:wave-number": wavenumber,
-        "renderer:modulation-format": "qpsk",
+        "renderer:modulation-format": "dp-qpsk",
         "renderer:operation": operation,
-        "renderer:nodes": nodes}}
+        "renderer:nodes": nodes,
+        "renderer:center-freq": centerfreq,
+        "renderer:width": slotwidth,
+        "renderer:min-freq": minfreq,
+        "renderer:max-freq": maxfreq,
+        "renderer:lower-spectral-slot-number": lowerslotnumber,
+        "renderer:higher-spectral-slot-number": higherslotnumber}}
     return post_request(URL_SERVICE_PATH, attr)