__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()
projects / osm / NBI.git / blobdiff