Loading src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/app.py +42 −0 Original line number Diff line number Diff line Loading @@ -20,6 +20,8 @@ from .Dispatch import RestConfDispatch from .HostMeta import HostMeta from .YangHandler import YangHandler from .YangModelDiscoverer import YangModuleDiscoverer from .simap_client.RestConfClient import RestConfClient from .simap_client.SimapClient import SimapClient logging.basicConfig( Loading @@ -27,6 +29,7 @@ logging.basicConfig( format="[Worker-%(process)d][%(asctime)s] %(levelname)s:%(name)s:%(message)s", ) LOGGER = logging.getLogger(__name__) logging.getLogger('RestConfClient').setLevel(logging.WARN) RESTCONF_PREFIX = '/restconf' Loading @@ -43,6 +46,45 @@ with open(STARTUP_FILE, mode='r', encoding='UTF-8') as fp: YANG_STARTUP_DATA = json.loads(fp.read()) restconf_client = RestConfClient( 'simap-client', port=8080, logger=logging.getLogger('RestConfClient') ) simap_client = SimapClient(restconf_client) te_topo = simap_client.network('admin') te_topo.update() networks = YANG_STARTUP_DATA.get('ietf-network:networks', dict()) networks = networks.get('network', list()) assert len(networks) == 1 network = networks[0] assert network['network-id'] == 'admin' nodes = network.get('node', list()) for node in nodes: node_id = node['node-id'] tp_ids = [ tp['tp-id'] for tp in node['ietf-network-topology:termination-point'] ] te_topo.node(node_id).create(termination_point_ids=tp_ids) links = network.get('ietf-network-topology:link', list()) for link in links: link_id = link['link-id'] link_src = link['source'] link_dst = link['destination'] link_src_node_id = link_src['source-node'] link_src_tp_id = link_src['source-tp'] link_dst_node_id = link_dst['dest-node'] link_dst_tp_id = link_dst['dest-tp'] te_topo.link(link_id).create( link_src_node_id, link_src_tp_id, link_dst_node_id, link_dst_tp_id ) yang_handler = YangHandler( YANG_SEARCH_PATH, YANG_MODULE_NAMES, YANG_STARTUP_DATA ) Loading src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/simap_client/RestConfClient.py 0 → 100644 +191 −0 Original line number Diff line number Diff line # Copyright 2022-2025 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. # 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 enum, logging, requests from requests.auth import HTTPBasicAuth from typing import Any, Dict, Optional, Set class RestRequestMethod(enum.Enum): GET = 'get' POST = 'post' PUT = 'put' PATCH = 'patch' DELETE = 'delete' EXPECTED_STATUS_CODES : Set[int] = { requests.codes['OK' ], # 200 - OK requests.codes['CREATED' ], # 201 - Created requests.codes['ACCEPTED' ], # 202 - Accepted requests.codes['NO_CONTENT'], # 204 - No Content } def compose_basic_auth( username : Optional[str] = None, password : Optional[str] = None ) -> Optional[HTTPBasicAuth]: if username is None or password is None: return None return HTTPBasicAuth(username, password) class SchemeEnum(enum.Enum): HTTP = 'http' HTTPS = 'https' def check_scheme(scheme : str) -> str: str_scheme = str(scheme).lower() enm_scheme = SchemeEnum._value2member_map_[str_scheme] return enm_scheme.value HOST_META_URL = '{:s}://{:s}:{:d}/.well-known/host-meta' RESTCONF_URL = '{:s}://{:s}:{:d}/{:s}' class RestConfClient: def __init__( self, address : str, port : int = 8080, scheme : str = 'http', username : Optional[str] = None, password : Optional[str] = None, timeout : int = 10, verify_certs : bool = True, allow_redirects : bool = True, logger : Optional[logging.Logger] = None ) -> None: self._address = address self._port = int(port) self._scheme = check_scheme(scheme) self._auth = compose_basic_auth(username=username, password=password) self._base_url = '' self._timeout = int(timeout) self._verify_certs = verify_certs self._allow_redirects = allow_redirects self._logger = logger self._discover_base_url() def _discover_base_url(self) -> None: host_meta_url = HOST_META_URL.format(self._scheme, self._address, self._port) host_meta : Dict = self.get(host_meta_url, expected_status_codes={requests.codes['OK']}) links = host_meta.get('links') if links is None: raise AttributeError('Missing attribute "links" in host-meta reply') if not isinstance(links, list): raise AttributeError('Attribute "links" must be a list') if len(links) != 1: raise AttributeError('Attribute "links" is expected to have exactly 1 item') link = links[0] if not isinstance(link, dict): raise AttributeError('Attribute "links[0]" must be a dict') rel = link.get('rel') if rel is None: raise AttributeError('Missing attribute "links[0].rel" in host-meta reply') if not isinstance(rel, str): raise AttributeError('Attribute "links[0].rel" must be a str') if rel != 'restconf': raise AttributeError('Attribute "links[0].rel" != "restconf"') href = link.get('href') if href is None: raise AttributeError('Missing attribute "links[0]" in host-meta reply') if not isinstance(href, str): raise AttributeError('Attribute "links[0].href" must be a str') self._base_url = str(href + '/data').replace('//', '/') def _log_msg_request( self, method : RestRequestMethod, request_url : str, body : Optional[Any], log_level : int = logging.INFO ) -> str: msg = 'Request: {:s} {:s}'.format(str(method.value).upper(), str(request_url)) if body is not None: msg += ' body={:s}'.format(str(body)) if self._logger is not None: self._logger.log(log_level, msg) return msg def _log_msg_check_reply( self, method : RestRequestMethod, request_url : str, body : Optional[Any], reply : requests.Response, expected_status_codes : Set[int], log_level : int = logging.INFO ) -> str: msg = 'Reply: {:s}'.format(str(reply.text)) if self._logger is not None: self._logger.log(log_level, msg) http_status_code = reply.status_code if http_status_code in expected_status_codes: return msg MSG = 'Request failed. method={:s} url={:s} body={:s} status_code={:s} reply={:s}' msg = MSG.format( str(method.value).upper(), str(request_url), str(body), str(http_status_code), str(reply.text) ) self._logger.error(msg) raise Exception(msg) def _do_rest_request( self, method : RestRequestMethod, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = EXPECTED_STATUS_CODES ) -> Optional[Any]: candidate_schemes = tuple(['{:s}://'.format(m).lower() for m in SchemeEnum.__members__.keys()]) if endpoint.lower().startswith(candidate_schemes): request_url = endpoint.lstrip('/') else: endpoint = str(self._base_url + '/' + endpoint).replace('//', '/').lstrip('/') request_url = '{:s}://{:s}:{:d}/{:s}'.format( self._scheme, self._address, self._port, endpoint.lstrip('/') ) self._log_msg_request(method, request_url, body) try: headers = {'accept': 'application/json'} reply = requests.request( method.value, request_url, headers=headers, json=body, auth=self._auth, verify=self._verify_certs, timeout=self._timeout, allow_redirects=self._allow_redirects ) except Exception as e: MSG = 'Request failed. method={:s} url={:s} body={:s}' msg = MSG.format(str(method.value).upper(), request_url, str(body)) self._logger.exception(msg) raise Exception(msg) from e self._log_msg_check_reply(method, request_url, body, reply, expected_status_codes) if reply.content and len(reply.content) > 0: return reply.json() return None def get( self, endpoint : str, expected_status_codes : Set[int] = {requests.codes['OK']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.GET, endpoint, expected_status_codes=expected_status_codes ) def post( self, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = {requests.codes['CREATED']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.POST, endpoint, body=body, expected_status_codes=expected_status_codes ) def put( self, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = {requests.codes['CREATED'], requests.codes['NO_CONTENT']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.PUT, endpoint, body=body, expected_status_codes=expected_status_codes ) def patch( self, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = {requests.codes['NO_CONTENT']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.PATCH, endpoint, body=body, expected_status_codes=expected_status_codes ) def delete( self, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = {requests.codes['NO_CONTENT']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.DELETE, endpoint, body=body, expected_status_codes=expected_status_codes ) src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/simap_client/SimapClient.py 0 → 100644 +242 −0 Original line number Diff line number Diff line # Copyright 2022-2025 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. # 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. from typing import Dict, List, Tuple from .RestConfClient import RestConfClient class TerminationPoint: ENDPOINT_NO_ID = '/ietf-network:networks/network[network-id="{:s}"]/node[node-id="{:s}"]' ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:termination-point[tp-id="{:s}"]' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str, tp_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id self._tp_id = tp_id def create(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) tp = {'tp-id': self._tp_id} stps = [ {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id} for snet_id,snode_id,stp_id in supporting_termination_point_ids ] if len(stps) > 0: tp['supporting-termination-point'] = stps node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) node : Dict = self._restconf_client.get(endpoint) return node['ietf-network-topology:termination-point'][0] def update(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) tp = {'tp-id': self._tp_id} stps = [ {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id} for snet_id,snode_id,stp_id in supporting_termination_point_ids ] if len(stps) > 0: tp['supporting-termination-point'] = stps node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) self._restconf_client.delete(endpoint) class Node: ENDPOINT_NO_ID = '/ietf-network:networks/network[network-id="{:s}"]' ENDPOINT_ID = ENDPOINT_NO_ID + '/node[node-id="{:s}"]' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id self._tps : Dict[str, TerminationPoint] = dict() def termination_points(self) -> List[Dict]: tps : Dict = self._restconf_client.get(TerminationPoint.ENDPOINT_NO_ID) return tps['ietf-network-topology:termination-point'].get('termination-point', list()) def termination_point(self, tp_id : str) -> TerminationPoint: _tp = self._tps.get(tp_id) if _tp is not None: return _tp _tp = TerminationPoint(self._restconf_client, self._network_id, self._node_id, tp_id) return self._tps.setdefault(tp_id, _tp) def create( self, termination_point_ids : List[str] = [], supporting_node_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node = {'node-id': self._node_id} tps = [{'tp-id': tp_id} for tp_id in termination_point_ids] if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids] if len(sns) > 0: node['supporting-node'] = sns network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node : Dict = self._restconf_client.get(endpoint) return node['ietf-network:node'][0] def update( self, termination_point_ids : List[str] = [], supporting_node_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node = {'node-id': self._node_id} tps = [{'tp-id': tp_id} for tp_id in termination_point_ids] if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids] if len(sns) > 0: node['supporting-node'] = sns network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) self._restconf_client.delete(endpoint) class Link: ENDPOINT_NO_ID = '/ietf-network:networks/network[network-id="{:s}"]' ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:link[link-id="{:s}"]' def __init__(self, restconf_client : RestConfClient, network_id : str, link_id : str): self._restconf_client = restconf_client self._network_id = network_id self._link_id = link_id def create( self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str, supporting_link_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link = { 'link-id' : self._link_id, 'source' : {'source-node': src_node_id, 'source-tp': src_tp_id}, 'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id}, } sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids] if len(sls) > 0: link['supporting-link'] = sls network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link : Dict = self._restconf_client.get(endpoint) return link['ietf-network-topology:link'][0] def update( self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str, supporting_link_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link = { 'link-id' : self._link_id, 'source' : {'source-node': src_node_id, 'source-tp': src_tp_id}, 'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id}, } sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids] if len(sls) > 0: link['supporting-link'] = sls network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) self._restconf_client.delete(endpoint) class Network: ENDPOINT_NO_ID = '/ietf-network:networks' ENDPOINT_ID = ENDPOINT_NO_ID + '/network[network-id="{:s}"]' def __init__(self, restconf_client : RestConfClient, network_id : str): self._restconf_client = restconf_client self._network_id = network_id self._nodes : Dict[str, Node] = dict() self._links : Dict[str, Link] = dict() def nodes(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Node.ENDPOINT_NO_ID.format(self._network_id)) return reply['ietf-network:network'][0].get('node', list()) def links(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Link.ENDPOINT_NO_ID.format(self._network_id)) return reply['ietf-network:network'][0].get('ietf-network-topology:link', list()) def node(self, node_id : str) -> Node: _node = self._nodes.get(node_id) if _node is not None: return _node _node = Node(self._restconf_client, self._network_id, node_id) return self._nodes.setdefault(node_id, _node) def link(self, link_id : str) -> Link: _link = self._links.get(link_id) if _link is not None: return _link _link = Link(self._restconf_client, self._network_id, link_id) return self._links.setdefault(link_id, _link) def create(self, supporting_network_ids : List[str] = []) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) network = {'network-id': self._network_id} sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids] if len(sns) > 0: network['supporting-network'] = sns payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Network.ENDPOINT_ID.format(self._network_id) networks : Dict = self._restconf_client.get(endpoint) return networks['ietf-network:network'][0] def update(self, supporting_network_ids : List[str] = []) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) network = {'network-id': self._network_id} sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids] if len(sns) > 0: network['supporting-network'] = sns payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) self._restconf_client.delete(endpoint) class SimapClient: def __init__(self, restconf_client : RestConfClient) -> None: self._restconf_client = restconf_client self._networks : Dict[str, Network] = dict() def networks(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Network.ENDPOINT_NO_ID) return reply['ietf-network:networks'].get('network', list()) def network(self, network_id : str) -> Network: _network = self._networks.get(network_id) if _network is not None: return _network _network = Network(self._restconf_client, network_id) return self._networks.setdefault(network_id, _network) src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/simap_client/__init__.py 0 → 100644 +14 −0 Original line number Diff line number Diff line # Copyright 2022-2025 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. # 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. Loading
src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/app.py +42 −0 Original line number Diff line number Diff line Loading @@ -20,6 +20,8 @@ from .Dispatch import RestConfDispatch from .HostMeta import HostMeta from .YangHandler import YangHandler from .YangModelDiscoverer import YangModuleDiscoverer from .simap_client.RestConfClient import RestConfClient from .simap_client.SimapClient import SimapClient logging.basicConfig( Loading @@ -27,6 +29,7 @@ logging.basicConfig( format="[Worker-%(process)d][%(asctime)s] %(levelname)s:%(name)s:%(message)s", ) LOGGER = logging.getLogger(__name__) logging.getLogger('RestConfClient').setLevel(logging.WARN) RESTCONF_PREFIX = '/restconf' Loading @@ -43,6 +46,45 @@ with open(STARTUP_FILE, mode='r', encoding='UTF-8') as fp: YANG_STARTUP_DATA = json.loads(fp.read()) restconf_client = RestConfClient( 'simap-client', port=8080, logger=logging.getLogger('RestConfClient') ) simap_client = SimapClient(restconf_client) te_topo = simap_client.network('admin') te_topo.update() networks = YANG_STARTUP_DATA.get('ietf-network:networks', dict()) networks = networks.get('network', list()) assert len(networks) == 1 network = networks[0] assert network['network-id'] == 'admin' nodes = network.get('node', list()) for node in nodes: node_id = node['node-id'] tp_ids = [ tp['tp-id'] for tp in node['ietf-network-topology:termination-point'] ] te_topo.node(node_id).create(termination_point_ids=tp_ids) links = network.get('ietf-network-topology:link', list()) for link in links: link_id = link['link-id'] link_src = link['source'] link_dst = link['destination'] link_src_node_id = link_src['source-node'] link_src_tp_id = link_src['source-tp'] link_dst_node_id = link_dst['dest-node'] link_dst_tp_id = link_dst['dest-tp'] te_topo.link(link_id).create( link_src_node_id, link_src_tp_id, link_dst_node_id, link_dst_tp_id ) yang_handler = YangHandler( YANG_SEARCH_PATH, YANG_MODULE_NAMES, YANG_STARTUP_DATA ) Loading
src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/simap_client/RestConfClient.py 0 → 100644 +191 −0 Original line number Diff line number Diff line # Copyright 2022-2025 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. # 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 enum, logging, requests from requests.auth import HTTPBasicAuth from typing import Any, Dict, Optional, Set class RestRequestMethod(enum.Enum): GET = 'get' POST = 'post' PUT = 'put' PATCH = 'patch' DELETE = 'delete' EXPECTED_STATUS_CODES : Set[int] = { requests.codes['OK' ], # 200 - OK requests.codes['CREATED' ], # 201 - Created requests.codes['ACCEPTED' ], # 202 - Accepted requests.codes['NO_CONTENT'], # 204 - No Content } def compose_basic_auth( username : Optional[str] = None, password : Optional[str] = None ) -> Optional[HTTPBasicAuth]: if username is None or password is None: return None return HTTPBasicAuth(username, password) class SchemeEnum(enum.Enum): HTTP = 'http' HTTPS = 'https' def check_scheme(scheme : str) -> str: str_scheme = str(scheme).lower() enm_scheme = SchemeEnum._value2member_map_[str_scheme] return enm_scheme.value HOST_META_URL = '{:s}://{:s}:{:d}/.well-known/host-meta' RESTCONF_URL = '{:s}://{:s}:{:d}/{:s}' class RestConfClient: def __init__( self, address : str, port : int = 8080, scheme : str = 'http', username : Optional[str] = None, password : Optional[str] = None, timeout : int = 10, verify_certs : bool = True, allow_redirects : bool = True, logger : Optional[logging.Logger] = None ) -> None: self._address = address self._port = int(port) self._scheme = check_scheme(scheme) self._auth = compose_basic_auth(username=username, password=password) self._base_url = '' self._timeout = int(timeout) self._verify_certs = verify_certs self._allow_redirects = allow_redirects self._logger = logger self._discover_base_url() def _discover_base_url(self) -> None: host_meta_url = HOST_META_URL.format(self._scheme, self._address, self._port) host_meta : Dict = self.get(host_meta_url, expected_status_codes={requests.codes['OK']}) links = host_meta.get('links') if links is None: raise AttributeError('Missing attribute "links" in host-meta reply') if not isinstance(links, list): raise AttributeError('Attribute "links" must be a list') if len(links) != 1: raise AttributeError('Attribute "links" is expected to have exactly 1 item') link = links[0] if not isinstance(link, dict): raise AttributeError('Attribute "links[0]" must be a dict') rel = link.get('rel') if rel is None: raise AttributeError('Missing attribute "links[0].rel" in host-meta reply') if not isinstance(rel, str): raise AttributeError('Attribute "links[0].rel" must be a str') if rel != 'restconf': raise AttributeError('Attribute "links[0].rel" != "restconf"') href = link.get('href') if href is None: raise AttributeError('Missing attribute "links[0]" in host-meta reply') if not isinstance(href, str): raise AttributeError('Attribute "links[0].href" must be a str') self._base_url = str(href + '/data').replace('//', '/') def _log_msg_request( self, method : RestRequestMethod, request_url : str, body : Optional[Any], log_level : int = logging.INFO ) -> str: msg = 'Request: {:s} {:s}'.format(str(method.value).upper(), str(request_url)) if body is not None: msg += ' body={:s}'.format(str(body)) if self._logger is not None: self._logger.log(log_level, msg) return msg def _log_msg_check_reply( self, method : RestRequestMethod, request_url : str, body : Optional[Any], reply : requests.Response, expected_status_codes : Set[int], log_level : int = logging.INFO ) -> str: msg = 'Reply: {:s}'.format(str(reply.text)) if self._logger is not None: self._logger.log(log_level, msg) http_status_code = reply.status_code if http_status_code in expected_status_codes: return msg MSG = 'Request failed. method={:s} url={:s} body={:s} status_code={:s} reply={:s}' msg = MSG.format( str(method.value).upper(), str(request_url), str(body), str(http_status_code), str(reply.text) ) self._logger.error(msg) raise Exception(msg) def _do_rest_request( self, method : RestRequestMethod, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = EXPECTED_STATUS_CODES ) -> Optional[Any]: candidate_schemes = tuple(['{:s}://'.format(m).lower() for m in SchemeEnum.__members__.keys()]) if endpoint.lower().startswith(candidate_schemes): request_url = endpoint.lstrip('/') else: endpoint = str(self._base_url + '/' + endpoint).replace('//', '/').lstrip('/') request_url = '{:s}://{:s}:{:d}/{:s}'.format( self._scheme, self._address, self._port, endpoint.lstrip('/') ) self._log_msg_request(method, request_url, body) try: headers = {'accept': 'application/json'} reply = requests.request( method.value, request_url, headers=headers, json=body, auth=self._auth, verify=self._verify_certs, timeout=self._timeout, allow_redirects=self._allow_redirects ) except Exception as e: MSG = 'Request failed. method={:s} url={:s} body={:s}' msg = MSG.format(str(method.value).upper(), request_url, str(body)) self._logger.exception(msg) raise Exception(msg) from e self._log_msg_check_reply(method, request_url, body, reply, expected_status_codes) if reply.content and len(reply.content) > 0: return reply.json() return None def get( self, endpoint : str, expected_status_codes : Set[int] = {requests.codes['OK']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.GET, endpoint, expected_status_codes=expected_status_codes ) def post( self, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = {requests.codes['CREATED']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.POST, endpoint, body=body, expected_status_codes=expected_status_codes ) def put( self, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = {requests.codes['CREATED'], requests.codes['NO_CONTENT']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.PUT, endpoint, body=body, expected_status_codes=expected_status_codes ) def patch( self, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = {requests.codes['NO_CONTENT']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.PATCH, endpoint, body=body, expected_status_codes=expected_status_codes ) def delete( self, endpoint : str, body : Optional[Any] = None, expected_status_codes : Set[int] = {requests.codes['NO_CONTENT']} ) -> Optional[Any]: return self._do_rest_request( RestRequestMethod.DELETE, endpoint, body=body, expected_status_codes=expected_status_codes )
src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/simap_client/SimapClient.py 0 → 100644 +242 −0 Original line number Diff line number Diff line # Copyright 2022-2025 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. # 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. from typing import Dict, List, Tuple from .RestConfClient import RestConfClient class TerminationPoint: ENDPOINT_NO_ID = '/ietf-network:networks/network[network-id="{:s}"]/node[node-id="{:s}"]' ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:termination-point[tp-id="{:s}"]' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str, tp_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id self._tp_id = tp_id def create(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) tp = {'tp-id': self._tp_id} stps = [ {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id} for snet_id,snode_id,stp_id in supporting_termination_point_ids ] if len(stps) > 0: tp['supporting-termination-point'] = stps node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) node : Dict = self._restconf_client.get(endpoint) return node['ietf-network-topology:termination-point'][0] def update(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) tp = {'tp-id': self._tp_id} stps = [ {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id} for snet_id,snode_id,stp_id in supporting_termination_point_ids ] if len(stps) > 0: tp['supporting-termination-point'] = stps node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) self._restconf_client.delete(endpoint) class Node: ENDPOINT_NO_ID = '/ietf-network:networks/network[network-id="{:s}"]' ENDPOINT_ID = ENDPOINT_NO_ID + '/node[node-id="{:s}"]' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id self._tps : Dict[str, TerminationPoint] = dict() def termination_points(self) -> List[Dict]: tps : Dict = self._restconf_client.get(TerminationPoint.ENDPOINT_NO_ID) return tps['ietf-network-topology:termination-point'].get('termination-point', list()) def termination_point(self, tp_id : str) -> TerminationPoint: _tp = self._tps.get(tp_id) if _tp is not None: return _tp _tp = TerminationPoint(self._restconf_client, self._network_id, self._node_id, tp_id) return self._tps.setdefault(tp_id, _tp) def create( self, termination_point_ids : List[str] = [], supporting_node_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node = {'node-id': self._node_id} tps = [{'tp-id': tp_id} for tp_id in termination_point_ids] if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids] if len(sns) > 0: node['supporting-node'] = sns network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node : Dict = self._restconf_client.get(endpoint) return node['ietf-network:node'][0] def update( self, termination_point_ids : List[str] = [], supporting_node_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node = {'node-id': self._node_id} tps = [{'tp-id': tp_id} for tp_id in termination_point_ids] if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids] if len(sns) > 0: node['supporting-node'] = sns network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) self._restconf_client.delete(endpoint) class Link: ENDPOINT_NO_ID = '/ietf-network:networks/network[network-id="{:s}"]' ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:link[link-id="{:s}"]' def __init__(self, restconf_client : RestConfClient, network_id : str, link_id : str): self._restconf_client = restconf_client self._network_id = network_id self._link_id = link_id def create( self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str, supporting_link_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link = { 'link-id' : self._link_id, 'source' : {'source-node': src_node_id, 'source-tp': src_tp_id}, 'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id}, } sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids] if len(sls) > 0: link['supporting-link'] = sls network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link : Dict = self._restconf_client.get(endpoint) return link['ietf-network-topology:link'][0] def update( self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str, supporting_link_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link = { 'link-id' : self._link_id, 'source' : {'source-node': src_node_id, 'source-tp': src_tp_id}, 'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id}, } sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids] if len(sls) > 0: link['supporting-link'] = sls network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) self._restconf_client.delete(endpoint) class Network: ENDPOINT_NO_ID = '/ietf-network:networks' ENDPOINT_ID = ENDPOINT_NO_ID + '/network[network-id="{:s}"]' def __init__(self, restconf_client : RestConfClient, network_id : str): self._restconf_client = restconf_client self._network_id = network_id self._nodes : Dict[str, Node] = dict() self._links : Dict[str, Link] = dict() def nodes(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Node.ENDPOINT_NO_ID.format(self._network_id)) return reply['ietf-network:network'][0].get('node', list()) def links(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Link.ENDPOINT_NO_ID.format(self._network_id)) return reply['ietf-network:network'][0].get('ietf-network-topology:link', list()) def node(self, node_id : str) -> Node: _node = self._nodes.get(node_id) if _node is not None: return _node _node = Node(self._restconf_client, self._network_id, node_id) return self._nodes.setdefault(node_id, _node) def link(self, link_id : str) -> Link: _link = self._links.get(link_id) if _link is not None: return _link _link = Link(self._restconf_client, self._network_id, link_id) return self._links.setdefault(link_id, _link) def create(self, supporting_network_ids : List[str] = []) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) network = {'network-id': self._network_id} sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids] if len(sns) > 0: network['supporting-network'] = sns payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Network.ENDPOINT_ID.format(self._network_id) networks : Dict = self._restconf_client.get(endpoint) return networks['ietf-network:network'][0] def update(self, supporting_network_ids : List[str] = []) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) network = {'network-id': self._network_id} sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids] if len(sns) > 0: network['supporting-network'] = sns payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) self._restconf_client.delete(endpoint) class SimapClient: def __init__(self, restconf_client : RestConfClient) -> None: self._restconf_client = restconf_client self._networks : Dict[str, Network] = dict() def networks(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Network.ENDPOINT_NO_ID) return reply['ietf-network:networks'].get('network', list()) def network(self, network_id : str) -> Network: _network = self._networks.get(network_id) if _network is not None: return _network _network = Network(self._restconf_client, network_id) return self._networks.setdefault(network_id, _network)
src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/simap_client/__init__.py 0 → 100644 +14 −0 Original line number Diff line number Diff line # Copyright 2022-2025 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. # 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.
