osm_nbi/tests/test_base_topic.py

   1 #! /usr/bin/python3
   2 # -*- coding: utf-8 -*-
   3
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
  18 __date__ = "2020-06-17"
  19
  20 from copy import deepcopy
  21 import unittest
  22 from unittest import TestCase
  23 from unittest.mock import patch, Mock
  24 from osm_nbi.base_topic import (
  25     BaseTopic,
  26     EngineException,
  27     NBIBadArgumentsException,
  28     detect_descriptor_usage,
  29     update_descriptor_usage_state,
  30 )
  31 from osm_common import dbbase
  32 from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text
  33 import yaml
  34
  35 db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
  36 db_nsd_content = yaml.safe_load(db_nsds_text)[0]
  37
  38
  39 class Test_BaseTopic(TestCase):
  40     @classmethod
  41     def setUpClass(cls):
  42         cls.test_name = "test-base-topic"
  43
  44     @classmethod
  45     def tearDownClass(cls):
  46         pass
  47
  48     def setUp(self):
  49         self.db = Mock(dbbase.DbBase())
  50
  51     def test_update_input_with_kwargs(self):
  52         test_set = (
  53             # (descriptor content, kwargs, expected descriptor (None=fails), message)
  54             (
  55                 {"a": {"none": None}},
  56                 {"a.b.num": "v"},
  57                 {"a": {"none": None, "b": {"num": "v"}}},
  58                 "create dict",
  59             ),
  60             (
  61                 {"a": {"none": None}},
  62                 {"a.none.num": "v"},
  63                 {"a": {"none": {"num": "v"}}},
  64                 "create dict over none",
  65             ),
  66             (
  67                 {"a": {"b": {"num": 4}}},
  68                 {"a.b.num": "v"},
  69                 {"a": {"b": {"num": "v"}}},
  70                 "replace_number",
  71             ),
  72             (
  73                 {"a": {"b": {"num": 4}}},
  74                 {"a.b.num.c.d": "v"},
  75                 {"a": {"b": {"num": {"c": {"d": "v"}}}}},
  76                 "create dict over number",
  77             ),
  78             (
  79                 {"a": {"b": {"num": 4}}},
  80                 {"a.b": "v"},
  81                 {"a": {"b": "v"}},
  82                 "replace dict with a string",
  83             ),
  84             (
  85                 {"a": {"b": {"num": 4}}},
  86                 {"a.b": None},
  87                 {"a": {}},
  88                 "replace dict with None",
  89             ),
  90             (
  91                 {"a": [{"b": {"num": 4}}]},
  92                 {"a.b.num": "v"},
  93                 None,
  94                 "create dict over list should fail",
  95             ),
  96             (
  97                 {"a": [{"b": {"num": 4}}]},
  98                 {"a.0.b.num": "v"},
  99                 {"a": [{"b": {"num": "v"}}]},
 100                 "set list",
 101             ),
 102             (
 103                 {"a": [{"b": {"num": 4}}]},
 104                 {"a.3.b.num": "v"},
 105                 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
 106                 "expand list",
 107             ),
 108             ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
 109             (
 110                 {"a": [[4]]},
 111                 {"a.0.2": "v"},
 112                 {"a": [[4, None, "v"]]},
 113                 "expand nested list",
 114             ),
 115             (
 116                 {"a": [[4]]},
 117                 {"a.2.2": "v"},
 118                 {"a": [[4], None, {"2": "v"}]},
 119                 "expand list and add number key",
 120             ),
 121             ({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"),
 122         )
 123         for desc, kwargs, expected, message in test_set:
 124             if expected is None:
 125                 self.assertRaises(
 126                     EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs
 127                 )
 128             else:
 129                 BaseTopic._update_input_with_kwargs(desc, kwargs)
 130                 self.assertEqual(desc, expected, message)
 131
 132     def test_detect_descriptor_usage_empty_descriptor(self):
 133         descriptor = {}
 134         db_collection = "vnfds"
 135         with self.assertRaises(EngineException) as error:
 136             detect_descriptor_usage(descriptor, db_collection, self.db)
 137             self.assertIn(
 138                 "Argument is mandatory and can not be empty, Bad arguments: descriptor",
 139                 error,
 140                 "Error message is wrong.",
 141             )
 142         self.db.get_list.assert_not_called()
 143
 144     def test_detect_descriptor_usage_empty_db_argument(self):
 145         descriptor = deepcopy(db_vnfd_content)
 146         db_collection = "vnfds"
 147         db = None
 148         with self.assertRaises(EngineException) as error:
 149             detect_descriptor_usage(descriptor, db_collection, db)
 150             self.assertIn(
 151                 "A valid DB object should be provided, Bad arguments: db",
 152                 error,
 153                 "Error message is wrong.",
 154             )
 155         self.db.get_list.assert_not_called()
 156
 157     def test_detect_descriptor_usage_which_is_in_use(self):
 158         descriptor = deepcopy(db_vnfd_content)
 159         db_collection = "vnfds"
 160         self.db.get_list.side_effect = [deepcopy(db_vnfd_content)]
 161         expected = True
 162         result = detect_descriptor_usage(descriptor, db_collection, self.db)
 163         self.assertEqual(result, expected, "wrong result")
 164         self.db.get_list.assert_called_once_with(
 165             "vnfrs", {"vnfd-id": descriptor["_id"]}
 166         )
 167
 168     def test_detect_descriptor_usage_which_is_not_in_use(self):
 169         descriptor = deepcopy(db_nsd_content)
 170         self.db.get_list.return_value = []
 171         db_collection = "nsds"
 172         expected = None
 173         result = detect_descriptor_usage(descriptor, db_collection, self.db)
 174         self.assertEqual(result, expected, "wrong result")
 175         self.db.get_list.assert_called_once_with("nsrs", {"nsd-id": descriptor["_id"]})
 176
 177     def test_detect_descriptor_usage_wrong_desc_format(self):
 178         descriptor = deepcopy(db_nsd_content)
 179         descriptor.pop("_id")
 180         db_collection = "nsds"
 181         with self.assertRaises(EngineException) as error:
 182             detect_descriptor_usage(descriptor, db_collection, self.db)
 183             self.assertIn("KeyError", error, "wrong error type")
 184         self.db.get_list.assert_not_called()
 185
 186     def test_detect_descriptor_usage_wrong_db_collection(self):
 187         descriptor = deepcopy(db_vnfd_content)
 188         descriptor.pop("_id")
 189         db_collection = "vnf"
 190         with self.assertRaises(EngineException) as error:
 191             detect_descriptor_usage(descriptor, db_collection, self.db)
 192             self.assertIn(
 193                 "db_collection should be equal to vnfds or nsds, db_collection",
 194                 error,
 195                 "wrong error type",
 196             )
 197
 198         self.db.get_list.assert_not_called()
 199
 200     @patch("osm_nbi.base_topic.detect_descriptor_usage")
 201     def test_update_descriptor_usage_state_to_in_use(self, mock_descriptor_usage):
 202         db_collection = "vnfds"
 203         descriptor = deepcopy(db_vnfd_content)
 204         mock_descriptor_usage.return_value = True
 205         descriptor_update = {"_admin.usageState": "IN_USE"}
 206         update_descriptor_usage_state(descriptor, db_collection, self.db)
 207         self.db.set_one.assert_called_once_with(
 208             db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
 209         )
 210
 211     @patch("osm_nbi.base_topic.detect_descriptor_usage")
 212     def test_update_descriptor_usage_state_to_not_in_use(self, mock_descriptor_usage):
 213         db_collection = "nsds"
 214         descriptor = deepcopy(db_nsd_content)
 215         mock_descriptor_usage.return_value = False
 216         descriptor_update = {"_admin.usageState": "NOT_IN_USE"}
 217         update_descriptor_usage_state(descriptor, db_collection, self.db)
 218         self.db.set_one.assert_called_once_with(
 219             db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
 220         )
 221
 222     @patch("osm_nbi.base_topic.detect_descriptor_usage")
 223     def test_update_descriptor_usage_state_db_exception(self, mock_descriptor_usage):
 224         db_collection = "nsd"
 225         descriptor = deepcopy(db_nsd_content)
 226         mock_descriptor_usage.side_effect = NBIBadArgumentsException
 227         with self.assertRaises(EngineException) as error:
 228             update_descriptor_usage_state(descriptor, db_collection, self.db)
 229             self.assertIn(
 230                 "db_collection should be equal to vnfds or nsds, db_collection",
 231                 error,
 232                 "wrong error type",
 233             )
 234         self.db.set_one.assert_not_called()
 235
 236
 237 if __name__ == "__main__":
 238     unittest.main()