Resolve "Fix OFC'24 end-to-end test for Optical Controller component"

Closes #145 (closed)

src/service/service/tools/OpticalTools.py

@@ -13,35 +13,58 @@
 # limitations under the License.
 # 
 
-import json
-import requests
-import uuid
-from common.Constants import *
+import functools, json, logging, requests, uuid
 from typing import List
+from common.Constants import ServiceNameEnum
 from common.proto.context_pb2 import(
     Device, DeviceId, Service, Connection, EndPointId, TopologyId, ContextId, Uuid,
     ConfigRule, ConfigActionEnum, ConfigRule_Custom
 )
 from common.proto.pathcomp_pb2 import PathCompReply
 from common.Settings import (
-    ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, find_environment_variables, get_env_var_name
+    ENVVAR_SUFIX_SERVICE_BASEURL_HTTP, ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC,
+    find_environment_variables, get_env_var_name
+)
+from service.service.tools.replies import (
+    reply_uni_txt, optical_band_uni_txt, reply_bid_txt, optical_band_bid_txt
 )
-from service.service.tools.replies import reply_uni_txt, optical_band_uni_txt, reply_bid_txt, optical_band_bid_txt
 
 log = logging.getLogger(__name__)
 
-testing = False
+TESTING = False
+
+get_optical_controller_setting = functools.partial(get_env_var_name, ServiceNameEnum.OPTICALCONTROLLER)
+VAR_NAME_OPTICAL_CTRL_BASEURL_HTTP = get_optical_controller_setting(ENVVAR_SUFIX_SERVICE_BASEURL_HTTP)
+VAR_NAME_OPTICAL_CTRL_SCHEMA       = get_optical_controller_setting('SCHEMA')
+VAR_NAME_OPTICAL_CTRL_HOST         = get_optical_controller_setting(ENVVAR_SUFIX_SERVICE_HOST)
+VAR_NAME_OPTICAL_CTRL_PORT         = get_optical_controller_setting(ENVVAR_SUFIX_SERVICE_PORT_GRPC)
+
+OPTICAL_CTRL_BASE_URL = '{:s}://{:s}:{:s}/OpticalTFS'
+
+def get_optical_controller_base_url() -> str:
+    settings = find_environment_variables([
+        VAR_NAME_OPTICAL_CTRL_BASEURL_HTTP,
+        VAR_NAME_OPTICAL_CTRL_SCHEMA,
+        VAR_NAME_OPTICAL_CTRL_HOST,
+        VAR_NAME_OPTICAL_CTRL_PORT,
+    ])
+    base_url = settings.get(VAR_NAME_OPTICAL_CTRL_BASEURL_HTTP)
+    if base_url is not None:
+        log.debug('Optical Controller: base_url={:s}'.format(str(base_url)))
+        return base_url
+
+    host = settings.get(VAR_NAME_OPTICAL_CTRL_HOST)
+    port = int(settings.get(VAR_NAME_OPTICAL_CTRL_PORT, 80))
+
+    MSG = 'Optical Controller not found: settings={:s}'
+    if host is None: raise Exception(MSG.format(str(settings)))
+    if port is None: raise Exception(MSG.format(str(settings)))
 
-VAR_NAME_OPTICAL_CONTROLLER_HOST = get_env_var_name(ServiceNameEnum.OPTICALCONTROLLER, ENVVAR_SUFIX_SERVICE_HOST)
-VAR_NAME_OPTICAL_CONTROLLER_PORT = get_env_var_name(ServiceNameEnum.OPTICALCONTROLLER, ENVVAR_SUFIX_SERVICE_PORT_GRPC)
+    schema = settings.get(VAR_NAME_OPTICAL_CTRL_SCHEMA, 'http')
+    base_url = OPTICAL_CTRL_BASE_URL.format(schema, host, port)
+    log.debug('Optical Controller: base_url={:s}'.format(str(base_url)))
+    return base_url
 
-opticalcontrollers_url = find_environment_variables([
-    VAR_NAME_OPTICAL_CONTROLLER_HOST,
-    VAR_NAME_OPTICAL_CONTROLLER_PORT,
-])
-OPTICAL_IP   = opticalcontrollers_url.get(VAR_NAME_OPTICAL_CONTROLLER_HOST)
-OPTICAL_PORT = opticalcontrollers_url.get(VAR_NAME_OPTICAL_CONTROLLER_PORT)
-log.info(str(OPTICAL_IP), str(OPTICAL_PORT))
 
 def get_uuids_from_names(devices: List[Device], device_name: str, port_name: str):
     device_uuid = ""
@@ -79,17 +102,18 @@ def get_device_name_from_uuid(devices: List[Device], device_uuid: str):
 
 
 def add_lightpath(src, dst, bitrate, bidir, ob_band) -> str:
-    if not testing:
+    if not TESTING:
         urlx = ""
         headers = {"Content-Type": "application/json"}
+        base_url = get_optical_controller_base_url()
         if ob_band is None:
             if bidir is None:
                 bidir = 1
-            urlx = "http://{}:{}/OpticalTFS/AddFlexLightpath/{}/{}/{}/{}".format(OPTICAL_IP, OPTICAL_PORT, src, dst, bitrate, bidir)
+            urlx = "{:s}/AddFlexLightpath/{:s}/{:s}/{:s}/{:s}".format(base_url, src, dst, bitrate, bidir)
         else:
             if bidir is None:
                 bidir = 1
-            urlx = "http://{}:{}/OpticalTFS/AddFlexLightpath/{}/{}/{}/{}/{}".format(OPTICAL_IP, OPTICAL_PORT, src, dst, bitrate, bidir, ob_band)
+            urlx = "{:s}/AddFlexLightpath/{:s}/{:s}/{:s}/{:s}/{:s}".format(base_url, src, dst, bitrate, bidir, ob_band)
         r = requests.put(urlx, headers=headers)
         reply = r.text 
         return reply
@@ -101,8 +125,9 @@ def add_lightpath(src, dst, bitrate, bidir, ob_band) -> str:
                 
 
 def get_optical_band(idx) -> str:
-    if not testing:
-        urlx = "http://{}:{}/OpticalTFS/GetOpticalBand/{}".format(OPTICAL_IP, OPTICAL_PORT, idx)
+    if not TESTING:
+        base_url = get_optical_controller_base_url()
+        urlx = "{:s}/GetOpticalBand/{:s}".format(base_url, idx)
         headers = {"Content-Type": "application/json"}
         r = requests.get(urlx, headers=headers)
         reply = r.text 
@@ -116,8 +141,9 @@ def get_optical_band(idx) -> str:
     
 def delete_lightpath(flow_id, src, dst, bitrate) -> str:
     reply = "200"
-    if not testing:
-        urlx = "http://{}:{}/OpticalTFS/DelLightpath/{}/{}/{}/{}".format(OPTICAL_IP, OPTICAL_PORT, flow_id, src, dst, bitrate)
+    if not TESTING:
+        base_url = get_optical_controller_base_url()
+        urlx = "{:s}/DelLightpath/{:s}/{:s}/{:s}/{:s}".format(base_url, flow_id, src, dst, bitrate)
 
         headers = {"Content-Type": "application/json"}
         r = requests.delete(urlx, headers=headers)
@@ -126,7 +152,8 @@ def delete_lightpath(flow_id, src, dst, bitrate) -> str:
 
 
 def get_lightpaths() -> str:
-    urlx = "http://{}:{}/OpticalTFS/GetLightpaths".format(OPTICAL_IP, OPTICAL_PORT)
+    base_url = get_optical_controller_base_url()
+    urlx = "{:s}/GetLightpaths".format(base_url)
 
     headers = {"Content-Type": "application/json"}
     r = requests.get(urlx, headers=headers)