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
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):
"""
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
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
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)
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",
def _validate_descriptor_changes(
self,
+ descriptor_id: str,
descriptor_file_name: str,
old_descriptor_directory: str,
new_descriptor_directory: str,
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
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,
).keys()
],
)
+
raise EngineException(
f"Error in validating new descriptor: {changed_nodes} cannot be modified, "
"there are disallowed changes in the vnf descriptor.",
def _validate_descriptor_changes(
self,
+ descriptor_id: str,
descriptor_file_name: str,
old_descriptor_directory: str,
new_descriptor_directory: str,
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
"""
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,
).keys()
],
)
+
raise EngineException(
f"Error in validating new descriptor: {changed_nodes} cannot be modified, "
"there are disallowed changes in the ns descriptor. ",
get_iterable,
deep_get,
increment_ip_mac,
+ update_descriptor_usage_state,
)
from yaml import safe_dump
from osm_common.dbbase import DbException
)
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)
__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):
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)
(
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()
import yaml
test_name = "test-user"
-db_vnfd_content = yaml.load(db_vnfds_text, Loader=yaml.Loader)[0]
-db_nsd_content = yaml.load(db_nsds_text, Loader=yaml.Loader)[0]
+db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
+db_nsd_content = yaml.safe_load(db_nsds_text)[0]
test_pid = db_vnfd_content["_admin"]["projects_read"][0]
fake_session = {
"username": test_name,
def test_new_vnfd_check_input_validation_vdu_int_cpd(
self, mock_rename, mock_shutil
):
+ """Testing input validation during new vnfd creation
+ for vdu internal connection point"""
did, test_vnfd = self.prepare_vnfd_creation()
test_vnfd = self.prepare_test_vnfd(test_vnfd)
ext_cpd = test_vnfd["ext-cpd"][1]
def test_new_vnfd_check_input_validation_duplicated_vld(
self, mock_rename, mock_shutil
):
+ """Testing input validation during new vnfd creation
+ for dublicated virtual link description"""
did, test_vnfd = self.prepare_vnfd_creation()
test_vnfd = self.prepare_test_vnfd(test_vnfd)
test_vnfd["int-virtual-link-desc"].insert(0, {"id": "internal"})
def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
self, mock_rename, mock_shutil
):
+ """Testing input validation during new vnfd creation
+ for vdu internal virtual link description"""
did, test_vnfd = self.prepare_vnfd_creation()
test_vnfd = self.prepare_test_vnfd(test_vnfd)
vdu = test_vnfd["vdu"][0]
def test_new_vnfd_check_input_validation_virtual_link_profile(
self, mock_rename, mock_shutil
):
+ """Testing input validation during new vnfd creation
+ for virtual link profile"""
did, test_vnfd = self.prepare_vnfd_creation()
test_vnfd = self.prepare_test_vnfd(test_vnfd)
fake_ivld_profile = {"id": "fake-profile-ref", "flavour": "fake-flavour"}
def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
self, mock_rename, mock_shutil
):
+ """Testing input validation during new vnfd creation
+ for scaling criteria without monitoring parameter"""
did, test_vnfd = self.prepare_vnfd_creation()
test_vnfd = self.prepare_test_vnfd(test_vnfd)
vdu = test_vnfd["vdu"][1]
def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
self, mock_rename, mock_shutil
):
+ """Testing input validation during new vnfd creation
+ for scaling criteria without day12 configuration"""
did, test_vnfd = self.prepare_vnfd_creation()
test_vnfd = self.prepare_test_vnfd(test_vnfd)
test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
def test_new_vnfd_check_input_validation_scaling_config_action(
self, mock_rename, mock_shutil
):
+ """Testing input validation during new vnfd creation
+ for scaling criteria wrong config primitive"""
did, test_vnfd = self.prepare_vnfd_creation()
test_vnfd = self.prepare_test_vnfd(test_vnfd)
df = test_vnfd["df"][0]
def test_new_vnfd_check_input_validation_everything_right(
self, mock_rename, mock_shutil
):
+ """Testing input validation during new vnfd creation
+ everything correct"""
did, test_vnfd = self.prepare_vnfd_creation()
test_vnfd = self.prepare_test_vnfd(test_vnfd)
test_vnfd["id"] = "fake-vnfd-id"
"Wrong DB NSD vnfd-id",
)
- self.db.del_list.call_args[0]
self.assertEqual(
self.db.del_list.call_args[0][0],
self.topic.topic + "_revisions",
old_vnfd, new_vnfd = self.create_desc_temp(db_vnfd_content)
return descriptor_name, old_vnfd, new_vnfd
- @patch("osm_nbi.descriptor_topics.yaml")
- def test_validate_vnfd_changes_day12_config_primitive_changed(self, mock_yaml):
+ @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+ @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+ def test_validate_vnfd_changes_day12_config_primitive_changed(
+ self, mock_safe_load, mock_detect_usage
+ ):
+ """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
+ did = old_vnfd["_id"]
new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
"day1-2"
][0]["config-primitive"][0]["name"] = "new_action"
- mock_yaml.load.side_effect = [old_vnfd, new_vnfd]
- with self.assertNotRaises(EngineException):
- self.topic._validate_descriptor_changes(descriptor_name, "/tmp/", "/tmp:1/")
+ mock_safe_load.side_effect = [old_vnfd, new_vnfd]
+ mock_detect_usage.return_value = True
+ self.db.get_one.return_value = old_vnfd
- @patch("osm_nbi.descriptor_topics.yaml")
- def test_validate_vnfd_changes_sw_version_changed(self, mock_yaml):
+ with self.assertNotRaises(EngineException):
+ self.topic._validate_descriptor_changes(
+ did, descriptor_name, "/tmp/", "/tmp:1/"
+ )
+ self.db.get_one.assert_called_once()
+ mock_detect_usage.assert_called_once()
+ self.assertEqual(mock_safe_load.call_count, 2)
+
+ @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+ @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+ def test_validate_vnfd_changes_sw_version_changed(
+ self, mock_safe_load, mock_detect_usage
+ ):
+ """Validating VNFD for updates, software version has changed"""
# old vnfd uses the default software version: 1.0
descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
+ did = old_vnfd["_id"]
new_vnfd["software-version"] = "1.3"
new_vnfd["sw-image-desc"][0]["name"] = "new-image"
- mock_yaml.load.side_effect = [old_vnfd, new_vnfd]
+ mock_safe_load.side_effect = [old_vnfd, new_vnfd]
+ mock_detect_usage.return_value = True
+ self.db.get_one.return_value = old_vnfd
+
with self.assertNotRaises(EngineException):
- self.topic._validate_descriptor_changes(descriptor_name, "/tmp/", "/tmp:1/")
+ self.topic._validate_descriptor_changes(
+ did, descriptor_name, "/tmp/", "/tmp:1/"
+ )
+ self.db.get_one.assert_called_once()
+ mock_detect_usage.assert_called_once()
+ self.assertEqual(mock_safe_load.call_count, 2)
- @patch("osm_nbi.descriptor_topics.yaml")
+ @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+ @patch("osm_nbi.descriptor_topics.yaml.safe_load")
def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
- self, mock_yaml
+ self, mock_safe_load, mock_detect_usage
):
+ """Validating VNFD for updates, software version has not
+ changed, mgmt-cp has changed."""
descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
new_vnfd["mgmt-cp"] = "new-mgmt-cp"
- mock_yaml.load.side_effect = [old_vnfd, new_vnfd]
- with self.assertRaises(EngineException) as e:
- self.topic._validate_descriptor_changes(descriptor_name, "/tmp/", "/tmp:1/")
+ mock_safe_load.side_effect = [old_vnfd, new_vnfd]
+ did = old_vnfd["_id"]
+ mock_detect_usage.return_value = True
+ self.db.get_one.return_value = old_vnfd
+
+ with self.assertRaises(
+ EngineException, msg="there are disallowed changes in the vnf descriptor"
+ ) as e:
+ self.topic._validate_descriptor_changes(
+ did, descriptor_name, "/tmp/", "/tmp:1/"
+ )
+
self.assertEqual(
e.exception.http_code,
HTTPStatus.UNPROCESSABLE_ENTITY,
norm(str(e.exception)),
"Wrong exception text",
)
+ self.db.get_one.assert_called_once()
+ mock_detect_usage.assert_called_once()
+ self.assertEqual(mock_safe_load.call_count, 2)
+
+ @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+ @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+ def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
+ self, mock_safe_load, mock_detect_usage
+ ):
+ """Validating VNFD for updates, software version has not
+ changed, mgmt-cp has changed, vnfd is not in use."""
+ descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
+ new_vnfd["mgmt-cp"] = "new-mgmt-cp"
+ mock_safe_load.side_effect = [old_vnfd, new_vnfd]
+ did = old_vnfd["_id"]
+ mock_detect_usage.return_value = None
+ self.db.get_one.return_value = old_vnfd
+
+ with self.assertNotRaises(EngineException):
+ self.topic._validate_descriptor_changes(
+ did, descriptor_name, "/tmp/", "/tmp:1/"
+ )
+
+ self.db.get_one.assert_called_once()
+ mock_detect_usage.assert_called_once()
+ mock_safe_load.assert_not_called()
def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self):
indata = deepcopy(db_vnfd_content)
old_vnfd = {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}
old_vnfd["_admin"]["revision"] = old_revision
- self.db.get_one.side_effect = [old_vnfd, None]
+
+ self.db.get_one.side_effect = [old_vnfd, old_vnfd, None]
self.topic.upload_content(fake_session, did, new_vnfd, {}, {"Content-Type": []})
db_args = self.db.replace.call_args[0]
old_nsd, new_nsd = self.create_desc_temp(db_nsd_content)
return descriptor_name, old_nsd, new_nsd
- @patch("osm_nbi.descriptor_topics.yaml")
- def test_validate_descriptor_ns_configuration_changed(self, mock_yaml):
- # NSD has changes in ns-configuration:config-primitive
+ @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+ @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+ def test_validate_descriptor_ns_configuration_changed(
+ self, mock_safe_load, mock_detect_usage
+ ):
+ """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
+ mock_safe_load.side_effect = [old_nsd, new_nsd]
+ mock_detect_usage.return_value = True
+ self.db.get_one.return_value = old_nsd
old_nsd.update(
{"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
)
new_nsd.update(
{"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
)
- mock_yaml.load.side_effect = [old_nsd, new_nsd]
- with self.assertNotRaises(EngineException):
- self.topic._validate_descriptor_changes(descriptor_name, "/tmp", "/tmp:1")
- @patch("osm_nbi.descriptor_topics.yaml")
- def test_validate_descriptor_nsd_name_changed(self, mock_yaml):
+ with self.assertNotRaises(EngineException):
+ self.topic._validate_descriptor_changes(
+ old_nsd["_id"], descriptor_name, "/tmp", "/tmp:1"
+ )
+ self.db.get_one.assert_called_once()
+ mock_detect_usage.assert_called_once()
+ self.assertEqual(mock_safe_load.call_count, 2)
+
+ @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+ @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+ def test_validate_descriptor_nsd_name_changed(
+ self, mock_safe_load, mock_detect_usage
+ ):
+ """Validating NSD, NSD name has changed."""
descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
+ did = old_nsd["_id"]
new_nsd["name"] = "nscharm-ns2"
- mock_yaml.load.side_effect = [old_nsd, new_nsd]
-
- with self.assertRaises(EngineException) as e:
- self.topic._validate_descriptor_changes(descriptor_name, "/tmp", "/tmp:1")
+ mock_safe_load.side_effect = [old_nsd, new_nsd]
+ mock_detect_usage.return_value = True
+ self.db.get_one.return_value = old_nsd
+ with self.assertRaises(
+ EngineException, msg="there are disallowed changes in the ns descriptor"
+ ) as e:
+ self.topic._validate_descriptor_changes(
+ did, descriptor_name, "/tmp", "/tmp:1"
+ )
self.assertEqual(
e.exception.http_code,
HTTPStatus.UNPROCESSABLE_ENTITY,
"Wrong exception text",
)
+ self.db.get_one.assert_called_once()
+ mock_detect_usage.assert_called_once()
+ self.assertEqual(mock_safe_load.call_count, 2)
+
+ @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+ @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+ def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
+ self, mock_safe_load, mock_detect_usage
+ ):
+ """Validating NSD, NSD name has changed, NSD is not in use."""
+ descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
+ did = old_nsd["_id"]
+ new_nsd["name"] = "nscharm-ns2"
+ mock_safe_load.side_effect = [old_nsd, new_nsd]
+ mock_detect_usage.return_value = None
+ self.db.get_one.return_value = old_nsd
+
+ with self.assertNotRaises(Exception):
+ self.topic._validate_descriptor_changes(
+ did, descriptor_name, "/tmp", "/tmp:1"
+ )
+
+ self.db.get_one.assert_called_once()
+ mock_detect_usage.assert_called_once()
+ mock_safe_load.assert_not_called()
+
def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
self,
):
E123,
E125,
E226,
- E241
+ E241,
+ E501
exclude =
.git,
__pycache__,
projects / osm / NBI.git / commitdiff