Fixing Bug 2103 and Bug 2134 in NBI (2b5e123a) · Commits · osm / NBI

osm_nbi/base_topic.py

+88 −1

Original line number	Diff line number	Diff line
		@@ -17,7 +17,7 @@ import logging
		from uuid import uuid4
		from http import HTTPStatus
		from time import time
		from osm_common.dbbase import deep_update_rfc7396
		from osm_common.dbbase import deep_update_rfc7396, DbException
		from osm_nbi.validation import validate_input, ValidationError, is_valid_uuid
		from yaml import safe_load, YAMLError

		@@ -29,6 +29,20 @@ class EngineException(Exception):
		self.http_code = http_code
		super(Exception, self).__init__(message)

		class NBIBadArgumentsException(Exception):
		"""
		Bad argument values exception
		"""

		def __init__(self, message: str = "", bad_args: list = None):
		Exception.__init__(self, message)
		self.message = message
		self.bad_args = bad_args

		def __str__(self):
		return "{}, Bad arguments: {}".format(
		self.message, self.bad_args
		)

		def deep_get(target_dict, key_list):
		"""
		@@ -45,6 +59,79 @@ def deep_get(target_dict, key_list):
		return target_dict


		def detect_descriptor_usage(
		descriptor: dict, db_collection: str, db: object
		) -> bool:
		"""Detect the descriptor usage state.

		Args:
		descriptor (dict): VNF or NS Descriptor as dictionary
		db_collection (str): collection name which is looked for in DB
		db (object): name of db object

		Returns:
		True if descriptor is in use else None

		"""
		try:
		if not descriptor:
		raise NBIBadArgumentsException(
		"Argument is mandatory and can not be empty", "descriptor"
		)

		if not db:
		raise NBIBadArgumentsException("A valid DB object should be provided", "db")

		search_dict = {
		"vnfds": ("vnfrs", "vnfd-id"),
		"nsds": ("nsrs", "nsd-id"),
		}

		if db_collection not in search_dict:
		raise NBIBadArgumentsException("db_collection should be equal to vnfds or nsds", "db_collection")

		record_list = db.get_list(
		search_dict[db_collection][0],
		{search_dict[db_collection][1]: descriptor["_id"]},
		)

		if record_list:
		return True

		except (DbException, KeyError, NBIBadArgumentsException) as error:
		raise EngineException(f"Error occured while detecting the descriptor usage: {error}")


		def update_descriptor_usage_state(
		descriptor: dict, db_collection: str, db: object
		) -> None:
		"""Updates the descriptor usage state.

		Args:
		descriptor (dict): VNF or NS Descriptor as dictionary
		db_collection (str): collection name which is looked for in DB
		db (object): name of db object

		Returns:
		None

		"""
		try:
		descriptor_update = {
		"_admin.usageState": "NOT_IN_USE",
		}

		if detect_descriptor_usage(descriptor, db_collection, db):
		descriptor_update = {
		"_admin.usageState": "IN_USE",
		}

		db.set_one(db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update)

		except (DbException, KeyError, NBIBadArgumentsException) as error:
		raise EngineException(f"Error occured while updating the descriptor usage state: {error}")


		def get_iterable(input_var):
		"""
		Returns an iterable, in case input_var is None it just returns an empty tuple

osm_nbi/descriptor_topics.py

+51 −21

Original line number	Diff line number	Diff line
		@@ -37,7 +37,12 @@ from osm_nbi.validation import (
		validate_input,
		vnfpkgop_new_schema,
		)
		from osm_nbi.base_topic import BaseTopic, EngineException, get_iterable
		from osm_nbi.base_topic import (
		BaseTopic,
		EngineException,
		get_iterable,
		detect_descriptor_usage,
		)
		from osm_im import etsi_nfv_vnfd, etsi_nfv_nsd
		from osm_im.nst import nst as nst_im
		from pyangbind.lib.serialise import pybindJSONDecoder
		@@ -456,9 +461,11 @@ class DescriptorTopic(BaseTopic):
		if revision > 1:
		try:
		self._validate_descriptor_changes(
		_id,
		descriptor_file_name,
		current_revision_path,
		proposed_revision_path)
		proposed_revision_path,
		)
		except Exception as e:
		shutil.rmtree(self.fs.path + current_revision_path, ignore_errors=True)
		shutil.rmtree(self.fs.path + proposed_revision_path, ignore_errors=True)
		@@ -694,11 +701,13 @@ class DescriptorTopic(BaseTopic):

		return indata

		def _validate_descriptor_changes(self,
		def _validate_descriptor_changes(
		self,
		descriptor_id,
		descriptor_file_name,
		old_descriptor_directory,
		new_descriptor_directory):
		# Todo: compare changes and throw a meaningful exception for the user to understand
		new_descriptor_directory
		):
		# Example:
		# raise EngineException(
		# "Error in validating new descriptor: <NODE> cannot be modified",
		@@ -1290,6 +1299,7 @@ class VnfdTopic(DescriptorTopic):

		def _validate_descriptor_changes(
		self,
		descriptor_id: str,
		descriptor_file_name: str,
		old_descriptor_directory: str,
		new_descriptor_directory: str,
		@@ -1298,7 +1308,7 @@ class VnfdTopic(DescriptorTopic):

		Args:
		old_descriptor_directory (str): Directory of descriptor which is in-use
		new_descriptor_directory (str): Directory of directory which is proposed to update (new revision)
		new_descriptor_directory (str): Directory of descriptor which is proposed to update (new revision)

		Returns:
		None
		@@ -1307,27 +1317,37 @@ class VnfdTopic(DescriptorTopic):
		EngineException: In case of error when there are unallowed changes
		"""
		try:
		# If VNFD does not exist in DB or it is not in use by any NS,
		# validation is not required.
		vnfd = self.db.get_one("vnfds", {"_id": descriptor_id})
		if not vnfd or not detect_descriptor_usage(vnfd, "vnfds", self.db):
		return

		# Get the old and new descriptor contents in order to compare them.
		with self.fs.file_open(
		(old_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
		) as old_descriptor_file:

		with self.fs.file_open(
		(new_descriptor_directory, descriptor_file_name), "r"
		(new_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
		) as new_descriptor_file:
		old_content = yaml.load(
		old_descriptor_file.read(), Loader=yaml.SafeLoader
		)
		new_content = yaml.load(
		new_descriptor_file.read(), Loader=yaml.SafeLoader
		)

		old_content = yaml.safe_load(old_descriptor_file.read())
		new_content = yaml.safe_load(new_descriptor_file.read())

		# If software version has changed, we do not need to validate
		# the differences anymore.
		if old_content and new_content:
		if self.find_software_version(
		old_content
		) != self.find_software_version(new_content):
		return

		disallowed_change = DeepDiff(
		self.remove_modifiable_items(old_content),
		self.remove_modifiable_items(new_content),
		)

		if disallowed_change:
		changed_nodes = functools.reduce(
		lambda a, b: a + " , " + b,
		@@ -1338,6 +1358,7 @@ class VnfdTopic(DescriptorTopic):
		).keys()
		],
		)

		raise EngineException(
		f"Error in validating new descriptor: {changed_nodes} cannot be modified, "
		"there are disallowed changes in the vnf descriptor.",
		@@ -1673,6 +1694,7 @@ class NsdTopic(DescriptorTopic):

		def _validate_descriptor_changes(
		self,
		descriptor_id: str,
		descriptor_file_name: str,
		old_descriptor_directory: str,
		new_descriptor_directory: str,
		@@ -1681,7 +1703,7 @@ class NsdTopic(DescriptorTopic):

		Args:
		old_descriptor_directory: Directory of descriptor which is in-use
		new_descriptor_directory: Directory of directory which is proposed to update (new revision)
		new_descriptor_directory: Directory of descriptor which is proposed to update (new revision)

		Returns:
		None
		@@ -1691,23 +1713,30 @@ class NsdTopic(DescriptorTopic):
		"""

		try:
		# If NSD does not exist in DB, or it is not in use by any NS,
		# validation is not required.
		nsd = self.db.get_one("nsds", {"_id": descriptor_id}, fail_on_empty=False)
		if not nsd or not detect_descriptor_usage(nsd, "nsds", self.db):
		return

		# Get the old and new descriptor contents in order to compare them.
		with self.fs.file_open(
		(old_descriptor_directory, descriptor_file_name), "r"
		(old_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
		) as old_descriptor_file:

		with self.fs.file_open(
		(new_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
		) as new_descriptor_file:
		old_content = yaml.load(
		old_descriptor_file.read(), Loader=yaml.SafeLoader
		)
		new_content = yaml.load(
		new_descriptor_file.read(), Loader=yaml.SafeLoader
		)

		old_content = yaml.safe_load(old_descriptor_file.read())
		new_content = yaml.safe_load(new_descriptor_file.read())

		if old_content and new_content:
		disallowed_change = DeepDiff(
		self.remove_modifiable_items(old_content),
		self.remove_modifiable_items(new_content),
		)

		if disallowed_change:
		changed_nodes = functools.reduce(
		lambda a, b: a + ", " + b,
		@@ -1718,6 +1747,7 @@ class NsdTopic(DescriptorTopic):
		).keys()
		],
		)

		raise EngineException(
		f"Error in validating new descriptor: {changed_nodes} cannot be modified, "
		"there are disallowed changes in the ns descriptor. ",

osm_nbi/instance_topics.py

+5 −0

Original line number	Diff line number	Diff line
		@@ -38,6 +38,7 @@ from osm_nbi.base_topic import (
		get_iterable,
		deep_get,
		increment_ip_mac,
		update_descriptor_usage_state,
		)
		from yaml import safe_dump
		from osm_common.dbbase import DbException
		@@ -378,9 +379,13 @@ class NsrTopic(BaseTopic):
		)
		self._add_vnfr_to_db(vnfr_descriptor, rollback, session)
		nsr_descriptor["constituent-vnfr-ref"].append(vnfr_descriptor["id"])
		step = "Updating VNFD usageState"
		update_descriptor_usage_state(vnfd, "vnfds", self.db)

		step = "creating nsr at database"
		self._add_nsr_to_db(nsr_descriptor, rollback, session)
		step = "Updating NSD usageState"
		update_descriptor_usage_state(nsd, "nsds", self.db)

		step = "creating nsr temporal folder"
		self.fs.mkdir(nsr_id)

osm_nbi/tests/test_base_topic.py

+119 −9

Original line number	Diff line number	Diff line
		@@ -17,12 +17,23 @@
		__author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
		__date__ = "2020-06-17"

		from copy import deepcopy
		import unittest
		from unittest import TestCase
		from unittest.mock import patch, Mock
		from osm_nbi.base_topic import (
		BaseTopic,
		EngineException,
		NBIBadArgumentsException,
		detect_descriptor_usage,
		update_descriptor_usage_state,
		)
		from osm_common import dbbase
		from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text
		import yaml

		# from unittest.mock import Mock
		# from osm_common import dbbase, fsbase, msgbase
		from osm_nbi.base_topic import BaseTopic, EngineException
		db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
		db_nsd_content = yaml.safe_load(db_nsds_text)[0]


		class Test_BaseTopic(TestCase):
		@@ -35,14 +46,9 @@ class Test_BaseTopic(TestCase):
		pass

		def setUp(self):
		pass
		# self.db = Mock(dbbase.DbBase())
		# self.fs = Mock(fsbase.FsBase())
		# self.msg = Mock(msgbase.MsgBase())
		# self.auth = Mock(authconn.Authconn(None, None, None))
		self.db = Mock(dbbase.DbBase())

		def test_update_input_with_kwargs(self):

		test_set = (
		# (descriptor content, kwargs, expected descriptor (None=fails), message)
		(
		@@ -123,6 +129,110 @@ class Test_BaseTopic(TestCase):
		BaseTopic._update_input_with_kwargs(desc, kwargs)
		self.assertEqual(desc, expected, message)

		def test_detect_descriptor_usage_empty_descriptor(self):
		descriptor = {}
		db_collection = "vnfds"
		with self.assertRaises(EngineException) as error:
		detect_descriptor_usage(descriptor, db_collection, self.db)
		self.assertIn(
		"Argument is mandatory and can not be empty, Bad arguments: descriptor",
		error,
		"Error message is wrong.",
		)
		self.db.get_list.assert_not_called()

		def test_detect_descriptor_usage_empty_db_argument(self):
		descriptor = deepcopy(db_vnfd_content)
		db_collection = "vnfds"
		db = None
		with self.assertRaises(EngineException) as error:
		detect_descriptor_usage(descriptor, db_collection, db)
		self.assertIn(
		"A valid DB object should be provided, Bad arguments: db",
		error,
		"Error message is wrong.",
		)
		self.db.get_list.assert_not_called()

		def test_detect_descriptor_usage_which_is_in_use(self):
		descriptor = deepcopy(db_vnfd_content)
		db_collection = "vnfds"
		self.db.get_list.side_effect = [deepcopy(db_vnfd_content)]
		expected = True
		result = detect_descriptor_usage(descriptor, db_collection, self.db)
		self.assertEqual(result, expected, "wrong result")
		self.db.get_list.assert_called_once_with(
		"vnfrs", {"vnfd-id": descriptor["_id"]}
		)

		def test_detect_descriptor_usage_which_is_not_in_use(self):
		descriptor = deepcopy(db_nsd_content)
		self.db.get_list.return_value = []
		db_collection = "nsds"
		expected = None
		result = detect_descriptor_usage(descriptor, db_collection, self.db)
		self.assertEqual(result, expected, "wrong result")
		self.db.get_list.assert_called_once_with("nsrs", {"nsd-id": descriptor["_id"]})

		def test_detect_descriptor_usage_wrong_desc_format(self):
		descriptor = deepcopy(db_nsd_content)
		descriptor.pop("_id")
		db_collection = "nsds"
		with self.assertRaises(EngineException) as error:
		detect_descriptor_usage(descriptor, db_collection, self.db)
		self.assertIn("KeyError", error, "wrong error type")
		self.db.get_list.assert_not_called()

		def test_detect_descriptor_usage_wrong_db_collection(self):
		descriptor = deepcopy(db_vnfd_content)
		descriptor.pop("_id")
		db_collection = "vnf"
		with self.assertRaises(EngineException) as error:
		detect_descriptor_usage(descriptor, db_collection, self.db)
		self.assertIn(
		"db_collection should be equal to vnfds or nsds, db_collection",
		error,
		"wrong error type",
		)

		self.db.get_list.assert_not_called()

		@patch("osm_nbi.base_topic.detect_descriptor_usage")
		def test_update_descriptor_usage_state_to_in_use(self, mock_descriptor_usage):
		db_collection = "vnfds"
		descriptor = deepcopy(db_vnfd_content)
		mock_descriptor_usage.return_value = True
		descriptor_update = {"_admin.usageState": "IN_USE"}
		update_descriptor_usage_state(descriptor, db_collection, self.db)
		self.db.set_one.assert_called_once_with(
		db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
		)

		@patch("osm_nbi.base_topic.detect_descriptor_usage")
		def test_update_descriptor_usage_state_to_not_in_use(self, mock_descriptor_usage):
		db_collection = "nsds"
		descriptor = deepcopy(db_nsd_content)
		mock_descriptor_usage.return_value = False
		descriptor_update = {"_admin.usageState": "NOT_IN_USE"}
		update_descriptor_usage_state(descriptor, db_collection, self.db)
		self.db.set_one.assert_called_once_with(
		db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
		)

		@patch("osm_nbi.base_topic.detect_descriptor_usage")
		def test_update_descriptor_usage_state_db_exception(self, mock_descriptor_usage):
		db_collection = "nsd"
		descriptor = deepcopy(db_nsd_content)
		mock_descriptor_usage.side_effect = NBIBadArgumentsException
		with self.assertRaises(EngineException) as error:
		update_descriptor_usage_state(descriptor, db_collection, self.db)
		self.assertIn(
		"db_collection should be equal to vnfds or nsds, db_collection",
		error,
		"wrong error type",
		)
		self.db.set_one.assert_not_called()


		if __name__ == "__main__":
		unittest.main()

osm_nbi/tests/test_descriptor_topics.py

+157 −30

File changed.

Preview size limit exceeded, changes collapsed.