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 9d3bfec3eb62c86a8518b42363a68fb906905df0..1026df7e0583d20f743b32f57bd84a11fa224be5 100644 (file)
--- a/tests/transportpce_tests/common/test_utils.py
+++ b/tests/transportpce_tests/common/test_utils.py
@@ -11,7 +11,6 @@
 
 # pylint: disable=no-member
 
-import base64
 import json
 import os
 import sys
@@ -46,12 +45,14 @@ 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"
 URL_OTN_SERVICE_PATH = "{}/operations/transportpce-device-renderer:otn-service-path"
 URL_CREATE_OTS_OMS = "{}/operations/transportpce-device-renderer:create-ots-oms"
 URL_PATH_COMPUTATION_REQUEST = "{}/operations/transportpce-pce:path-computation-request"
+URL_FULL_PORTMAPPING = "{}/config/transportpce-portmapping:network"
 
 TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'}
 TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
@@ -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),
@@ -195,6 +198,13 @@ def rawput_request(url, data):
         headers=TYPE_APPLICATION_JSON,
         auth=(ODL_LOGIN, ODL_PWD))
 
+def rawpost_request(url, data):
+    return requests.request(
+        "POST", url.format(RESTCONF_BASE_URL),
+        data=data,
+        headers=TYPE_APPLICATION_JSON,
+        auth=(ODL_LOGIN, ODL_PWD))
+
 
 def delete_request(url):
     return requests.request(
@@ -328,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)
@@ -352,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)
 
 
@@ -465,20 +487,3 @@ class TimeOut:
         # pylint: disable=W0622
         signal.alarm(0)
 
-
-def check_freq_map(freq_map):
-    freq_map_array = [int(x) for x in freq_map]
-    return freq_map_array[0] == 255 and freq_map_array[1] == 255
-
-
-def set_used_index_for_freq_map(freq_map, index):
-    freq_map[index] = 0
-    return freq_map
-
-
-INDEX_1_USED_FREQ_MAP = base64.b64encode(set_used_index_for_freq_map(bytearray(b'\xFF' * 96), 0)).decode('UTF-8')
-
-INDEX_1_2_USED_FREQ_MAP = base64.b64encode(set_used_index_for_freq_map(
-    set_used_index_for_freq_map(bytearray(b'\xFF' * 96), 0), 1)).decode('utf-8')
-
-AVAILABLE_FREQ_MAP = base64.b64encode(bytearray(b'\xFF' * 96)).decode('UTF-8')