Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging, requests, threading
from typing import Any, Iterator, List, Optional, Tuple, Union
from common.type_checkers.Checkers import chk_string, chk_type
from device.service.driver_api._Driver import _Driver
from . import ALL_RESOURCE_KEYS
from .Tools import create_connectivity_service, find_key, config_getter, delete_connectivity_service
LOGGER = logging.getLogger(__name__)
class IETFApiDriver(_Driver):
def __init__(self, address: str, port: int, **settings) -> None: # pylint: disable=super-init-not-called
self.__lock = threading.Lock()
self.__started = threading.Event()
self.__terminate = threading.Event()
self.__tapi_root = 'https://' + address + ':' + str(port)
self.__timeout = int(settings.get('timeout', 120))
def Connect(self) -> bool:
url = self.__tapi_root + '/nmswebs/restconf/data/ietf-network:networks'
with self.__lock:
if self.__started.is_set(): return True
try:
requests.get(url, timeout=self.__timeout, verify=False)
except requests.exceptions.Timeout:
LOGGER.exception('Timeout connecting {:s}'.format(str(self.__tapi_root)))
return False
except Exception: # pylint: disable=broad-except
LOGGER.exception('Exception connecting {:s}'.format(str(self.__tapi_root)))
return False
else:
self.__started.set()
return True
def Disconnect(self) -> bool:
with self.__lock:
self.__terminate.set()
return True
def GetInitialConfig(self) -> List[Tuple[str, Any]]:
with self.__lock:
return []
def GetConfig(self, resource_keys : List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
chk_type('resources', resource_keys, list)
results = []
with self.__lock:
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)
chk_string(str_resource_name, resource_key, allow_empty=False)
results.extend(config_getter(self.__tapi_root, resource_key, self.__timeout))
return results
def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
results = []
if len(resources) == 0:
return results
with self.__lock:
for resource in resources:
LOGGER.info('resource = {:s}'.format(str(resource)))
node_id_src = find_key(resource, 'node_id_src')
tp_id_src = find_key(resource, 'tp_id_src')
node_id_dst = find_key(resource, 'node_id_dst')
tp_id_dst = find_key(resource, 'tp_id_dst')
vlan_id = find_key(resource, 'vlan_id')
uuid = find_key(resource, 'uuid')
data = create_connectivity_service(
self.__tapi_root, self.__timeout, uuid, node_id_src, tp_id_src, node_id_dst, tp_id_dst, vlan_id)
results.extend(data)
return results
def DeleteConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
results = []
if len(resources) == 0: return results
with self.__lock:
for resource in resources:
LOGGER.info('resource = {:s}'.format(str(resource)))
uuid = find_key(resource, 'uuid')
results.extend(delete_connectivity_service(self.__tapi_root, self.__timeout, uuid))
return results
def SubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
# TODO: TAPI does not support monitoring by now
return [False for _ in subscriptions]
def UnsubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
# TODO: TAPI 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: TAPI does not support monitoring by now
return []