+ 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()
+