Test - Tools - Mock NCE-T /NCE-FAN Ctrl (b4c1e700) · Commits · TFS / controller

src/tests/tools/mock_nce_fan_ctrl/nce_fan_ctrl/SimapClient.py

+109 −1

Original line number	Diff line number	Diff line
		@@ -13,7 +13,7 @@
		# limitations under the License.


		from typing import Dict, List, Tuple
		from typing import Dict, List, Optional, Tuple
		from common.tools.rest_conf.client.RestConfClient import RestConfClient


		@@ -62,6 +62,51 @@ class TerminationPoint:
		endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id)
		self._restconf_client.delete(endpoint)


		class NodeTelemetry:
		ENDPOINT = '/ietf-network:networks/network={:s}/node={:s}/simap-telemetry:simap-telemetry'

		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

		def create(
		self, cpu_utilization : float, related_service_ids : List[str] = []
		) -> None:
		endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id)
		telemetry = {
		'cpu-utilization': '{:.2f}'.format(cpu_utilization),
		}
		if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids
		node = {'node-id': self._node_id, 'simap-telemetry:simap-telemetry': telemetry}
		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 = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id)
		telemetry : Dict = self._restconf_client.get(endpoint)
		return telemetry

		def update(
		self, cpu_utilization : float, related_service_ids : List[str] = []
		) -> None:
		endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id)
		telemetry = {
		'cpu-utilization': '{:.2f}'.format(cpu_utilization),
		}
		if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids
		node = {'node-id': self._node_id, 'simap-telemetry:simap-telemetry': telemetry}
		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 = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id)
		self._restconf_client.delete(endpoint)


		class Node:
		ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}'
		ENDPOINT_ID = ENDPOINT_NO_ID + '/node={:s}'
		@@ -71,6 +116,13 @@ class Node:
		self._network_id = network_id
		self._node_id = node_id
		self._tps : Dict[str, TerminationPoint] = dict()
		self._telemetry : Optional[NodeTelemetry] = None

		@property
		def telemetry(self) -> NodeTelemetry:
		if self._telemetry is None:
		self._telemetry = NodeTelemetry(self._restconf_client, self._network_id, self._node_id)
		return self._telemetry

		def termination_points(self) -> List[Dict]:
		tps : Dict = self._restconf_client.get(TerminationPoint.ENDPOINT_NO_ID)
		@@ -119,6 +171,55 @@ class Node:
		endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id)
		self._restconf_client.delete(endpoint)


		class LinkTelemetry:
		ENDPOINT = '/ietf-network:networks/network={:s}/ietf-network-topology:link={:s}/simap-telemetry:simap-telemetry'

		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, bandwidth_utilization : float, latency : float,
		related_service_ids : List[str] = []
		) -> None:
		endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id)
		telemetry = {
		'bandwidth-utilization': bandwidth_utilization,
		'latency' : latency,
		}
		if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids
		link = {'link-id': self._link_id, 'simap-telemetry:simap-telemetry': telemetry}
		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 = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id)
		telemetry : Dict = self._restconf_client.get(endpoint)
		return telemetry

		def update(
		self, bandwidth_utilization : float, latency : float,
		related_service_ids : List[str] = []
		) -> None:
		endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id)
		telemetry = {
		'bandwidth-utilization': '{:.2f}'.format(bandwidth_utilization),
		'latency' : '{:.3f}'.format(latency),
		}
		if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids
		link = {'link-id': self._link_id, 'simap-telemetry:simap-telemetry': telemetry}
		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 = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id)
		self._restconf_client.delete(endpoint)


		class Link:
		ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}'
		ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:link={:s}'
		@@ -127,6 +228,13 @@ class Link:
		self._restconf_client = restconf_client
		self._network_id = network_id
		self._link_id = link_id
		self._telemetry : Optional[LinkTelemetry] = None

		@property
		def telemetry(self) -> LinkTelemetry:
		if self._telemetry is None:
		self._telemetry = LinkTelemetry(self._restconf_client, self._network_id, self._link_id)
		return self._telemetry

		def create(
		self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str,

src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/SimapClient.py

+109 −1

Original line number	Diff line number	Diff line
		@@ -13,7 +13,7 @@
		# limitations under the License.


		from typing import Dict, List, Tuple
		from typing import Dict, List, Optional, Tuple
		from common.tools.rest_conf.client.RestConfClient import RestConfClient


		@@ -62,6 +62,51 @@ class TerminationPoint:
		endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id)
		self._restconf_client.delete(endpoint)


		class NodeTelemetry:
		ENDPOINT = '/ietf-network:networks/network={:s}/node={:s}/simap-telemetry:simap-telemetry'

		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

		def create(
		self, cpu_utilization : float, related_service_ids : List[str] = []
		) -> None:
		endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id)
		telemetry = {
		'cpu-utilization': '{:.2f}'.format(cpu_utilization),
		}
		if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids
		node = {'node-id': self._node_id, 'simap-telemetry:simap-telemetry': telemetry}
		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 = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id)
		telemetry : Dict = self._restconf_client.get(endpoint)
		return telemetry

		def update(
		self, cpu_utilization : float, related_service_ids : List[str] = []
		) -> None:
		endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id)
		telemetry = {
		'cpu-utilization': '{:.2f}'.format(cpu_utilization),
		}
		if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids
		node = {'node-id': self._node_id, 'simap-telemetry:simap-telemetry': telemetry}
		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 = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id)
		self._restconf_client.delete(endpoint)


		class Node:
		ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}'
		ENDPOINT_ID = ENDPOINT_NO_ID + '/node={:s}'
		@@ -71,6 +116,13 @@ class Node:
		self._network_id = network_id
		self._node_id = node_id
		self._tps : Dict[str, TerminationPoint] = dict()
		self._telemetry : Optional[NodeTelemetry] = None

		@property
		def telemetry(self) -> NodeTelemetry:
		if self._telemetry is None:
		self._telemetry = NodeTelemetry(self._restconf_client, self._network_id, self._node_id)
		return self._telemetry

		def termination_points(self) -> List[Dict]:
		tps : Dict = self._restconf_client.get(TerminationPoint.ENDPOINT_NO_ID)
		@@ -119,6 +171,55 @@ class Node:
		endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id)
		self._restconf_client.delete(endpoint)


		class LinkTelemetry:
		ENDPOINT = '/ietf-network:networks/network={:s}/ietf-network-topology:link={:s}/simap-telemetry:simap-telemetry'

		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, bandwidth_utilization : float, latency : float,
		related_service_ids : List[str] = []
		) -> None:
		endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id)
		telemetry = {
		'bandwidth-utilization': bandwidth_utilization,
		'latency' : latency,
		}
		if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids
		link = {'link-id': self._link_id, 'simap-telemetry:simap-telemetry': telemetry}
		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 = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id)
		telemetry : Dict = self._restconf_client.get(endpoint)
		return telemetry

		def update(
		self, bandwidth_utilization : float, latency : float,
		related_service_ids : List[str] = []
		) -> None:
		endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id)
		telemetry = {
		'bandwidth-utilization': '{:.2f}'.format(bandwidth_utilization),
		'latency' : '{:.3f}'.format(latency),
		}
		if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids
		link = {'link-id': self._link_id, 'simap-telemetry:simap-telemetry': telemetry}
		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 = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id)
		self._restconf_client.delete(endpoint)


		class Link:
		ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}'
		ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:link={:s}'
		@@ -127,6 +228,13 @@ class Link:
		self._restconf_client = restconf_client
		self._network_id = network_id
		self._link_id = link_id
		self._telemetry : Optional[LinkTelemetry] = None

		@property
		def telemetry(self) -> LinkTelemetry:
		if self._telemetry is None:
		self._telemetry = LinkTelemetry(self._restconf_client, self._network_id, self._link_id)
		return self._telemetry

		def create(
		self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str,