NG-RO/osm_ng_ro/tests/test_ns_thread.py

   1 #######################################################################################
   2 # Copyright ETSI Contributors and Others.
   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
  18 import logging
  19 import unittest
  20 from unittest.mock import MagicMock, patch
  21
  22 from osm_ng_ro.ns_thread import VimInteractionAffinityGroup
  23
  24
  25 class TestVimInteractionAffinityGroup(unittest.TestCase):
  26     def setUp(self):
  27         module_name = "osm_ro_plugin"
  28         self.target_vim = MagicMock(name=f"{module_name}.vimconn.VimConnector")
  29         self.task_depends = None
  30
  31         patches = [patch(f"{module_name}.vimconn.VimConnector", self.target_vim)]
  32
  33         # Enabling mocks and add cleanups
  34         for mock in patches:
  35             mock.start()
  36             self.addCleanup(mock.stop)
  37
  38     def test__new_affinity_group_ok(self):
  39         """
  40         create affinity group with attributes set in params
  41         """
  42         db = "test_db"
  43         logger = "test_logger"
  44         my_vims = "test-vim"
  45         db_vims = {
  46             0: {
  47                 "config": {},
  48             },
  49         }
  50
  51         instance = VimInteractionAffinityGroup(db, logger, my_vims, db_vims)
  52         with patch.object(instance, "my_vims", [self.target_vim]), patch.object(
  53             instance, "logger", logging
  54         ), patch.object(instance, "db_vims", db_vims):
  55             ro_task = {
  56                 "target_id": 0,
  57                 "tasks": {
  58                     "task_index_1": {
  59                         "target_id": 0,
  60                         "action_id": "123456",
  61                         "nsr_id": "654321",
  62                         "task_id": "123456:1",
  63                         "status": "SCHEDULED",
  64                         "action": "CREATE",
  65                         "item": "test_item",
  66                         "target_record": "test_target_record",
  67                         "target_record_id": "test_target_record_id",
  68                         # values coming from extra_dict
  69                         "params": {
  70                             "affinity_group_data": {
  71                                 "name": "affinity_group_1",
  72                                 "type": "affinity",
  73                                 "scope": "nfvi-node",
  74                             }
  75                         },
  76                         "find_params": {},
  77                         "depends_on": "test_depends_on",
  78                     },
  79                 },
  80             }
  81
  82             task_index = "task_index_1"
  83             self.target_vim.new_affinity_group.return_value = (
  84                 "sample_affinity_group_id_1"
  85             )
  86             result = instance.new(ro_task, task_index, self.task_depends)
  87             self.assertEqual(result[0], "DONE")
  88             self.assertEqual(result[1].get("vim_id"), "sample_affinity_group_id_1")
  89             self.assertEqual(result[1].get("created"), True)
  90             self.assertEqual(result[1].get("vim_status"), "DONE")
  91
  92     def test__new_affinity_group_failed(self):
  93         """
  94         create affinity group with no attributes set in params
  95         """
  96         db = "test_db"
  97         logger = "test_logger"
  98         my_vims = "test-vim"
  99         db_vims = {
 100             0: {
 101                 "config": {},
 102             },
 103         }
 104
 105         instance = VimInteractionAffinityGroup(db, logger, my_vims, db_vims)
 106         with patch.object(instance, "my_vims", [self.target_vim]), patch.object(
 107             instance, "logger", logging
 108         ), patch.object(instance, "db_vims", db_vims):
 109             ro_task = {
 110                 "target_id": 0,
 111                 "tasks": {
 112                     "task_index_2": {
 113                         "target_id": 0,
 114                         "action_id": "123456",
 115                         "nsr_id": "654321",
 116                         "task_id": "123456:1",
 117                         "status": "SCHEDULED",
 118                         "action": "CREATE",
 119                         "item": "test_item",
 120                         "target_record": "test_target_record",
 121                         "target_record_id": "test_target_record_id",
 122                         # values coming from extra_dict
 123                         "params": {},
 124                         "find_params": {},
 125                         "depends_on": "test_depends_on",
 126                     },
 127                 },
 128             }
 129
 130             task_index = "task_index_2"
 131             self.target_vim.new_affinity_group.return_value = (
 132                 "sample_affinity_group_id_1"
 133             )
 134             result = instance.new(ro_task, task_index, self.task_depends)
 135             self.assertEqual(result[0], "DONE")
 136             self.assertEqual(result[1].get("vim_id"), None)
 137             self.assertEqual(result[1].get("created"), False)
 138             self.assertEqual(result[1].get("vim_status"), "DONE")
 139
 140     def test__delete_affinity_group_ok(self):
 141         """
 142         delete affinity group with a proper vim_id
 143         """
 144         db = "test_db"
 145         logger = "test_logger"
 146         my_vims = "test-vim"
 147         db_vims = {
 148             0: {
 149                 "config": {},
 150             },
 151         }
 152
 153         instance = VimInteractionAffinityGroup(db, logger, my_vims, db_vims)
 154         with patch.object(instance, "my_vims", [self.target_vim]), patch.object(
 155             instance, "logger", logging
 156         ), patch.object(instance, "db_vims", db_vims):
 157             ro_task = {
 158                 "target_id": 0,
 159                 "tasks": {
 160                     "task_index_3": {
 161                         "target_id": 0,
 162                         "task_id": "123456:1",
 163                     },
 164                 },
 165                 "vim_info": {
 166                     "created": False,
 167                     "created_items": None,
 168                     "vim_id": "sample_affinity_group_id_3",
 169                     "vim_name": "sample_affinity_group_id_3",
 170                     "vim_status": None,
 171                     "vim_details": "some-details",
 172                     "refresh_at": None,
 173                 },
 174             }
 175
 176             task_index = "task_index_3"
 177             self.target_vim.delete_affinity_group.return_value = (
 178                 "sample_affinity_group_id_3"
 179             )
 180             result = instance.delete(ro_task, task_index)
 181             self.assertEqual(result[0], "DONE")
 182             self.assertEqual(result[1].get("vim_details"), "DELETED")
 183             self.assertEqual(result[1].get("created"), False)
 184             self.assertEqual(result[1].get("vim_status"), "DELETED")
 185
 186     def test__delete_affinity_group_failed(self):
 187         """
 188         delete affinity group with missing vim_id
 189         """
 190         db = "test_db"
 191         logger = "test_logger"
 192         my_vims = "test-vim"
 193         db_vims = {
 194             0: {
 195                 "config": {},
 196             },
 197         }
 198
 199         instance = VimInteractionAffinityGroup(db, logger, my_vims, db_vims)
 200         with patch.object(instance, "my_vims", [self.target_vim]), patch.object(
 201             instance, "logger", logging
 202         ), patch.object(instance, "db_vims", db_vims):
 203             ro_task = {
 204                 "target_id": 0,
 205                 "tasks": {
 206                     "task_index_4": {
 207                         "target_id": 0,
 208                         "task_id": "123456:1",
 209                     },
 210                 },
 211                 "vim_info": {
 212                     "created": False,
 213                     "created_items": None,
 214                     "vim_id": None,
 215                     "vim_name": None,
 216                     "vim_status": None,
 217                     "vim_details": "some-details",
 218                     "refresh_at": None,
 219                 },
 220             }
 221
 222             task_index = "task_index_4"
 223             self.target_vim.delete_affinity_group.return_value = ""
 224             result = instance.delete(ro_task, task_index)
 225             self.assertEqual(result[0], "DONE")
 226             self.assertEqual(result[1].get("vim_details"), "DELETED")
 227             self.assertEqual(result[1].get("created"), False)
 228             self.assertEqual(result[1].get("vim_status"), "DELETED")