osm_nbi/tests/test_base_topic.py - osm/NBI - Gitiles

 #! /usr/bin/python3
 # -*- coding: utf-8 -*-

 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #    http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 # implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.

 __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

 db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
 db_nsd_content = yaml.safe_load(db_nsds_text)[0]


 class Test_BaseTopic(TestCase):
     @classmethod
     def setUpClass(cls):
         cls.test_name = "test-base-topic"

     @classmethod
     def tearDownClass(cls):
         pass

     def setUp(self):
         self.db = Mock(dbbase.DbBase())

     def test_update_input_with_kwargs(self):
         test_set = (
             # (descriptor content, kwargs, expected descriptor (None=fails), message)
             (
                 {"a": {"none": None}},
                 {"a.b.num": "v"},
                 {"a": {"none": None, "b": {"num": "v"}}},
                 "create dict",
             ),
             (
                 {"a": {"none": None}},
                 {"a.none.num": "v"},
                 {"a": {"none": {"num": "v"}}},
                 "create dict over none",
             ),
             (
                 {"a": {"b": {"num": 4}}},
                 {"a.b.num": "v"},
                 {"a": {"b": {"num": "v"}}},
                 "replace_number",
             ),
             (
                 {"a": {"b": {"num": 4}}},
                 {"a.b.num.c.d": "v"},
                 {"a": {"b": {"num": {"c": {"d": "v"}}}}},
                 "create dict over number",
             ),
             (
                 {"a": {"b": {"num": 4}}},
                 {"a.b": "v"},
                 {"a": {"b": "v"}},
                 "replace dict with a string",
             ),
             (
                 {"a": {"b": {"num": 4}}},
                 {"a.b": None},
                 {"a": {}},
                 "replace dict with None",
             ),
             (
                 {"a": [{"b": {"num": 4}}]},
                 {"a.b.num": "v"},
                 None,
                 "create dict over list should fail",
             ),
             (
                 {"a": [{"b": {"num": 4}}]},
                 {"a.0.b.num": "v"},
                 {"a": [{"b": {"num": "v"}}]},
                 "set list",
             ),
             (
                 {"a": [{"b": {"num": 4}}]},
                 {"a.3.b.num": "v"},
                 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
                 "expand list",
             ),
             ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
             (
                 {"a": [[4]]},
                 {"a.0.2": "v"},
                 {"a": [[4, None, "v"]]},
                 "expand nested list",
             ),
             (
                 {"a": [[4]]},
                 {"a.2.2": "v"},
                 {"a": [[4], None, {"2": "v"}]},
                 "expand list and add number key",
             ),
             ({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"),
         )
         for desc, kwargs, expected, message in test_set:
             if expected is None:
                 self.assertRaises(
                     EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs
                 )
             else:
                 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()
	#! /usr/bin/python3
	# -- coding: utf-8 --

	# Licensed under the Apache License, Version 2.0 (the "License");
	# you may not use this file except in compliance with the License.
	# You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS,
	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
	# implied.
	# See the License for the specific language governing permissions and
	# limitations under the License.

	__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

	db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
	db_nsd_content = yaml.safe_load(db_nsds_text)[0]


	class Test_BaseTopic(TestCase):
	@classmethod
	def setUpClass(cls):
	cls.test_name = "test-base-topic"

	@classmethod
	def tearDownClass(cls):
	pass

	def setUp(self):
	self.db = Mock(dbbase.DbBase())

	def test_update_input_with_kwargs(self):
	test_set = (
	# (descriptor content, kwargs, expected descriptor (None=fails), message)
	(
	{"a": {"none": None}},
	{"a.b.num": "v"},
	{"a": {"none": None, "b": {"num": "v"}}},
	"create dict",
	),
	(
	{"a": {"none": None}},
	{"a.none.num": "v"},
	{"a": {"none": {"num": "v"}}},
	"create dict over none",
	),
	(
	{"a": {"b": {"num": 4}}},
	{"a.b.num": "v"},
	{"a": {"b": {"num": "v"}}},
	"replace_number",
	),
	(
	{"a": {"b": {"num": 4}}},
	{"a.b.num.c.d": "v"},
	{"a": {"b": {"num": {"c": {"d": "v"}}}}},
	"create dict over number",
	),
	(
	{"a": {"b": {"num": 4}}},
	{"a.b": "v"},
	{"a": {"b": "v"}},
	"replace dict with a string",
	),
	(
	{"a": {"b": {"num": 4}}},
	{"a.b": None},
	{"a": {}},
	"replace dict with None",
	),
	(
	{"a": [{"b": {"num": 4}}]},
	{"a.b.num": "v"},
	None,
	"create dict over list should fail",
	),
	(
	{"a": [{"b": {"num": 4}}]},
	{"a.0.b.num": "v"},
	{"a": [{"b": {"num": "v"}}]},
	"set list",
	),
	(
	{"a": [{"b": {"num": 4}}]},
	{"a.3.b.num": "v"},
	{"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
	"expand list",
	),
	({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
	(
	{"a": [[4]]},
	{"a.0.2": "v"},
	{"a": [[4, None, "v"]]},
	"expand nested list",
	),
	(
	{"a": [[4]]},
	{"a.2.2": "v"},
	{"a": [[4], None, {"2": "v"}]},
	"expand list and add number key",
	),
	({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"),
	)
	for desc, kwargs, expected, message in test_set:
	if expected is None:
	self.assertRaises(
	EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs
	)
	else:
	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()