diff --git a/src/device/service/drivers/__init__.py b/src/device/service/drivers/__init__.py
index 70ca1764f7b82c06f9f79c157cf097c4fda18d6d..2169881af366e0e05e670ec5e2a1552c5dd67f93 100644
--- a/src/device/service/drivers/__init__.py
+++ b/src/device/service/drivers/__init__.py
@@ -82,7 +82,7 @@ DRIVERS.append(
]))
-from .ietf_l3vpn.driver import IetfL3VpnDriver # pylint: disable=wrong-import-position
+from .ietf_l3vpn.IetfL3VpnDriver import IetfL3VpnDriver # pylint: disable=wrong-import-position
DRIVERS.append(
(IetfL3VpnDriver, [
{
diff --git a/src/device/service/drivers/ietf_l2vpn/TfsApiClient.py b/src/device/service/drivers/ietf_l2vpn/TfsApiClient.py
index dd6924de0464131a81d6a4abaff8dc17c4bef67f..e9b21f2498ba37abd8d45f278724c7afc82b4573 100644
--- a/src/device/service/drivers/ietf_l2vpn/TfsApiClient.py
+++ b/src/device/service/drivers/ietf_l2vpn/TfsApiClient.py
@@ -13,21 +13,13 @@
# limitations under the License.
import logging, requests
-from requests.auth import HTTPBasicAuth
from typing import Dict, List, Optional
+from common.tools.client.RestClient import RestClient
from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum
-GET_DEVICES_URL = '{:s}://{:s}:{:d}/tfs-api/devices'
-GET_LINKS_URL = '{:s}://{:s}:{:d}/tfs-api/links'
-
-TIMEOUT = 30
-
-HTTP_OK_CODES = {
- 200, # OK
- 201, # Created
- 202, # Accepted
- 204, # No Content
-}
+GET_CONTEXT_IDS_URL = '/tfs-api/context_ids'
+GET_DEVICES_URL = '/tfs-api/devices'
+GET_LINKS_URL = '/tfs-api/links'
MAPPING_STATUS = {
'DEVICEOPERATIONALSTATUS_UNDEFINED': 0,
@@ -54,23 +46,23 @@ MAPPING_DRIVER = {
'DEVICEDRIVER_NCE' : 15,
}
-MSG_ERROR = 'Could not retrieve devices in remote TeraFlowSDN instance({:s}). status_code={:s} reply={:s}'
-
LOGGER = logging.getLogger(__name__)
-class TfsApiClient:
+class TfsApiClient(RestClient):
def __init__(
self, address : str, port : int, scheme : str = 'http',
- username : Optional[str] = None, password : Optional[str] = None
+ username : Optional[str] = None, password : Optional[str] = None,
+ timeout : Optional[int] = 30
) -> None:
- self._devices_url = GET_DEVICES_URL.format(scheme, address, port)
- self._links_url = GET_LINKS_URL.format(scheme, address, port)
- self._auth = (
- HTTPBasicAuth(username, password)
- if username is not None and password is not None
- else None
+ super().__init__(
+ address, port, scheme=scheme, username=username, password=password,
+ timeout=timeout, verify_certs=False, allow_redirects=True, logger=LOGGER
)
+ def check_credentials(self) -> None:
+ self.get(GET_CONTEXT_IDS_URL, expected_status_codes={requests.codes['OK']})
+ LOGGER.info('Credentials checked')
+
def get_devices_endpoints(
self, import_topology : ImportTopologyEnum = ImportTopologyEnum.DEVICES
) -> List[Dict]:
@@ -78,20 +70,14 @@ class TfsApiClient:
MSG = '[get_devices_endpoints] import_topology={:s}'
LOGGER.debug(MSG.format(str(import_topology)))
- reply = requests.get(self._devices_url, timeout=TIMEOUT, verify=False, auth=self._auth)
- if reply.status_code not in HTTP_OK_CODES:
- msg = MSG_ERROR.format(
- str(self._devices_url), str(reply.status_code), str(reply)
- )
- LOGGER.error(msg)
- raise Exception(msg)
-
if import_topology == ImportTopologyEnum.DISABLED:
MSG = 'Unsupported import_topology mode: {:s}'
raise Exception(MSG.format(str(import_topology)))
+ devices = self.get(GET_DEVICES_URL, expected_status_codes={requests.codes['OK']})
+
result = list()
- for json_device in reply.json()['devices']:
+ for json_device in devices['devices']:
device_uuid : str = json_device['device_id']['device_uuid']['uuid']
device_type : str = json_device['device_type']
#if not device_type.startswith('emu-'): device_type = 'emu-' + device_type
@@ -124,15 +110,9 @@ class TfsApiClient:
LOGGER.debug('[get_devices_endpoints] devices only; returning')
return result
- reply = requests.get(self._links_url, timeout=TIMEOUT, verify=False, auth=self._auth)
- if reply.status_code not in HTTP_OK_CODES:
- msg = MSG_ERROR.format(
- str(self._links_url), str(reply.status_code), str(reply)
- )
- LOGGER.error(msg)
- raise Exception(msg)
+ links = self.get(GET_LINKS_URL, expected_status_codes={requests.codes['OK']})
- for json_link in reply.json()['links']:
+ for json_link in links['links']:
link_uuid : str = json_link['link_id']['link_uuid']['uuid']
link_url = '/links/link[{:s}]'.format(link_uuid)
link_endpoint_ids = [
diff --git a/src/device/service/drivers/ietf_l3vpn/driver.py b/src/device/service/drivers/ietf_l3vpn/IetfL3VpnDriver.py
similarity index 67%
rename from src/device/service/drivers/ietf_l3vpn/driver.py
rename to src/device/service/drivers/ietf_l3vpn/IetfL3VpnDriver.py
index 2aca83b6a645bf2e793b08841949813f0413a531..7a7e336489affbe0f522328f22ec7c8e6461cb16 100644
--- a/src/device/service/drivers/ietf_l3vpn/driver.py
+++ b/src/device/service/drivers/ietf_l3vpn/IetfL3VpnDriver.py
@@ -1,4 +1,4 @@
-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -12,41 +12,20 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import json
-import logging
-import re
-import threading
-from typing import Any, Iterator, List, Optional, Tuple, Union
-
-import anytree
-import requests
-from requests.auth import HTTPBasicAuth
+import anytree, json, logging, re, requests, threading
+from typing import Any, Iterator, List, Optional, Tuple, Union
from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
from common.type_checkers.Checkers import chk_length, chk_string, chk_type
-from device.service.driver_api._Driver import (
- RESOURCE_ENDPOINTS,
- RESOURCE_SERVICES,
- _Driver,
-)
-from device.service.driver_api.AnyTreeTools import (
- TreeNode,
- dump_subtree,
- get_subnode,
- set_subnode_value,
-)
-from device.service.driver_api.ImportTopologyEnum import (
- ImportTopologyEnum,
- get_import_topology,
-)
-
+from device.service.driver_api._Driver import _Driver, RESOURCE_ENDPOINTS, RESOURCE_SERVICES
+from device.service.driver_api.AnyTreeTools import TreeNode, dump_subtree, get_subnode, set_subnode_value
+from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum, get_import_topology
from .Constants import SPECIAL_RESOURCE_MAPPINGS
from .TfsApiClient import TfsApiClient
from .Tools import compose_resource_endpoint
LOGGER = logging.getLogger(__name__)
-
ALL_RESOURCE_KEYS = [
RESOURCE_ENDPOINTS,
RESOURCE_SERVICES,
@@ -57,40 +36,34 @@ RE_GET_ENDPOINT_FROM_INTERFACE = re.compile(r"^\/interface\[([^\]]+)\].*")
RE_IETF_L3VPN_DATA = re.compile(r"^\/service\[[^\]]+\]\/IETFL3VPN$")
RE_IETF_L3VPN_OPERATION = re.compile(r"^\/service\[[^\]]+\]\/IETFL3VPN\/operation$")
-DRIVER_NAME = "ietf_l3vpn"
-METRICS_POOL = MetricsPool("Device", "Driver", labels={"driver": DRIVER_NAME})
-
+DRIVER_NAME = 'ietf_l3vpn'
+METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})
class IetfL3VpnDriver(_Driver):
- def __init__(self, address: str, port: str, **settings) -> None:
+ def __init__(self, address : str, port : str, **settings) -> None:
super().__init__(DRIVER_NAME, address, int(port), **settings)
self.__lock = threading.Lock()
self.__started = threading.Event()
self.__terminate = threading.Event()
- self.__running = TreeNode(".")
- scheme = self.settings.get("scheme", "http")
- username = self.settings.get("username")
- password = self.settings.get("password")
+ self.__running = TreeNode('.')
+ username = self.settings.get('username')
+ password = self.settings.get('password')
+ scheme = self.settings.get('scheme', 'http')
+ timeout = int(self.settings.get('timeout', 60))
self.tac = TfsApiClient(
- self.address,
- self.port,
- scheme=scheme,
- username=username,
- password=password,
- )
- self.__auth = None
- # (
- # HTTPBasicAuth(username, password)
- # if username is not None and password is not None
- # else None
- # )
- self.__tfs_nbi_root = "{:s}://{:s}:{:d}".format(
- scheme, self.address, int(self.port)
- )
- self.__timeout = int(self.settings.get("timeout", 120))
- self.__import_topology = get_import_topology(
- self.settings, default=ImportTopologyEnum.DEVICES
+ self.address, self.port, scheme=scheme, username=username,
+ password=password, timeout=timeout
)
+ #self.__tfs_nbi_root = "{:s}://{:s}:{:d}".format(scheme, self.address, int(self.port))
+
+ # Options are:
+ # disabled --> just import endpoints as usual
+ # devices --> imports sub-devices but not links connecting them.
+ # (a remotely-controlled transport domain might exist between them)
+ # topology --> imports sub-devices and links connecting them.
+ # (not supported by XR driver)
+ self.__import_topology = get_import_topology(self.settings, default=ImportTopologyEnum.DEVICES)
+
endpoints = self.settings.get("endpoints", [])
endpoint_resources = []
for endpoint in endpoints:
@@ -139,20 +112,12 @@ class IetfL3VpnDriver(_Driver):
return results
def Connect(self) -> bool:
- url = (
- self.__tfs_nbi_root + "/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services"
- )
with self.__lock:
- if self.__started.is_set():
- return True
+ if self.__started.is_set(): return True
try:
- # requests.get(url, timeout=self.__timeout, auth=self.__auth)
- ...
- except requests.exceptions.Timeout:
- LOGGER.exception("Timeout connecting {:s}".format(url))
- return False
- except Exception: # pylint: disable=broad-except
- LOGGER.exception("Exception connecting {:s}".format(url))
+ self.tac.check_credentials()
+ except: # pylint: disable=bare-except
+ LOGGER.exception('Exception checking credentials')
return False
else:
self.__started.set()
@@ -170,50 +135,46 @@ class IetfL3VpnDriver(_Driver):
@metered_subclass_method(METRICS_POOL)
def GetConfig(
- self, resource_keys: List[str] = []
+ self, resource_keys : List[str] = []
) -> List[Tuple[str, Union[Any, None, Exception]]]:
- chk_type("resources", resource_keys, list)
+ chk_type('resources', resource_keys, list)
+ results = []
with self.__lock:
- if len(resource_keys) == 0:
- return dump_subtree(self.__running)
- results = []
- resolver = anytree.Resolver(pathattr="name")
+ self.tac.check_credentials()
+ if len(resource_keys) == 0: resource_keys = ALL_RESOURCE_KEYS
+ #if len(resource_keys) == 0:
+ # return dump_subtree(self.__running)
+ resolver = anytree.Resolver(pathattr='name')
for i, resource_key in enumerate(resource_keys):
- str_resource_name = "resource_key[#{:d}]".format(i)
+ str_resource_name = 'resource_key[#{:d}]'.format(i)
try:
chk_string(str_resource_name, resource_key, allow_empty=False)
- resource_key = SPECIAL_RESOURCE_MAPPINGS.get(
- resource_key, resource_key
- )
- resource_path = resource_key.split("/")
- except Exception as e: # pylint: disable=broad-except
- LOGGER.exception(
- "Exception validating {:s}: {:s}".format(
- str_resource_name, str(resource_key)
+ if resource_key == RESOURCE_ENDPOINTS:
+ # return endpoints through TFS NBI API and list-devices method
+ results.extend(self.tac.get_devices_endpoints(self.__import_topology))
+ else:
+ resource_key = SPECIAL_RESOURCE_MAPPINGS.get(
+ resource_key, resource_key
)
- )
- results.append(
- (resource_key, e)
- ) # if validation fails, store the exception
- continue
-
- resource_node = get_subnode(
- resolver, self.__running, resource_path, default=None
- )
- # if not found, resource_node is None
- if resource_node is None:
- continue
- results.extend(dump_subtree(resource_node))
- return results
+ resource_path = resource_key.split('/')
+ resource_node = get_subnode(
+ resolver, self.__running, resource_path, default=None
+ )
+ # if not found, resource_node is None
+ if resource_node is None: continue
+ results.extend(dump_subtree(resource_node))
+ except Exception as e:
+ MSG = 'Unhandled error processing {:s}: resource_key({:s})'
+ LOGGER.exception(MSG.format(str_resource_name, str(resource_key)))
+ results.append((resource_key, e))
return results
@metered_subclass_method(METRICS_POOL)
def SetConfig(
- self, resources: List[Tuple[str, Any]]
+ self, resources : List[Tuple[str, Any]]
) -> List[Union[bool, Exception]]:
results = []
- if len(resources) == 0:
- return results
+ if len(resources) == 0: return results
with self.__lock:
for resource in resources:
resource_key, resource_value = resource
@@ -224,7 +185,7 @@ class IetfL3VpnDriver(_Driver):
else:
raise Exception("operation type not found in resources")
for resource in resources:
- LOGGER.info("resource = {:s}".format(str(resource)))
+ LOGGER.info('resource = {:s}'.format(str(resource)))
resource_key, resource_value = resource
if not RE_IETF_L3VPN_DATA.match(resource_key):
continue
@@ -261,7 +222,7 @@ class IetfL3VpnDriver(_Driver):
@metered_subclass_method(METRICS_POOL)
def DeleteConfig(
- self, resources: List[Tuple[str, Any]]
+ self, resources : List[Tuple[str, Any]]
) -> List[Union[bool, Exception]]:
results = []
if len(resources) == 0:
@@ -290,20 +251,20 @@ class IetfL3VpnDriver(_Driver):
@metered_subclass_method(METRICS_POOL)
def SubscribeState(
- self, subscriptions: List[Tuple[str, float, float]]
+ self, subscriptions : List[Tuple[str, float, float]]
) -> List[Union[bool, Exception]]:
- # TODO: IETF L3VPN does not support monitoring by now
+ # TODO: does not support monitoring by now
return [False for _ in subscriptions]
@metered_subclass_method(METRICS_POOL)
def UnsubscribeState(
- self, subscriptions: List[Tuple[str, float, float]]
+ self, subscriptions : List[Tuple[str, float, float]]
) -> List[Union[bool, Exception]]:
- # TODO: IETF L3VPN does not support monitoring by now
+ # TODO: does not support monitoring by now
return [False for _ in subscriptions]
def GetState(
- self, blocking=False, terminate: Optional[threading.Event] = None
+ self, blocking=False, terminate : Optional[threading.Event] = None
) -> Iterator[Tuple[float, str, Any]]:
- # TODO: IETF L3VPN does not support monitoring by now
+ # TODO: does not support monitoring by now
return []
diff --git a/src/device/service/drivers/ietf_l3vpn/TfsApiClient.py b/src/device/service/drivers/ietf_l3vpn/TfsApiClient.py
index 2db898059f6ac22785a00a80179334a22f929767..6efe3712f327af35a114434e09efa5d4996ee848 100644
--- a/src/device/service/drivers/ietf_l3vpn/TfsApiClient.py
+++ b/src/device/service/drivers/ietf_l3vpn/TfsApiClient.py
@@ -13,22 +13,14 @@
# limitations under the License.
import logging, requests
-from requests.auth import HTTPBasicAuth
from typing import Dict, List, Optional
+from common.tools.client.RestClient import RestClient
from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum
-GET_DEVICES_URL = '{:s}://{:s}:{:d}/tfs-api/devices'
-GET_LINKS_URL = '{:s}://{:s}:{:d}/tfs-api/links'
-L3VPN_URL = '{:s}://{:s}:{:d}/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services'
-
-TIMEOUT = 30
-
-HTTP_OK_CODES = {
- 200, # OK
- 201, # Created
- 202, # Accepted
- 204, # No Content
-}
+GET_CONTEXT_IDS_URL = '/tfs-api/context_ids'
+GET_DEVICES_URL = '/tfs-api/devices'
+GET_LINKS_URL = '/tfs-api/links'
+L3VPN_URL = '/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services'
MAPPING_STATUS = {
'DEVICEOPERATIONALSTATUS_UNDEFINED': 0,
@@ -55,24 +47,23 @@ MAPPING_DRIVER = {
'DEVICEDRIVER_NCE' : 15,
}
-MSG_ERROR = 'Could not retrieve devices in remote TeraFlowSDN instance({:s}). status_code={:s} reply={:s}'
-
LOGGER = logging.getLogger(__name__)
-class TfsApiClient:
+class TfsApiClient(RestClient):
def __init__(
self, address : str, port : int, scheme : str = 'http',
- username : Optional[str] = None, password : Optional[str] = None
+ username : Optional[str] = None, password : Optional[str] = None,
+ timeout : Optional[int] = 30
) -> None:
- self._devices_url = GET_DEVICES_URL.format(scheme, address, port)
- self._links_url = GET_LINKS_URL.format(scheme, address, port)
- self._l3vpn_url = L3VPN_URL.format(scheme, address, port)
- self._auth = (
- HTTPBasicAuth(username, password)
- if username is not None and password is not None
- else None
+ super().__init__(
+ address, port, scheme=scheme, username=username, password=password,
+ timeout=timeout, verify_certs=False, allow_redirects=True, logger=LOGGER
)
+ def check_credentials(self) -> None:
+ self.get(GET_CONTEXT_IDS_URL, expected_status_codes={requests.codes['OK']})
+ LOGGER.info('Credentials checked')
+
def get_devices_endpoints(
self, import_topology : ImportTopologyEnum = ImportTopologyEnum.DEVICES
) -> List[Dict]:
@@ -80,20 +71,14 @@ class TfsApiClient:
MSG = '[get_devices_endpoints] import_topology={:s}'
LOGGER.debug(MSG.format(str(import_topology)))
- reply = requests.get(self._devices_url, timeout=TIMEOUT, verify=False, auth=self._auth)
- if reply.status_code not in HTTP_OK_CODES:
- msg = MSG_ERROR.format(
- str(self._devices_url), str(reply.status_code), str(reply)
- )
- LOGGER.error(msg)
- raise Exception(msg)
-
if import_topology == ImportTopologyEnum.DISABLED:
MSG = 'Unsupported import_topology mode: {:s}'
raise Exception(MSG.format(str(import_topology)))
+ devices = self.get(GET_DEVICES_URL, expected_status_codes={requests.codes['OK']})
+
result = list()
- for json_device in reply.json()['devices']:
+ for json_device in devices['devices']:
device_uuid : str = json_device['device_id']['device_uuid']['uuid']
device_type : str = json_device['device_type']
#if not device_type.startswith('emu-'): device_type = 'emu-' + device_type
@@ -126,15 +111,9 @@ class TfsApiClient:
LOGGER.debug('[get_devices_endpoints] devices only; returning')
return result
- reply = requests.get(self._links_url, timeout=TIMEOUT, verify=False, auth=self._auth)
- if reply.status_code not in HTTP_OK_CODES:
- msg = MSG_ERROR.format(
- str(self._links_url), str(reply.status_code), str(reply)
- )
- LOGGER.error(msg)
- raise Exception(msg)
+ links = self.get(GET_LINKS_URL, expected_status_codes={requests.codes['OK']})
- for json_link in reply.json()['links']:
+ for json_link in links['links']:
link_uuid : str = json_link['link_id']['link_uuid']['uuid']
link_url = '/links/link[{:s}]'.format(link_uuid)
link_endpoint_ids = [
@@ -154,29 +133,31 @@ class TfsApiClient:
LOGGER.debug('[get_devices_endpoints] topology; returning')
return result
- def create_connectivity_service(self, l3vpn_data: dict) -> None:
+ def create_connectivity_service(self, l3vpn_data : dict) -> None:
+ MSG = '[create_connectivity_service] l3vpn_data={:s}'
+ LOGGER.debug(MSG.format(str(l3vpn_data)))
try:
- requests.post(self._l3vpn_url, json=l3vpn_data)
- MSG = '[create_connectivity_service] l3vpn_data={:s}'
- LOGGER.debug(MSG.format(str(l3vpn_data)))
- except requests.exceptions.ConnectionError:
- raise Exception('Failed to send POST request to TFS L3VPN NBI')
-
- def update_connectivity_service(self, l3vpn_data: dict) -> None:
+ self.post(L3VPN_URL, body=l3vpn_data)
+ except requests.exceptions.ConnectionError as e:
+ MSG = 'Failed to send POST request to TFS L3VPN NBI'
+ raise Exception(MSG) from e
+
+ def update_connectivity_service(self, l3vpn_data : dict) -> None:
+ MSG = '[update_connectivity_service] l3vpn_data={:s}'
+ LOGGER.debug(MSG.format(str(l3vpn_data)))
vpn_id = l3vpn_data['ietf-l3vpn-svc:l3vpn-svc']['vpn-services']['vpn-service'][0]['vpn-id']
- url = self._l3vpn_url + f'/vpn-service={vpn_id}'
try:
- requests.put(url, json=l3vpn_data)
- MSG = '[update_connectivity_service] l3vpn_data={:s}'
- LOGGER.debug(MSG.format(str(l3vpn_data)))
- except requests.exceptions.ConnectionError:
- raise Exception('Failed to send PUT request to TFS L3VPN NBI')
-
- def delete_connectivity_service(self, service_uuid: str) -> None:
- url = self._l3vpn_url + f'/vpn-service={service_uuid}'
+ self.put(L3VPN_URL + f'/vpn-service={vpn_id}', body=l3vpn_data)
+ except requests.exceptions.ConnectionError as e:
+ MSG = 'Failed to send PUT request to TFS L3VPN NBI'
+ raise Exception(MSG) from e
+
+ def delete_connectivity_service(self, service_uuid : str) -> None:
+ url = L3VPN_URL + f'/vpn-service={service_uuid}'
+ MSG = '[delete_connectivity_service] url={:s}'
+ LOGGER.debug(MSG.format(str(url)))
try:
- requests.delete(url)
- MSG = '[delete_connectivity_service] url={:s}'
- LOGGER.debug(MSG.format(str(url)))
- except requests.exceptions.ConnectionError:
- raise Exception('Failed to send DELETE request to TFS L3VPN NBI')
+ self.delete(url)
+ except requests.exceptions.ConnectionError as e:
+ MSG = 'Failed to send DELETE request to TFS L3VPN NBI'
+ raise Exception(MSG) from e
diff --git a/src/device/service/drivers/optical_tfs/OpticalTfsDriver.py b/src/device/service/drivers/optical_tfs/OpticalTfsDriver.py
index 8af8004542f8eca14a36a77ac610457bcd1ceb7a..8d7d95251b3a455ceb23440daf03a9c51dc5e8b3 100644
--- a/src/device/service/drivers/optical_tfs/OpticalTfsDriver.py
+++ b/src/device/service/drivers/optical_tfs/OpticalTfsDriver.py
@@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+
import json, logging, threading
from typing import Any, Iterator, List, Optional, Tuple, Union
from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
@@ -32,8 +33,8 @@ DRIVER_NAME = 'optical_tfs'
METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})
class OpticalTfsDriver(_Driver):
- def __init__(self, address: str, port: int, **settings) -> None:
- super().__init__(DRIVER_NAME, address, port, **settings)
+ def __init__(self, address : str, port : str, **settings) -> None:
+ super().__init__(DRIVER_NAME, address, int(port), **settings)
self.__lock = threading.Lock()
self.__started = threading.Event()
self.__terminate = threading.Event()
@@ -42,7 +43,7 @@ class OpticalTfsDriver(_Driver):
scheme = self.settings.get('scheme', 'http')
timeout = int(self.settings.get('timeout', 60))
self.tac = TfsApiClient(
- self.address, int(self.port), scheme=scheme, username=username,
+ self.address, self.port, scheme=scheme, username=username,
password=password, timeout=timeout
)
self.toc = TfsOpticalClient(
@@ -62,7 +63,7 @@ class OpticalTfsDriver(_Driver):
with self.__lock:
if self.__started.is_set(): return True
try:
- self.toc.check_credentials()
+ self.tac.check_credentials()
except: # pylint: disable=bare-except
LOGGER.exception('Exception checking credentials')
return False
@@ -81,11 +82,13 @@ class OpticalTfsDriver(_Driver):
return []
@metered_subclass_method(METRICS_POOL)
- def GetConfig(self, resource_keys : List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
+ def GetConfig(
+ self, resource_keys : List[str] = []
+ ) -> List[Tuple[str, Union[Any, None, Exception]]]:
chk_type('resources', resource_keys, list)
results = []
with self.__lock:
- self.toc.check_credentials()
+ self.tac.check_credentials()
if len(resource_keys) == 0: resource_keys = ALL_RESOURCE_KEYS
for i, resource_key in enumerate(resource_keys):
str_resource_name = 'resource_key[#{:d}]'.format(i)
@@ -96,24 +99,28 @@ class OpticalTfsDriver(_Driver):
results.extend(self.tac.get_devices_endpoints(self.__import_topology))
elif resource_key == RESOURCE_SERVICES:
# return all services through
- results.extend(self.toc.get_lightpaths())
+ #results.extend(self.toc.get_lightpaths())
+ pass
else:
MSG = 'ResourceKey({:s}) not implemented'
LOGGER.warning(MSG.format(str(resource_key)))
except Exception as e:
- LOGGER.exception('Unhandled error processing resource_key({:s})'.format(str(resource_key)))
+ MSG = 'Unhandled error processing {:s}: resource_key({:s})'
+ LOGGER.exception(MSG.format(str_resource_name, str(resource_key)))
results.append((resource_key, e))
return results
@metered_subclass_method(METRICS_POOL)
- def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+ def SetConfig(
+ self, resources : List[Tuple[str, Any]]
+ ) -> List[Union[bool, Exception]]:
results = []
if len(resources) == 0: return results
with self.__lock:
- self.toc.check_credentials()
+ self.tac.check_credentials()
for resource in resources:
LOGGER.info('resource = {:s}'.format(str(resource)))
- resource_key,resource_value = resource
+ resource_key, resource_value = resource
try:
resource_value = json.loads(resource_value)
src_node = resource_value['src_node']
@@ -127,11 +134,13 @@ class OpticalTfsDriver(_Driver):
return results
@metered_subclass_method(METRICS_POOL)
- def DeleteConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+ def DeleteConfig(
+ self, resources : List[Tuple[str, Any]]
+ ) -> List[Union[bool, Exception]]:
results = []
if len(resources) == 0: return results
with self.__lock:
- self.toc.check_credentials()
+ self.tac.check_credentials()
for resource in resources:
LOGGER.info('resource = {:s}'.format(str(resource)))
resource_key,resource_value = resource
@@ -149,17 +158,21 @@ class OpticalTfsDriver(_Driver):
return results
@metered_subclass_method(METRICS_POOL)
- def SubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
- # TODO: Optical TFS does not support monitoring by now
+ def SubscribeState(
+ self, subscriptions : List[Tuple[str, float, float]]
+ ) -> List[Union[bool, Exception]]:
+ # TODO: does not support monitoring by now
return [False for _ in subscriptions]
@metered_subclass_method(METRICS_POOL)
- def UnsubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
- # TODO: Optical TFS does not support monitoring by now
+ def UnsubscribeState(
+ self, subscriptions : List[Tuple[str, float, float]]
+ ) -> List[Union[bool, Exception]]:
+ # TODO: does not support monitoring by now
return [False for _ in subscriptions]
def GetState(
self, blocking=False, terminate : Optional[threading.Event] = None
) -> Iterator[Tuple[float, str, Any]]:
- # TODO: Optical TFS does not support monitoring by now
+ # TODO: does not support monitoring by now
return []
diff --git a/src/device/service/drivers/optical_tfs/TfsApiClient.py b/src/device/service/drivers/optical_tfs/TfsApiClient.py
index 8df8e5261b707c0a5545fd7ace62b5a707df60f8..e9b21f2498ba37abd8d45f278724c7afc82b4573 100644
--- a/src/device/service/drivers/optical_tfs/TfsApiClient.py
+++ b/src/device/service/drivers/optical_tfs/TfsApiClient.py
@@ -17,8 +17,9 @@ from typing import Dict, List, Optional
from common.tools.client.RestClient import RestClient
from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum
-GET_DEVICES_URL = '/tfs-api/devices'
-GET_LINKS_URL = '/tfs-api/links'
+GET_CONTEXT_IDS_URL = '/tfs-api/context_ids'
+GET_DEVICES_URL = '/tfs-api/devices'
+GET_LINKS_URL = '/tfs-api/links'
MAPPING_STATUS = {
'DEVICEOPERATIONALSTATUS_UNDEFINED': 0,
@@ -58,6 +59,10 @@ class TfsApiClient(RestClient):
timeout=timeout, verify_certs=False, allow_redirects=True, logger=LOGGER
)
+ def check_credentials(self) -> None:
+ self.get(GET_CONTEXT_IDS_URL, expected_status_codes={requests.codes['OK']})
+ LOGGER.info('Credentials checked')
+
def get_devices_endpoints(
self, import_topology : ImportTopologyEnum = ImportTopologyEnum.DEVICES
) -> List[Dict]:
diff --git a/src/device/tests/test_unitary_ietf_l3vpn.py b/src/device/tests/test_unitary_ietf_l3vpn.py
index 728ca691332c8abee7b5d6f5ad6c151240e540ed..f9f7ae99070f6a55137be0924eb4e69863b86fbe 100644
--- a/src/device/tests/test_unitary_ietf_l3vpn.py
+++ b/src/device/tests/test_unitary_ietf_l3vpn.py
@@ -3,7 +3,7 @@ from json import dumps
import requests
-from device.service.drivers.ietf_l3vpn.driver import IetfL3VpnDriver
+from device.service.drivers.ietf_l3vpn.IetfL3VpnDriver import IetfL3VpnDriver
from device.service.Tools import RESOURCE_ENDPOINTS
settings = {