Tests - Tools - Mock NCE FAN Ctrl: (1e344945) · Commits · TFS / controller

src/tests/tools/mock_nce_fan_ctrl/nce_fan_client/Requests.py

0 → 100644

+54 −0

Original line number	Diff line number	Diff line
		QOS_PROFILE_NAME = 'AR_VR_Gaming'
		URL_QOS_PROFILE_ITEM = '/huawei-nce-app-flow:qos-profiles/qos-profile={:s}'.format(QOS_PROFILE_NAME)
		REQUEST_QOS_PROFILE = {"huawei-nce-app-flow:qos-profiles": {"qos-profile": [
		{
		"downstream": {
		"assure-bandwidth": "1000000000",
		"max-bandwidth": "2000000000"
		},
		"max-jitter": 10,
		"max-latency": 10,
		"max-loss": "0.001",
		"name": QOS_PROFILE_NAME,
		"upstream": {
		"assure-bandwidth": "5000000000",
		"max-bandwidth": "10000000000"
		}
		}
		]}}

		APPLICATION_NAME = 'App_1_2_slice1'
		URL_APPLICATION_ITEM = '/huawei-nce-app-flow:applications/application={:s}'.format(APPLICATION_NAME)
		REQUEST_APPLICATION = {"huawei-nce-app-flow:applications": {"application": [
		{
		"app-features": {
		"app-feature": [
		{
		"dest-ip": "172.1.101.22",
		"dest-port": "10200",
		"id": "feature_1_2_slice1",
		"protocol": "tcp",
		"src-ip": "172.16.204.221",
		"src-port": "10500"
		}
		]
		},
		"app-id": ["app_1_2_slice1"],
		"name": APPLICATION_NAME
		}
		]}}

		APP_FLOW_NAME = "App_Flow_1_2_slice1"
		URL_APP_FLOW_ITEM = '/huawei-nce-app-flow:app-flows/app-flow={:s}'.format(APP_FLOW_NAME)
		REQUEST_APP_FLOW = {"huawei-nce-app-flow:app-flows": {"app-flow": [
		{
		"app-name": APPLICATION_NAME,
		"duration": 9999,
		"max-online-users": 1,
		"name": APP_FLOW_NAME,
		"qos-profile": QOS_PROFILE_NAME,
		"service-profile": "service_1_2_slice1",
		"stas": ["00:3D:E1:18:82:9E"],
		"user-id": "ad2c2a94-3415-4676-867a-39eedfb9f205"
		}
		]}}

src/tests/tools/mock_nce_fan_ctrl/nce_fan_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.

src/tests/tools/mock_nce_fan_ctrl/nce_fan_client/main.py

0 → 100644

+45 −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 logging
		from common.tools.rest_conf.client.RestConfClient import RestConfClient
		from .Requests import (
		URL_QOS_PROFILE_ITEM, REQUEST_QOS_PROFILE,
		URL_APPLICATION_ITEM, REQUEST_APPLICATION,
		URL_APP_FLOW_ITEM, REQUEST_APP_FLOW,
		)

		logging.basicConfig(level=logging.INFO)
		logging.getLogger('RestConfClient').setLevel(logging.DEBUG)
		LOGGER = logging.getLogger(__name__)

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

		LOGGER.info('Creating QoS Profile: {:s}'.format(str(REQUEST_QOS_PROFILE)))
		restconf_client.post(URL_QOS_PROFILE_ITEM, body=REQUEST_QOS_PROFILE)

		LOGGER.info('Creating Application: {:s}'.format(str(REQUEST_APPLICATION)))
		restconf_client.post(URL_APPLICATION_ITEM, body=REQUEST_APPLICATION)

		LOGGER.info('Creating App Flow: {:s}'.format(str(REQUEST_APP_FLOW)))
		restconf_client.post(URL_APP_FLOW_ITEM, body=REQUEST_APP_FLOW)


		if __name__ == '__main__':
		main()

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

0 → 100644

+69 −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 logging, re
		from typing import Dict, Optional
		from common.tools.rest_conf.server.restconf_server.Callbacks import _Callback


		LOGGER = logging.getLogger(__name__)


		class CallbackQosProfile(_Callback):
		def __init__(self) -> None:
		pattern = r'/restconf/data'
		pattern += r'/huawei-nce-app-flow:qos-profiles'
		pattern += r'/qos-profile=(?P<qos_profile_name>[^/]+)'
		super().__init__(pattern)

		def execute(
		self, match : re.Match, path : str, old_data : Optional[Dict],
		new_data : Optional[Dict]
		) -> bool:
		MSG = '[on_qos_profile] match={:s} path={:s} old_data={:s} new_data={:s}'
		LOGGER.warning(MSG.format(str(match.groupdict()), str(path), str(old_data), str(new_data)))
		return False


		class CallbackApplication(_Callback):
		def __init__(self) -> None:
		pattern = r'/restconf/data'
		pattern += r'/huawei-nce-app-flow:applications'
		pattern += r'/application=(?P<application_name>[^/]+)'
		super().__init__(pattern)

		def execute(
		self, match : re.Match, path : str, old_data : Optional[Dict],
		new_data : Optional[Dict]
		) -> bool:
		MSG = '[on_application] match={:s} path={:s} old_data={:s} new_data={:s}'
		LOGGER.warning(MSG.format(str(match.groupdict()), str(path), str(old_data), str(new_data)))
		return False


		class CallbackAppFlow(_Callback):
		def __init__(self) -> None:
		pattern = r'/restconf/data'
		pattern += r'/huawei-nce-app-flow:app-flows'
		pattern += r'/app-flow=(?P<app_flow_name>[^/]+)'
		super().__init__(pattern)

		def execute(
		self, match : re.Match, path : str, old_data : Optional[Dict],
		new_data : Optional[Dict]
		) -> bool:
		MSG = '[on_app_flow] match={:s} path={:s} old_data={:s} new_data={:s}'
		LOGGER.warning(MSG.format(str(match.groupdict()), str(path), str(old_data), str(new_data)))
		return False

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

+7 −0

Original line number	Diff line number	Diff line
		@@ -18,6 +18,7 @@

		import logging
		from common.tools.rest_conf.server.restconf_server.RestConfServerApplication import RestConfServerApplication
		from .Callbacks import CallbackApplication, CallbackAppFlow, CallbackQosProfile
		from .SimapUpdater import SimapUpdater


		@@ -31,6 +32,7 @@ logging.getLogger('RestConfClient').setLevel(logging.WARN)

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

		rcs_app.register_host_meta()
		rcs_app.register_restconf()
		LOGGER.info('All connectors registered')
		@@ -43,6 +45,11 @@ if len(networks) == 1 and networks[0]['network-id'] == 'admin':
		simap_updater = SimapUpdater()
		simap_updater.upload_topology(networks[0])

		rcs_app.callback_dispatcher.register(CallbackApplication())
		rcs_app.callback_dispatcher.register(CallbackAppFlow())
		rcs_app.callback_dispatcher.register(CallbackQosProfile())
		LOGGER.info('All callbacks registered')

		rcs_app.dump_configuration()
		app = rcs_app.get_flask_app()