Updated version of CAMARA NBI (c6b134a6) · Commits · TFS / controller

scripts/run_tests_locally-nbi-camara-qod.sh

0 → 100755

+30 −0

Original line number	Diff line number	Diff line
		#!/bin/bash
		# Copyright 2022-2024 ETSI OSG/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.

		PROJECTDIR=`pwd`

		cd $PROJECTDIR/src
		RCFILE=$PROJECTDIR/coverage/.coveragerc
		COVERAGEFILE=$PROJECTDIR/coverage/.coverage

		# Destroy old coverage file and configure the correct folder on the .coveragerc file
		rm -f $COVERAGEFILE
		cat $PROJECTDIR/coverage/.coveragerc.template \| sed s+~/tfs-ctrl+$PROJECTDIR+g > $RCFILE


		# Run unitary tests and analyze coverage of code at same time
		# helpful pytest flags: --log-level=INFO -o log_cli=true --verbose --maxfail=1 --durations=0
		coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
		nbi/tests/test_camara_qod_profile.py nbi/tests/test_camara_qos_service.py

src/nbi/service/rest_server/nbi_plugins/camara_qod/Resources.py

+26 −56

Original line number	Diff line number	Diff line
		@@ -11,27 +11,30 @@
		# 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 venv import logger
		from flask.json import jsonify
		from flask_restful import Resource, request
		from enum import Enum
		import grpc._channel
		from qos_profile.service.database.QoSProfile import grpc_message_to_qos_table_data
		from qos_profile.tests.test_crud import create_qos_profile_from_json,test_update_qos_profile
		from qos_profile.client.QoSProfileClient import QoSProfileClient
		from werkzeug.exceptions import UnsupportedMediaType
		from common.proto.context_pb2 import QoSProfile, QoSProfileId, Uuid, QoSProfileValueUnitPair,Empty
		from common.proto.qos_profile_pb2 import QoSProfile, QoDConstraintsRequest
		from context.client.ContextClient import ContextClient
		from service.client.ServiceClient import ServiceClient
		from common.proto.context_pb2 import QoSProfileId, Uuid,Empty
		from common.proto.qos_profile_pb2 import QoDConstraintsRequest
		from typing import Dict
		from uuid import uuid4
		import grpc, logging
		import copy, deepmerge, json, logging
		from typing import Dict
		from flask_restful import Resource, request
		from werkzeug.exceptions import UnsupportedMediaType
		from common.Constants import DEFAULT_CONTEXT_NAME
		from context.client.ContextClient import ContextClient
		from service.client.ServiceClient import ServiceClient
		from .Tools import (
		format_grpc_to_json, grpc_context_id, grpc_service_id, QOD_2_service, service_2_qod,grpc_message_to_qos_table_data,create_qos_profile_from_json
		)


		LOGGER = logging.getLogger(__name__)


		#Initiate the QoSProfileClient
		class _Resource(Resource):
		def __init__(self) -> None:
		@@ -74,30 +77,6 @@ class ProfileList(_Resource):

		return jsonify(qos_profile_list)

		class ProfileListCons(_Resource):
		def get(self):
		qos_profile_id = request.args.get("qos_profile_id")
		start_timestamp = request.args.get("start_timestamp", type=float)
		duration = request.args.get("duration", type=int)

		qos_constraints_request = QoDConstraintsRequest(
		qos_profile_id=qos_profile_id,
		start_timestamp=start_timestamp,
		duration=duration
		)
		try:
		qos_profiles = self.qos_profile_client.GetConstraintListFromQoSProfile(qos_constraints_request)
		qos_profile_list = [
		grpc_message_to_qos_table_data(profile) for profile in qos_profiles
		]
		return jsonify(qos_profile_list)
		except grpc._channel._InactiveRpcError as exc:
		LOGGER.error(f"gRPC error while fetching constraints: {exc}")
		return {"error": "Internal Server Error"}, 500
		except Exception as e:
		LOGGER.error(f"Error while fetching constraints: {e}")
		return {"error": "Internal Server Error"}, 500

		#getting,updating,deleting using the qos profile id
		class ProfileDetail(_Resource):
		def get(self, qos_profile_id):
		@@ -160,25 +139,8 @@ class ProfileDetail(_Resource):
		LOGGER.error(f"Error in DELETE /profiles/{qos_profile_id}: {e}")
		return {"error": "Internal Server Error"}, 500


		import copy, deepmerge, json, logging
		from typing import Dict
		from flask_restful import Resource, request
		from werkzeug.exceptions import UnsupportedMediaType
		from common.Constants import DEFAULT_CONTEXT_NAME
		from context.client.ContextClient import ContextClient
		from service.client.ServiceClient import ServiceClient
		from .Tools import (
		format_grpc_to_json, grpc_context_id, grpc_service_id, QOD_2_service, service_2_qod
		)

		###SESSION##########################################################
		LOGGER = logging.getLogger(__name__)
		class _Resource(Resource):
		def __init__(self) -> None:
		super().__init__()
		self.client = ContextClient()
		self.service_client = ServiceClient()

		class qodinfo(_Resource):
		def post(self):
		if not request.is_json:
		@@ -186,12 +148,13 @@ class qodinfo(_Resource):
		request_data: Dict = request.get_json()
		qos_profile_id = request_data.get('qos_profile_id')
		qos_session_id = request_data.get('qos_session_id')
		duration = request_data.get('duration')
		LOGGER.info(f'qos_profile_id:{qos_profile_id}')
		if not qos_profile_id:
		return jsonify({'error': 'qos_profile_id is required'}), 400
		if qos_session_id:
		return jsonify({'error': 'qos_session_id is not allowed in creation'}), 400
		service = QOD_2_service(self.client, request_data,qos_profile_id)
		service = QOD_2_service(self.client, request_data,qos_profile_id,duration)
		stripped_service = copy.deepcopy(service)
		stripped_service.ClearField('service_endpoint_ids')
		stripped_service.ClearField('service_constraints')
		@@ -206,8 +169,8 @@ class qodinfo(_Resource):
		return response

		def get(self):
		service_list = self.client.ListServices(grpc_context_id(DEFAULT_CONTEXT_NAME))
		qod_info = [service_2_qod(service) for service in service_list.services]
		service_list = self.client.ListServices(grpc_context_id(DEFAULT_CONTEXT_NAME)) #return context id as json
		qod_info = [service_2_qod(service) for service in service_list.services] #iterating over service list
		LOGGER.info(f"error related to qod_info: {qod_info}")
		return qod_info

		@@ -231,7 +194,14 @@ class qodinfoId(_Resource):
		qos_profile_id = request_data.get('qos_profile_id')
		if not qos_profile_id:
		return jsonify({'error': 'qos_profile_id is required'}), 400
		service = self.client.GetService(grpc_service_id(DEFAULT_CONTEXT_NAME, sessionId))
		duration = request_data.get('duration')
		service = self.client.GetService(grpc_service_id(DEFAULT_CONTEXT_NAME, sessionId)) #to get service we should have the context and the session id
		if qos_profile_id:
		service.name = qos_profile_id # if we provide a new qos profile , update the service name with new qos_profile_id
		if duration:
		for constraint in service.service_constraints:
		if constraint.WhichOneof('constraint') == 'schedule':
		constraint.schedule.duration_days = duration
		updated_service = self.service_client.UpdateService(service)
		qod_response = service_2_qod(updated_service)
		return qod_response, 200

src/nbi/service/rest_server/nbi_plugins/camara_qod/Tools.py

+86 −41

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

		import json, logging, re, time
		from decimal import ROUND_HALF_EVEN, Decimal
		from flask.json import jsonify
		from common.proto.context_pb2 import (
		ContextId, Empty, EndPointId, ServiceId, ServiceStatusEnum, ServiceTypeEnum,
		Service, Constraint, Constraint_SLA_Capacity, ConfigRule, ConfigRule_Custom,
		ConfigActionEnum,QoSProfile
		)
		Service,ConfigRule, ConfigRule_Custom,
		ConfigActionEnum)
		from common.tools.grpc.ConfigRules import update_config_rule_custom
		from common.tools.grpc.Tools import grpc_message_to_json
		from common.tools.object_factory.Context import json_context_id
		@@ -27,10 +25,11 @@ from common.tools.object_factory.Service import json_service_id
		from uuid import uuid4
		from nbi.service.rest_server.nbi_plugins.ietf_network.bindings.networks import network
		from qos_profile.client.QoSProfileClient import QoSProfileClient
		#from context.service.database.QoSProfile import grpc_message_to_qos_table_data
		from common.proto.context_pb2 import QoSProfile, QoSProfileId, Uuid, QoSProfileValueUnitPair,Empty,ServiceId
		from context.client.ContextClient import ContextClient
		from common.proto.context_pb2 import QoSProfileId, Uuid,Empty,ServiceId
		from common.proto.context_pb2 import Uuid, QoSProfileId
		from common.proto.qos_profile_pb2 import QoSProfileValueUnitPair, QoSProfile,QoDConstraintsRequest
		import logging
		import grpc
		from netaddr import IPAddress, IPNetwork

		LOGGER = logging.getLogger(__name__)
		@@ -40,17 +39,60 @@ DEVICE_SETTINGS_KEY = '/device[{:s}]/settings'
		RE_CONFIG_RULE_IF_SUBIF = re.compile(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]$')
		MEC_CONSIDERED_FIELDS = ['device', 'applicationServer', 'qosProfile', 'sessionId', 'duration', 'startedAt', 'expiresAt', 'qosStatus']


		def __init__(self) -> None:
		super().__init__()
		self.qos_profile_client = QoSProfileClient()
		self.client = ContextClient()

		def grpc_message_to_qos_table_data(message: QoSProfile) -> dict:
		return {
		'qos_profile_id' : message.qos_profile_id.qos_profile_id.uuid,
		'name' : message.name,
		'description' : message.description,
		'status' : message.status,
		'targetMinUpstreamRate' : grpc_message_to_json(message.targetMinUpstreamRate),
		'maxUpstreamRate' : grpc_message_to_json(message.maxUpstreamRate),
		'maxUpstreamBurstRate' : grpc_message_to_json(message.maxUpstreamBurstRate),
		'targetMinDownstreamRate' : grpc_message_to_json(message.targetMinDownstreamRate),
		'maxDownstreamRate' : grpc_message_to_json(message.maxDownstreamRate),
		'maxDownstreamBurstRate' : grpc_message_to_json(message.maxDownstreamBurstRate),
		'minDuration' : grpc_message_to_json(message.minDuration),
		'maxDuration' : grpc_message_to_json(message.maxDuration),
		'priority' : message.priority,
		'packetDelayBudget' : grpc_message_to_json(message.packetDelayBudget),
		'jitter' : grpc_message_to_json(message.jitter),
		'packetErrorLossRate' : message.packetErrorLossRate,
		}

		def create_qos_profile_from_json(qos_profile_data: dict) -> QoSProfile:
		def create_QoSProfileValueUnitPair(data) -> QoSProfileValueUnitPair:
		return QoSProfileValueUnitPair(value=data['value'], unit=data['unit'])
		qos_profile = QoSProfile()
		qos_profile.qos_profile_id.CopyFrom(QoSProfileId(qos_profile_id=Uuid(uuid=qos_profile_data['qos_profile_id'])))
		qos_profile.name = qos_profile_data['name']
		qos_profile.description = qos_profile_data['description']
		qos_profile.status = qos_profile_data['status']
		qos_profile.targetMinUpstreamRate.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['targetMinUpstreamRate']))
		qos_profile.maxUpstreamRate.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['maxUpstreamRate']))
		qos_profile.maxUpstreamBurstRate.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['maxUpstreamBurstRate']))
		qos_profile.targetMinDownstreamRate.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['targetMinDownstreamRate']))
		qos_profile.maxDownstreamRate.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['maxDownstreamRate']))
		qos_profile.maxDownstreamBurstRate.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['maxDownstreamBurstRate']))
		qos_profile.minDuration.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['minDuration']))
		qos_profile.maxDuration.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['maxDuration']))
		qos_profile.priority = qos_profile_data['priority']
		qos_profile.packetDelayBudget.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['packetDelayBudget']))
		qos_profile.jitter.CopyFrom(create_QoSProfileValueUnitPair(qos_profile_data['jitter']))
		qos_profile.packetErrorLossRate = qos_profile_data['packetErrorLossRate']
		return qos_profile

		def ip_withoutsubnet(ip_withsubnet,neededip):
		network=IPNetwork(ip_withsubnet)
		return IPAddress(neededip) in network


		def QOD_2_service(client,qod_info: dict,qos_profile_id) -> Service:
		def QOD_2_service(client,qod_info: dict,qos_profile_id,duration) -> Service:
		qos_profile_client = QoSProfileClient()
		service = Service()
		service_config_rules = service.service_config.config_rules

		@@ -112,35 +154,44 @@ def QOD_2_service(client,qod_info: dict,qos_profile_id) -> Service:
		update_config_rule_custom(service_config_rules, settings_cr_key, settings_cr_value)
		service.service_status.service_status = ServiceStatusEnum.SERVICESTATUS_PLANNED
		service.service_type = ServiceTypeEnum.SERVICETYPE_L3NM

		qod_info["sessionID"]=str(uuid4())
		qod_info["context"]='admin'
		service.service_id.service_uuid.uuid = qod_info['sessionID']
		service.service_id.context_id.context_uuid.uuid = qod_info["context"]
		#service.service_constraints.CopyFrom() = qod_info['contraints']
		#LOGGER.info(f'this is the error: {qod_info["context"]}')


		try:
		#try:
		# id = QoSProfileId(qos_profile_id=Uuid(uuid=qos_profile_id))
		# id= QoSProfileId()
		# id.qos_profile_id.uuid="qos_profile_id"
		# qos_profile = qos_profile_client.GetQoSProfile(id)
		#except grpc._channel._InactiveRpcError as exc:
		# if exc.code() == grpc.StatusCode.NOT_FOUND:
		# return {"error": f"QoSProfile {qos_profile_id} not found"}, 404
		#if qos_profile.qos_profile_id:
		# qos_profile_id = qod_info.get('qos_profile_id')
		service.name = qod_info.get('qos_profile_id', qos_profile_id)
		current_time = time.time()
		duration_days = duration # days as i saw it in the proto files
		id = QoSProfileId(qos_profile_id=Uuid(uuid=qos_profile_id))
		qos_profile = client.GetQoSProfile(id)
		except grpc._channel._InactiveRpcError as exc:
		if exc.code() == grpc.StatusCode.NOT_FOUND:
		return {"error": f"QoSProfile {qos_profile_id} not found"}, 404

		if qos_profile.qos_profile_id:
		qos_profile_id = qod_info.get('qos_profile_id')
		service.name = qod_info.get('QosProfileId', qos_profile_id)
		request = QoDConstraintsRequest(qos_profile_id=id,start_timestamp=current_time,duration=duration_days) #defined attributes in proto file for the QoDconstraint rquest message
		qos_profiles_constraint = qos_profile_client.GetConstraintListFromQoSProfile(request)
		LOGGER.info(f'current time contains{current_time}')
		LOGGER.info(f'current duration time contains{duration_days}')
		LOGGER.info(f'id : {id}')
		for cs in qos_profiles_constraint:
		if cs.WhichOneof('constraint') == 'schedule' and cs.WhichOneof('constraint') == 'qos_profile': #the method of which one of
		cs.schedule.start_timestamp = current_time
		cs.schedule.duration_days = duration_days
		cs.qos_profile.qos_profile_id=id
		cs.qos_profile.qos_profile_name.CopyFrom(qos_profile_client.name) #i copied this from the qosprofile
		LOGGER.info(f'the cs : {cs}')
		service.service_constraints.append(cs)
		#qos_profile = QoSProfile()
		#qos_profile.qos_profile_id.qos_profile_id.uuid = qod_info['qosProfile']


		#if 'qosProfile' in qos_profile_list:
		# qos_profile = QoSProfile()
		# qos_profile.qos_profile_id.qos_profile_id.uuid = qod_info['qosProfile']



		return service


		@@ -167,20 +218,14 @@ def service_2_qod(service: Service) -> dict:

		if service.service_id:
		response['sessionId'] = service.service_id.service_uuid.uuid
		if service.service_constraints:
		for constraint in service.service_constraints:
		if constraint.WhichOneof('constraint') == 'schedule':
		response['duration'] = float(constraint.schedule.duration_days* (86400))
		LOGGER.info(f'the duration in seconds: {response["duration"]}')
		response['startedAt'] = int(constraint.schedule.start_timestamp)
		response['expiresAt'] = response['startedAt'] + response['duration']

		if service.timestamp:
		response['duration'] = service.timestamp.timestamp
		LOGGER.info(f"time stamp contains{response['duration']}")

		current_time = time.time()
		response['startedAt'] = int(current_time)
		response['expiresAt'] = int(current_time + response['duration'])

		# unixtime = time.time()
		# response['timeStamp'] = {
		# "seconds": int(unixtime),
		# "nanoseconds": int(unixtime % 1 * 1e9)
		# }

		return response

src/nbi/service/rest_server/nbi_plugins/camara_qod/init.py

+1 −5

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

		from nbi.service.rest_server.RestServer import RestServer
		from .Resources import ProfileList, ProfileDetail, qodinfo, qodinfoId, ProfileListCons
		from .Resources import ProfileList, ProfileDetail, qodinfo, qodinfoId

		URL_PREFIX = '/camara/qod/v0'

		@@ -25,10 +25,6 @@ RESOURCES = [
		('camara.qod_info_session_id', qodinfoId, '/sessions/<sessionId>'),
		('camara.qod.profile_list',ProfileList,'/profiles'),
		('camara.qod.profile_detail',ProfileDetail,'/profiles/<string:qos_profile_id>'),
		('camara.qod.profile_all',ProfileListCons,'/profiles/constraints'),
		#('camara.qod.profile_delete_by_name',Profile_delete_by_name,'/profiles/delete_by_name/<string:name>'),
		#('camara.qod.profile_delete_all',Delete_all_profile,'/profiles/delete_all/'),

		]

		def register_camara_qod(rest_server : RestServer):

src/nbi/tests/test_camara_qod_profile.py

+10 −83

Original line number	Diff line number	Diff line
		@@ -90,12 +90,15 @@ def test_create_profile():
		assert post_response['targetMinUpstreamRate'] == get_response['targetMinUpstreamRate']
		#assert response.status_code == 200, f"Failed to retrieve profile with status code {response.status_code}"




		#def test_update_profile():
		# qos_profile_id = '1b4689d8-02a4-4a6c-bd0a-18ffecc1a336'
		# qos_profile_id = '0898e7e8-ef15-4522-8e93-623e31c92efa'
		# qos_profile_data = {
		# "qos_profile_id": "1b4689d8-02a4-4a6c-bd0a-18ffecc1a336",
		# "qos_profile_id": "0898e7e8-ef15-4522-8e93-623e31c92efa",
		# "name": "Updated Name",
		# "description": "Updated Description",
		# "description": "NEW GAMING PROFILE",
		# "status": "ACTIVE",
		# "targetMinUpstreamRate": {
		# "value": 20,
		@@ -140,87 +143,11 @@ def test_create_profile():
		# },
		# "packetErrorLossRate": 1
		#}
		#
		# response = requests.put(f'{BASE_URL}/profiles/{qos_profile_id}', json=qos_profile_data)
		#
		#def test_delete_profile_by_id():
		# qos_profile_id = 'adcbc52e-85e1-42e2-aa33-b6d022798fb3'
		# response = requests.delete(f'{BASE_URL}/profiles/{qos_profile_id}')

		import logging
		from flask import jsonify
		import requests

		logging.basicConfig(level=logging.DEBUG)
		LOGGER = logging.getLogger()

		# Define the base URL for the API
		BASE_URL = 'http://10.1.7.197/camara/qod/v0'

		def test_create_profile():
		# Define the QoS profile data
		qos_profile_data = {
		"name": "QCI_2_voice",
		"description": "QoS profile for video streaming",
		"status": "ACTIVE",
		"targetMinUpstreamRate": {"value": 10, "unit": "bps"},
		"maxUpstreamRate": {"value": 10, "unit": "bps"},
		"maxUpstreamBurstRate": {"value": 10, "unit": "bps"},
		"targetMinDownstreamRate": {"value": 10, "unit": "bps"},
		"maxDownstreamRate": {"value": 10, "unit": "bps"},
		"maxDownstreamBurstRate": {"value": 10, "unit": "bps"},
		"minDuration": {"value": 12, "unit": "Minutes"},
		"maxDuration": {"value": 12, "unit": "Minutes"},
		"priority": 20,
		"packetDelayBudget": {"value": 12, "unit": "Minutes"},
		"jitter": {"value": 12, "unit": "Minutes"},
		"packetErrorLossRate": 3
		}

		# Test profile creation
		post_response = requests.post(f'{BASE_URL}/profiles', json=qos_profile_data).json()
		id = post_response['qos_profile_id']

		# Test profile retrieval
		get_response = requests.get(f'{BASE_URL}/profiles/{id}').json()

		# Assertions to check if post and get responses match
		assert post_response['qos_profile_id'] == get_response['qos_profile_id']
		assert post_response['jitter'] == get_response['jitter']
		assert post_response['maxDownstreamBurstRate'] == get_response['maxDownstreamBurstRate']
		assert post_response['maxDownstreamRate'] == get_response['maxDownstreamRate']
		assert post_response['maxUpstreamBurstRate'] == get_response['maxUpstreamBurstRate']
		assert post_response['maxUpstreamRate'] == get_response['maxUpstreamRate']
		assert post_response['minDuration'] == get_response['minDuration']
		assert post_response['name'] == get_response['name']
		assert post_response['packetDelayBudget'] == get_response['packetDelayBudget']
		assert post_response['packetErrorLossRate'] == get_response['packetErrorLossRate']
		assert post_response['priority'] == get_response['priority']
		assert post_response['status'] == get_response['status']
		assert post_response['targetMinDownstreamRate'] == get_response['targetMinDownstreamRate']
		assert post_response['targetMinUpstreamRate'] == get_response['targetMinUpstreamRate']

		def test_get_constraints():
		# Replace with actual qos_profile_id you want to test with
		qos_profile_id = "contraints"
		start_timestamp = 1726063284.25332
		duration = 86400

		# Send GET request to fetch constraints
		response = requests.get(f'{BASE_URL}/profiles/constraints', params={
		"qos_profile_id": qos_profile_id,
		"start_timestamp": start_timestamp,
		"duration": duration
		})

		# Convert response to JSON and add error checking
		if response.status_code == 200:
		constraints = response.json()
		LOGGER.debug(f"Constraints retrieved: {constraints}")
		# def test_delete_profile_by_id():
		# qos_profile_id = '0898e7e8-ef15-4522-8e93-623e31c92efa'
		# response = requests.delete(f'{BASE_URL}/profiles/{qos_profile_id}')

		# Additional assertions for constraints
		assert len(constraints) > 0, "Expected at least one constraint"
		first_constraint = constraints[0]
		assert "constraint_type" in first_constraint, "Constraint type missing in response"
		else:
		LOGGER.error(f"Failed to fetch constraints: Status code {response.status_code}")