Test - Tools - Mock NCE-T Ctrl (7e619115) · Commits · TFS / controller

src/tests/tools/mock_nce_t_ctrl/nce_t_client/main.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -26,7 +26,7 @@ LOGGER = logging.getLogger(__name__)

		def main() -> None:
		restconf_client = RestConfClient(
		'172.17.0.1', port=8081,
		'172.17.0.1', port=8081, restconf_version='v2',
		logger=logging.getLogger('RestConfClient')
		)

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

0 → 100644

+80 −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.

		# REST-API resource implementing minimal support for "IETF YANG Data Model for Transport Network Client Signals".
		# Ref: https://www.ietf.org/archive/id/draft-ietf-ccamp-client-signal-yang-10.html

		from flask import abort, jsonify, make_response, request
		from flask_restful import Resource
		from .SimapUpdater import SimapUpdater

		ETHT_SERVICES = {}

		class EthServices(Resource):
		def __init__(self, simap_updater : SimapUpdater):
		super().__init__()
		self._simap_updater = simap_updater

		def get(self):
		etht_services = [etht_service for etht_service in ETHT_SERVICES.values()]
		data = {'ietf-eth-tran-service:etht-svc': {'etht-svc-instances': etht_services}}
		return make_response(jsonify(data), 200)

		def post(self):
		json_request = request.get_json()
		if not json_request: abort(400)
		if not isinstance(json_request, dict): abort(400)
		if 'ietf-eth-tran-service:etht-svc' not in json_request: abort(400)
		json_request = json_request['ietf-eth-tran-service:etht-svc']
		if 'etht-svc-instances' not in json_request: abort(400)
		etht_services = json_request['etht-svc-instances']
		if not isinstance(etht_services, list): abort(400)
		if len(etht_services) != 1: abort(400)
		etht_service = etht_services[0]
		etht_service_name = etht_service['etht-svc-name']
		ETHT_SERVICES[etht_service_name] = etht_service
		self._simap_updater.create_simap_trans_otn(etht_service)
		return make_response(jsonify({}), 201)

		class EthService(Resource):
		def __init__(self, simap_updater : SimapUpdater):
		super().__init__()
		self._simap_updater = simap_updater

		def get(self, etht_service_name : str):
		etht_service = ETHT_SERVICES.get(etht_service_name, None)
		data,status = ({}, 404) if etht_service is None else (etht_service, 200)
		return make_response(jsonify(data), status)

		def post(self, etht_service_name : str):
		json_request = request.get_json()
		if not json_request: abort(400)
		if not isinstance(json_request, dict): abort(400)
		if 'ietf-eth-tran-service:etht-svc' not in json_request: abort(400)
		json_request = json_request['ietf-eth-tran-service:etht-svc']
		if 'etht-svc-instances' not in json_request: abort(400)
		etht_services = json_request['etht-svc-instances']
		if not isinstance(etht_services, list): abort(400)
		if len(etht_services) != 1: abort(400)
		etht_service = etht_services[0]
		assert etht_service_name == etht_service['etht-svc-name']
		ETHT_SERVICES[etht_service_name] = etht_service
		self._simap_updater.create_simap_trans_otn(etht_service)
		return make_response(jsonify({}), 201)

		def delete(self, etht_service_name : str):
		etht_service = ETHT_SERVICES.pop(etht_service_name, None)
		data,status = ({}, 404) if etht_service is None else (etht_service, 204)
		self._simap_updater.delete_simap_trans_otn(etht_service_name)
		return make_response(jsonify(data), status)

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

0 → 100644

+85 −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.

		# REST-API resource implementing minimal support for "IETF YANG Data Model for Traffic Engineering Tunnels,
		# Label Switched Paths and Interfaces".
		# Ref: https://www.ietf.org/archive/id/draft-ietf-teas-yang-te-34.html


		from flask import abort, jsonify, make_response, request
		from flask_restful import Resource
		from .SimapUpdater import SimapUpdater

		OSU_TUNNELS = {}

		class OsuTunnels(Resource):
		def __init__(self, simap_updater : SimapUpdater):
		super().__init__()
		self._simap_updater = simap_updater

		def get(self):
		osu_tunnels = [osu_tunnel for osu_tunnel in OSU_TUNNELS.values()]
		data = {'ietf-te:te': {'tunnels': {'tunnel': osu_tunnels}}}
		return make_response(jsonify(data), 200)

		def post(self):
		json_request = request.get_json()
		if not json_request: abort(400)
		if not isinstance(json_request, dict): abort(400)
		if 'ietf-te:te' not in json_request: abort(400)
		te_data = json_request['ietf-te:te']
		if not isinstance(te_data, dict): abort(400)
		if 'tunnels' not in te_data: abort(400)
		te_tunnels = te_data['tunnels']
		if 'tunnel' not in te_tunnels: abort(400)
		osu_tunnels = te_tunnels['tunnel']
		if not isinstance(osu_tunnels, list): abort(400)
		if len(osu_tunnels) != 1: abort(400)
		osu_tunnel = osu_tunnels[0]
		osu_tunnel_name = osu_tunnel['name']
		OSU_TUNNELS[osu_tunnel_name] = osu_tunnel
		return make_response(jsonify({}), 201)

		class OsuTunnel(Resource):
		def __init__(self, simap_updater : SimapUpdater):
		super().__init__()
		self._simap_updater = simap_updater

		def get(self, osu_tunnel_name : str):
		osu_tunnel = OSU_TUNNELS.get(osu_tunnel_name, None)
		data,status = ({}, 404) if osu_tunnel is None else (osu_tunnel, 200)
		return make_response(jsonify(data), status)

		def post(self, osu_tunnel_name : str):
		json_request = request.get_json()
		if not json_request: abort(400)
		if not isinstance(json_request, dict): abort(400)
		if 'ietf-te:te' not in json_request: abort(400)
		te_data = json_request['ietf-te:te']
		if not isinstance(te_data, dict): abort(400)
		if 'tunnels' not in te_data: abort(400)
		te_tunnels = te_data['tunnels']
		if 'tunnel' not in te_tunnels: abort(400)
		osu_tunnels = te_tunnels['tunnel']
		if not isinstance(osu_tunnels, list): abort(400)
		if len(osu_tunnels) != 1: abort(400)
		osu_tunnel = osu_tunnels[0]
		assert osu_tunnel_name == osu_tunnel['name']
		OSU_TUNNELS[osu_tunnel_name] = osu_tunnel
		return make_response(jsonify({}), 201)

		def delete(self, osu_tunnel_name : str):
		osu_tunnel = OSU_TUNNELS.pop(osu_tunnel_name, None)
		data,status = ({}, 404) if osu_tunnel is None else (osu_tunnel, 204)
		return make_response(jsonify(data), status)

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

+54 −0

Original line number	Diff line number	Diff line
		@@ -66,3 +66,57 @@ class SimapUpdater:
		te_topo.link(link_id).create(
		link_src_node_id, link_src_tp_id, link_dst_node_id, link_dst_tp_id
		)


		def create_simap_trans_otn(self, etht_service : Dict) -> None:
		#etht_svc_name = etht_service['etht-svc-name']

		#src_node_ep = etht_service['source-endpoints']['source-endpoint'][0]
		#src_node_id = src_node_ep['node-id']
		#src_tp_id = src_node_ep['tp-id']

		#dst_node_ep = etht_service['destination-endpoints']['destination-endpoint'][0]
		#dst_node_id = dst_node_ep['node-id']
		#dst_tp_id = dst_node_ep['tp-id']

		simap = self._simap_client.network('trans-otn')
		simap.update(supporting_network_ids=['admin'])

		node_a = simap.node('site1')
		node_a.update(supporting_node_ids=[('admin', 'O-PE1')])
		node_a.termination_point('200').update(supporting_termination_point_ids=[('admin', 'O-PE1', '200')])
		node_a.termination_point('500').update(supporting_termination_point_ids=[('admin', 'O-PE1', '500')])
		node_a.termination_point('501').update(supporting_termination_point_ids=[('admin', 'O-PE1', '501')])

		node_b = simap.node('site2')
		node_b.update(supporting_node_ids=[('admin', 'O-PE2')])
		node_b.termination_point('200').update(supporting_termination_point_ids=[('admin', 'O-PE2', '200')])
		node_b.termination_point('500').update(supporting_termination_point_ids=[('admin', 'O-PE2', '500')])
		node_b.termination_point('501').update(supporting_termination_point_ids=[('admin', 'O-PE2', '501')])

		link_ab = simap.link('Trans-L1ab')
		link_ab.update(
		'site1', '500', 'site2', '500',
		supporting_link_ids=[
		('admin', 'L7ab'), ('admin', 'L11ab'),
		]
		)

		link_ba = simap.link('Trans-L1ba')
		link_ba.update(
		'site2', '500', 'site1', '500',
		supporting_link_ids=[
		('admin', 'L11ba'), ('admin', 'L7ba'),
		]
		)


		def delete_simap_trans_otn(self, etht_svc_name : str) -> None:
		simap = self._simap_client.network('trans-otn')
		simap.update(supporting_network_ids=['admin'])

		link_ab = simap.link('Trans-L1ab')
		link_ab.delete()

		link_ba = simap.link('Trans-L1ba')
		link_ba.delete()

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

+40 −2

Original line number	Diff line number	Diff line
		@@ -15,10 +15,21 @@

		# This file overwrites default RestConf Server `app.py` file.

		# Mock IETF ACTN SDN controller
		# -----------------------------
		# REST server implementing minimal support for:
		# - IETF YANG Data Model for Transport Network Client Signals
		# Ref: https://www.ietf.org/archive/id/draft-ietf-ccamp-client-signal-yang-10.html
		# - IETF YANG Data Model for Traffic Engineering Tunnels, Label Switched Paths and Interfaces
		# Ref: https://www.ietf.org/archive/id/draft-ietf-teas-yang-te-34.html

		# NOTE: we need here OSUflex tunnels that are still not standardized; hardcoded.

		import logging
		from common.tools.rest_conf.server.restconf_server.RestConfServerApplication import RestConfServerApplication
		from .Callbacks import CallbackEthTService, CallbackOsuTunnel
		from .ResourceEthServices import EthService, EthServices
		from .ResourceOsuTunnels import OsuTunnel, OsuTunnels
		from .SimapUpdater import SimapUpdater


		@@ -31,10 +42,38 @@ logging.getLogger('RestConfClient').setLevel(logging.WARN)


		LOGGER.info('Starting...')
		rcs_app = RestConfServerApplication()

		simap_updater = SimapUpdater()

		rcs_app = RestConfServerApplication()
		rcs_app.register_host_meta()
		rcs_app.register_restconf()

		rcs_app.register_custom(
		OsuTunnels,
		'/restconf/v2/data/ietf-te:te/tunnels',
		add_prefix_to_urls=False,
		resource_class_args=(simap_updater,)
		)
		rcs_app.register_custom(
		OsuTunnel,
		'/restconf/v2/data/ietf-te:te/tunnels/tunnel=<string:osu_tunnel_name>',
		add_prefix_to_urls=False,
		resource_class_args=(simap_updater,)
		)
		rcs_app.register_custom(
		EthServices,
		'/restconf/v2/data/ietf-eth-tran-service:etht-svc',
		add_prefix_to_urls=False,
		resource_class_args=(simap_updater,)
		)
		rcs_app.register_custom(
		EthService,
		'/restconf/v2/data/ietf-eth-tran-service:etht-svc/etht-svc-instances=<string:etht_service_name>',
		add_prefix_to_urls=False,
		resource_class_args=(simap_updater,)
		)

		LOGGER.info('All connectors registered')

		startup_data = rcs_app.get_startup_data()
		@@ -42,7 +81,6 @@ startup_data = rcs_app.get_startup_data()
		networks = startup_data.get('ietf-network:networks', dict())
		networks = networks.get('network', list())
		if len(networks) == 1 and networks[0]['network-id'] == 'admin':
		simap_updater = SimapUpdater()
		simap_updater.upload_topology(networks[0])

		rcs_app.callback_dispatcher.register(CallbackOsuTunnel())