+class TestNsLcmOpTopicWithMock(unittest.TestCase):
+ def setUp(self):
+ self.db = Mock(dbbase.DbBase())
+ self.fs = Mock(FsBase())
+ self.fs.get_params.return_value = {"./fake/folder"}
+ self.fs.file_open = mock_open()
+ self.msg = Mock(MsgBase())
+ # create class
+ self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
+
+ def test_get_vnfd_from_vnf_member_revision(self):
+ test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
+ test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
+ self.db.get_one.side_effect = [test_vnfr, test_vnfd]
+ vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
+ self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
+ self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds', "Incorrect second DB lookup")
+
+ def test_get_vnfd_from_vnf_member_no_revision(self):
+ test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
+ test_vnfr['revision'] = 3
+ test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
+ self.db.get_one.side_effect = [test_vnfr, test_vnfd]
+ vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
+ self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
+ self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds_revisions', "Incorrect second DB lookup")
+
+ @contextmanager
+ def assertNotRaises(self, exception_type):
+ try:
+ yield None
+ except exception_type:
+ raise self.failureException("{} raised".format(exception_type.__name__))
+
+ def test_check_ns_update_operation(self):
+ self.db = DbMemory()
+ self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
+ session = {}
+
+ with self.subTest(i=1, t="VNF instance does not belong to NS"):
+ test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
+ test_vnfr[0]["revision"] = 2
+ test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
+ test_nsr[0]["constituent-vnfr-ref"][
+ 0
+ ] = "99d90b0c-faff-4b9f-bccd-017f33985984"
+ self.db.create_list("vnfrs", test_vnfr)
+ self.db.create_list("nsrs", test_nsr)
+ nsrs = self.db.get_list("nsrs")[0]
+ indata = {
+ "updateType": "CHANGE_VNFPKG",
+ "changeVnfPackageData": {
+ "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
+ },
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ }
+ with self.assertRaises(EngineException) as expected_exception:
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+ self.assertEqual(
+ str(expected_exception.exception),
+ "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984"
+ " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85",
+ )
+
+ with self.subTest(i=2, t="Ns update request validated with no exception"):
+ test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
+ test_vnfr[0]["revision"] = 2
+ test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
+ self.db.create_list("vnfrs", test_vnfr)
+ self.db.create_list("nsrs", test_nsr)
+ nsrs = self.db.get_list("nsrs")[1]
+ indata = {
+ "updateType": "CHANGE_VNFPKG",
+ "changeVnfPackageData": {
+ "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
+ },
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ }
+ with self.assertNotRaises(EngineException):
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+
+ with self.subTest(
+ i=3, t="Ns update request rejected because of too small timeout"
+ ):
+ indata = {
+ "updateType": "CHANGE_VNFPKG",
+ "changeVnfPackageData": {
+ "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
+ },
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ "timeout_ns_update": 50,
+ }
+ with self.assertRaises(EngineException) as expected_exception:
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+ self.assertEqual(
+ str(expected_exception.exception),
+ "Error in validating ns-update request: 50 second is not enough "
+ "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
+ )
+
+ with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"):
+ test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
+ test_vnfr[0]["revision"] = 2
+ test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
+ self.db.create_list("vnfrs", test_vnfr)
+ self.db.create_list("nsrs", test_nsr)
+ nsrs = self.db.get_list("nsrs")[2]
+ indata = {
+ "updateType": "CHANGE_VNFPKG",
+ "changeVnfPackageData": {
+ "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
+ },
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ }
+ with self.assertRaises(EngineException) as expected_exception:
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+ self.assertEqual(
+ str(expected_exception.exception),
+ "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not "
+ "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of "
+ "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
+ )
+
+ with self.subTest(i=5, t="Ns update REMOVE_VNF request validated with no exception"):
+ test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
+ test_vnfr[0]["revision"] = 2
+ test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
+ self.db.create_list("vnfrs", test_vnfr)
+ self.db.create_list("nsrs", test_nsr)
+ nsrs = self.db.get_list("nsrs")[1]
+ indata = {
+ "updateType": "REMOVE_VNF",
+ "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ }
+ with self.assertNotRaises(EngineException):
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+
+class TestNsrTopic(unittest.TestCase):