NBI component - TFS API connector: (8bfd93ab) · Commits · TFS / controller

src/nbi/service/tfs_api/Resources.py

+68 −0

Original line number	Diff line number	Diff line
		@@ -254,6 +254,74 @@ class Service(_Resource):
		def delete(self, context_uuid : str, service_uuid : str):
		return format_grpc_to_json(self.service_client.DeleteService(grpc_service_id(context_uuid, service_uuid)))

		def _decode_custom_value(value):
		if not isinstance(value, str):
		return value
		try:
		return json.loads(value)
		except json.JSONDecodeError:
		return value

		def _service_custom_settings(service):
		settings = {}
		for config_rule in service.get('service_config', {}).get('config_rules', []):
		custom = config_rule.get('custom')
		if custom is None:
		continue
		resource_key = custom.get('resource_key')
		if resource_key is None:
		continue
		settings[resource_key] = _decode_custom_value(custom.get('resource_value'))
		return settings

		def _slot_range(slots):
		if not isinstance(slots, list) or len(slots) == 0:
		return None
		slot_values = sorted([int(slot) for slot in slots])
		if slot_values == list(range(slot_values[0], slot_values[-1] + 1)):
		return {
		'n_start': slot_values[0],
		'n_end': slot_values[-1],
		}
		return None

		class OpticalServiceAllocation(_Resource):
		def get(self, context_uuid : str, service_uuid : str):
		service_id = grpc_service_id(context_uuid, service_uuid)
		service = grpc_message_to_json(self.context_client.GetService(service_id))
		connections = grpc_message_to_json(self.context_client.ListConnections(service_id))
		settings = _service_custom_settings(service)
		allocation_settings = settings.get('/settings')
		if not isinstance(allocation_settings, dict):
		return {
		'error': 'OPTICAL_ALLOCATION_NOT_FOUND',
		'message': 'Service does not expose optical allocation settings',
		}, 404
		if 'flow_id' not in allocation_settings:
		return {
		'error': 'OPTICAL_ALLOCATION_NOT_FOUND',
		'message': 'Service settings do not include an optical flow identifier',
		}, 404

		slots = allocation_settings.get('slots')
		allocation = {
		'service_id': service.get('service_id'),
		'service_status': service.get('service_status'),
		'service_type': service.get('service_type'),
		'flow_id': allocation_settings.get('flow_id'),
		'optical_band_id': allocation_settings.get('ob_id'),
		'band_type': allocation_settings.get('band_type'),
		'frequency': allocation_settings.get('frequency'),
		'bandwidth': allocation_settings.get('band'),
		'slots': slots,
		'slot_range': _slot_range(slots),
		'path': allocation_settings.get('path'),
		'links': allocation_settings.get('links'),
		'bidirectional': bool(allocation_settings.get('bidir', 0)),
		'connections': connections.get('connections', []),
		}
		return jsonify(allocation)

		class SliceIds(_Resource):
		def get(self, context_uuid : str):
		return format_grpc_to_json(self.context_client.ListSliceIds(grpc_context_id(context_uuid)))

src/nbi/service/tfs_api/init.py

+3 −1

Original line number	Diff line number	Diff line
		@@ -24,7 +24,7 @@ from .Resources import (
		OpticalSpectrumReservationRelease, OpticalSpectrumReservations,
		OpticalConnectivityCandidates,
		PolicyRule, PolicyRuleIds, PolicyRules,
		Service, ServiceIds, Services,
		Service, ServiceIds, Services, OpticalServiceAllocation,
		Slice, SliceIds, Slices,
		Topologies, Topology, TopologyDetails, TopologyIds,
		E2epathcomp
		@@ -51,6 +51,8 @@ _RESOURCES = [
		('api.service_ids', ServiceIds, '/context/<path:context_uuid>/service_ids'),
		('api.services', Services, '/context/<path:context_uuid>/services'),
		('api.service', Service, '/context/<path:context_uuid>/service/<path:service_uuid>'),
		('api.optical_service_allocation', OpticalServiceAllocation,
		'/context/<path:context_uuid>/service/<path:service_uuid>/optical_allocation'),

		('api.slice_ids', SliceIds, '/context/<path:context_uuid>/slice_ids'),
		('api.slices', Slices, '/context/<path:context_uuid>/slices'),

src/nbi/tests/test_tfs_api_optical_allocation.py

0 → 100644

+40 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2026 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 nbi.service.tfs_api.Resources import _service_custom_settings, _slot_range


		def test_slot_range_compacts_contiguous_slots():
		assert _slot_range([53, 51, 52]) == {'n_start': 51, 'n_end': 53}


		def test_slot_range_returns_none_for_sparse_slots():
		assert _slot_range([51, 53]) is None


		def test_service_custom_settings_decodes_json_values():
		service = {
		'service_config': {
		'config_rules': [{
		'custom': {
		'resource_key': '/settings',
		'resource_value': '{"flow_id": 1, "slots": [51, 52]}',
		},
		}],
		},
		}

		settings = _service_custom_settings(service)

		assert settings['/settings'] == {'flow_id': 1, 'slots': [51, 52]}