OSM Client component: (96d5ae7a) · Commits · TFS / controller

src/osm_client/service/OsmClientServiceServicerImpl.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -21,7 +21,7 @@ from common.proto.osm_client_pb2 import (
		)
		from common.proto.osm_client_pb2_grpc import OsmServiceServicer
		from osmclient import client
		from osmclient.common.exceptions import ClientException
		#from osmclient.common.exceptions import ClientException
		from osm_client.Config import (
		OSM_ADDRESS, OSM_PORT, OSM_USERNAME, OSM_PASSWORD, OSM_PROJECT, OSM_VERIFY_TLS
		)

src/tests/tools/mock_osm_nbi/ResourceOsmClient.py

+86 −80

Original line number	Diff line number	Diff line
		@@ -12,82 +12,84 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		import logging
		from flask import abort, jsonify, make_response, request
		import logging, uuid
		from typing import Dict
		from flask import jsonify, make_response, request
		from flask_restful import Resource
		from uuid import uuid4
		from hashlib import sha1

		LOG_LEVEL = logging.DEBUG

		LOG_LEVEL = logging.DEBUG
		logging.basicConfig(level=LOG_LEVEL, format="[%(asctime)s] %(levelname)s:%(name)s:%(message)s")
		logging.getLogger('werkzeug').setLevel(logging.INFO)
		LOGGER = logging.getLogger(__name__)

		logging.getLogger('werkzeug').setLevel(logging.INFO)

		OSM_NSI = []
		def generate_uuid(raw_string : str) -> str:
		''' Generate a str(UUID) from the SHA-1 hash of a string. '''
		_hash = sha1(raw_string.encode('utf-8')).digest()
		return str(uuid.UUID(bytes=_hash[:16]))

		OSM_VIM_DB = [
		{
		"_id": "account1",
		"name": "account1",
		"vim_type": "account1",
		"tenant": "account1",
		"vim_url": "http://account1.local"
		OSM_VIM : Dict[str, Dict] = {
		generate_uuid('account1') : {
		'_id' : generate_uuid('account1'),
		'name' : 'account1',
		'vim_type' : 'account1',
		'tenant' : 'account1',
		'vim_url' : 'http://account1.local',
		}
		}

		OSM_NST : Dict[str, Dict] = {
		generate_uuid('nst1') : {
		'_id' : generate_uuid('nst1'),
		'name' : 'nst1',
		'description': 'Nst_Mock',
		'vendor' : 'ExampleVendor',
		'version' : '1.0',
		},
		generate_uuid('nst2') : {
		'_id' : generate_uuid('nst2'),
		'name' : 'nst2',
		'description': 'Nst_Mock',
		'vendor' : 'AnotherVendor',
		'version' : '2.1',
		},
		}
		]

		OSM_NSI : Dict[str, Dict] = dict()


		class OsmNST(Resource):
		def get(self):
		LOGGER.info("get NST request received")
		mock_nsts = [
		{
		"_id": "nst1",
		"name": "nst1",
		"description": "Nst_Mock",
		"vendor": "ExampleVendor",
		"version": "1.0"
		},
		{
		"_id": "nst2",
		"name": "nst2",
		"description": "Nst_Mock",
		"vendor": "AnotherVendor",
		"version": "2.1"
		}
		]
		return make_response(jsonify(mock_nsts), 200)
		LOGGER.info('Get NST request received')
		return make_response(jsonify(list(OSM_NST.values())), 200)

		class OsmNBI(Resource):
		def get(self):
		LOGGER.info("get NBI request received")

		LOGGER.info(str(OSM_NSI))

		return [nsi["id"] for nsi in OSM_NSI]
		LOGGER.info('Get NBI request received')
		return make_response(jsonify(list(OSM_NSI.values())), 200)

		def post(self):
		LOGGER.info("post request received")

		LOGGER.info('Post request received')
		LOGGER.info(str(request))

		payload = request.get_json(silent=True) or {}
		qs = request.args

		nst_id = payload.get("nstId") or qs.get("nstId")
		name = payload.get("nsiName") or qs.get("nsiName")
		desc = payload.get("nsiDescription") or qs.get("nsiDescription", "")
		nst_id = payload.get('nstId' ) or qs.get('nstId' )
		name = payload.get('nsiName' ) or qs.get('nsiName' )
		desc = payload.get('nsiDescription') or qs.get('nsiDescription', '')

		new_nsi_id = str(generate_uuid(name))
		new_nsi = {
		"id": str(uuid4()),
		"name": name,
		"nstId": nst_id,
		"description": desc,
		"operationalStatus": "CREATED"
		'id' : new_nsi_id,
		'name' : name,
		'nstId' : nst_id,
		'description' : desc,
		'operationalStatus': 'CREATED'
		}

		OSM_NSI.append(new_nsi)

		OSM_NSI[new_nsi_id] = new_nsi
		return make_response(jsonify(new_nsi), 201)


		@@ -99,27 +101,28 @@ class VimAccounts(Resource):
		• POST → Create a VIM Account
		"""
		def get(self):
		return jsonify(OSM_VIM_DB)
		return make_response(jsonify(list(OSM_VIM.values())), 200)

		def post(self):
		payload = request.get_json(silent=True) or {}
		name = payload.get("name")
		vim_type = payload.get("vim_type")
		name = payload.get('name')
		vim_type = payload.get('vim_type')

		if not name or not vim_type:
		return {"error": "name and vim_type are required"}, 400
		error = {'error': 'name and vim_type are required'}
		return make_response(jsonify(error), 400)

		new_vim_id = str(generate_uuid(name))
		new_vim = {
		"_id": str(uuid4()),
		"name": name,
		"vim_type": vim_type,
		"tenant": payload.get("tenant", "admin"),
		"vim_url": payload.get("vim_url", "http://mock.local")
		'_id' : new_vim_id,
		'name' : name,
		'vim_type': vim_type,
		'tenant' : payload.get('tenant', 'admin'),
		'vim_url' : payload.get('vim_url', 'http://mock.local')
		}

		# Store new_vim in local DDBB
		OSM_VIM_DB.append(new_vim)
		return jsonify(new_vim), 201
		OSM_VIM[new_vim_id] = new_vim
		return make_response(jsonify(new_vim), 201)


		class VimAccountItem(Resource):
		@@ -135,7 +138,7 @@ class VimAccountItem(Resource):
		def _find(account_id):
		"""Search VIM account"""
		return next(
		(acc for acc in OSM_VIM_DB if acc["_id"] == account_id),
		(acc for acc in OSM_VIM if acc["_id"] == account_id),
		None
		)

		@@ -143,33 +146,36 @@ class VimAccountItem(Resource):
		# GET /vim_accounts/<id>
		# ------------------------
		def get(self, account_id):
		account = self._find(account_id)
		if not account:
		return {"error": "not found"}, 404
		return jsonify(account)
		vim_account = self._find(account_id)
		if not vim_account:
		error = {'error': 'not found'}
		return make_response(jsonify(error), 404)
		return make_response(jsonify(vim_account), 200)

		# ------------------------
		# PUT /vim_accounts/<id>
		# ------------------------
		def put(self, account_id):
		account = self._find(account_id)
		if not account:
		return {"error": "not found"}, 404
		vim_account = self._find(account_id)
		if not vim_account:
		error = {'error': 'not found'}
		return make_response(jsonify(error), 404)

		payload = request.get_json(silent=True) or {}
		for field in ("name", "vim_type", "tenant", "vim_url"):
		for field in ('name', 'vim_type', 'tenant', 'vim_url'):
		if field in payload:
		account[field] = payload[field]
		vim_account[field] = payload[field]

		return jsonify(account)
		return jsonify(vim_account)

		# ------------------------
		# DELETE /vim_accounts/<id>
		# ------------------------
		def delete(self, account_id):
		account = self._find(account_id)
		if not account:
		return {"error": "not found"}, 404
		vim_account = self._find(account_id)
		if not vim_account:
		error = {'error': 'not found'}
		return make_response(jsonify(error), 404)

		OSM_VIM_DB.remove(account)
		return "", 204
		del OSM_VIM[account_id]
		return make_response(jsonify({}), 204)