NG-RO/osm_ng_ro/tests/test_ns.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 unittest
  19 from unittest.mock import patch, Mock
  20
  21 from osm_ng_ro.ns import Ns
  22
  23
  24 __author__ = "Eduardo Sousa"
  25 __date__ = "$19-NOV-2021 00:00:00$"
  26
  27
  28 class TestNs(unittest.TestCase):
  29     def setUp(self):
  30         pass
  31
  32     def test__create_task_without_extra_dict(self):
  33         expected_result = {
  34             "target_id": "vim_openstack_1",
  35             "action_id": "123456",
  36             "nsr_id": "654321",
  37             "task_id": "123456:1",
  38             "status": "SCHEDULED",
  39             "action": "CREATE",
  40             "item": "test_item",
  41             "target_record": "test_target_record",
  42             "target_record_id": "test_target_record_id",
  43         }
  44         deployment_info = {
  45             "action_id": "123456",
  46             "nsr_id": "654321",
  47             "task_index": 1,
  48         }
  49
  50         task = Ns._create_task(
  51             deployment_info=deployment_info,
  52             target_id="vim_openstack_1",
  53             item="test_item",
  54             action="CREATE",
  55             target_record="test_target_record",
  56             target_record_id="test_target_record_id",
  57         )
  58
  59         self.assertEqual(deployment_info.get("task_index"), 2)
  60         self.assertDictEqual(task, expected_result)
  61
  62     def test__create_task(self):
  63         expected_result = {
  64             "target_id": "vim_openstack_1",
  65             "action_id": "123456",
  66             "nsr_id": "654321",
  67             "task_id": "123456:1",
  68             "status": "SCHEDULED",
  69             "action": "CREATE",
  70             "item": "test_item",
  71             "target_record": "test_target_record",
  72             "target_record_id": "test_target_record_id",
  73             # values coming from extra_dict
  74             "params": "test_params",
  75             "find_params": "test_find_params",
  76             "depends_on": "test_depends_on",
  77         }
  78         deployment_info = {
  79             "action_id": "123456",
  80             "nsr_id": "654321",
  81             "task_index": 1,
  82         }
  83
  84         task = Ns._create_task(
  85             deployment_info=deployment_info,
  86             target_id="vim_openstack_1",
  87             item="test_item",
  88             action="CREATE",
  89             target_record="test_target_record",
  90             target_record_id="test_target_record_id",
  91             extra_dict={
  92                 "params": "test_params",
  93                 "find_params": "test_find_params",
  94                 "depends_on": "test_depends_on",
  95             },
  96         )
  97
  98         self.assertEqual(deployment_info.get("task_index"), 2)
  99         self.assertDictEqual(task, expected_result)
 100
 101     @patch("osm_ng_ro.ns.time")
 102     def test__create_ro_task(self, mock_time: Mock):
 103         now = 1637324838.994551
 104         mock_time.return_value = now
 105         task = {
 106             "target_id": "vim_openstack_1",
 107             "action_id": "123456",
 108             "nsr_id": "654321",
 109             "task_id": "123456:1",
 110             "status": "SCHEDULED",
 111             "action": "CREATE",
 112             "item": "test_item",
 113             "target_record": "test_target_record",
 114             "target_record_id": "test_target_record_id",
 115             # values coming from extra_dict
 116             "params": "test_params",
 117             "find_params": "test_find_params",
 118             "depends_on": "test_depends_on",
 119         }
 120         expected_result = {
 121             "_id": "123456:1",
 122             "locked_by": None,
 123             "locked_at": 0.0,
 124             "target_id": "vim_openstack_1",
 125             "vim_info": {
 126                 "created": False,
 127                 "created_items": None,
 128                 "vim_id": None,
 129                 "vim_name": None,
 130                 "vim_status": None,
 131                 "vim_details": None,
 132                 "refresh_at": None,
 133             },
 134             "modified_at": now,
 135             "created_at": now,
 136             "to_check_at": now,
 137             "tasks": [task],
 138         }
 139
 140         ro_task = Ns._create_ro_task(
 141             target_id="vim_openstack_1",
 142             task=task,
 143         )
 144
 145         self.assertDictEqual(ro_task, expected_result)
 146
 147     def test__process_image_params_with_empty_target_image(self):
 148         expected_result = {
 149             "find_params": {},
 150         }
 151         target_image = {}
 152
 153         result = Ns._process_image_params(
 154             target_image=target_image,
 155             vim_info=None,
 156             target_record_id=None,
 157         )
 158
 159         self.assertDictEqual(expected_result, result)
 160
 161     def test__process_image_params_with_wrong_target_image(self):
 162         expected_result = {
 163             "find_params": {},
 164         }
 165         target_image = {
 166             "no_image": "to_see_here",
 167         }
 168
 169         result = Ns._process_image_params(
 170             target_image=target_image,
 171             vim_info=None,
 172             target_record_id=None,
 173         )
 174
 175         self.assertDictEqual(expected_result, result)
 176
 177     def test__process_image_params_with_image(self):
 178         expected_result = {
 179             "find_params": {
 180                 "filter_dict": {
 181                     "name": "cirros",
 182                 },
 183             },
 184         }
 185         target_image = {
 186             "image": "cirros",
 187         }
 188
 189         result = Ns._process_image_params(
 190             target_image=target_image,
 191             vim_info=None,
 192             target_record_id=None,
 193         )
 194
 195         self.assertDictEqual(expected_result, result)
 196
 197     def test__process_image_params_with_vim_image_id(self):
 198         expected_result = {
 199             "find_params": {
 200                 "filter_dict": {
 201                     "id": "123456",
 202                 },
 203             },
 204         }
 205         target_image = {
 206             "vim_image_id": "123456",
 207         }
 208
 209         result = Ns._process_image_params(
 210             target_image=target_image,
 211             vim_info=None,
 212             target_record_id=None,
 213         )
 214
 215         self.assertDictEqual(expected_result, result)
 216
 217     def test__process_image_params_with_image_checksum(self):
 218         expected_result = {
 219             "find_params": {
 220                 "filter_dict": {
 221                     "checksum": "e3fc50a88d0a364313df4b21ef20c29e",
 222                 },
 223             },
 224         }
 225         target_image = {
 226             "image_checksum": "e3fc50a88d0a364313df4b21ef20c29e",
 227         }
 228
 229         result = Ns._process_image_params(
 230             target_image=target_image,
 231             vim_info=None,
 232             target_record_id=None,
 233         )
 234
 235         self.assertDictEqual(expected_result, result)
 236
 237     def test__get_resource_allocation_params_with_empty_target_image(self):
 238         expected_result = {}
 239         quota_descriptor = {}
 240
 241         result = Ns._get_resource_allocation_params(
 242             quota_descriptor=quota_descriptor,
 243         )
 244
 245         self.assertDictEqual(expected_result, result)
 246
 247     def test__get_resource_allocation_params_with_wrong_target_image(self):
 248         expected_result = {}
 249         quota_descriptor = {
 250             "no_quota": "present_here",
 251         }
 252
 253         result = Ns._get_resource_allocation_params(
 254             quota_descriptor=quota_descriptor,
 255         )
 256
 257         self.assertDictEqual(expected_result, result)
 258
 259     def test__get_resource_allocation_params_with_limit(self):
 260         expected_result = {
 261             "limit": 10,
 262         }
 263         quota_descriptor = {
 264             "limit": "10",
 265         }
 266
 267         result = Ns._get_resource_allocation_params(
 268             quota_descriptor=quota_descriptor,
 269         )
 270
 271         self.assertDictEqual(expected_result, result)
 272
 273     def test__get_resource_allocation_params_with_reserve(self):
 274         expected_result = {
 275             "reserve": 20,
 276         }
 277         quota_descriptor = {
 278             "reserve": "20",
 279         }
 280
 281         result = Ns._get_resource_allocation_params(
 282             quota_descriptor=quota_descriptor,
 283         )
 284
 285         self.assertDictEqual(expected_result, result)
 286
 287     def test__get_resource_allocation_params_with_shares(self):
 288         expected_result = {
 289             "shares": 30,
 290         }
 291         quota_descriptor = {
 292             "shares": "30",
 293         }
 294
 295         result = Ns._get_resource_allocation_params(
 296             quota_descriptor=quota_descriptor,
 297         )
 298
 299         self.assertDictEqual(expected_result, result)
 300
 301     def test__get_resource_allocation_params(self):
 302         expected_result = {
 303             "limit": 10,
 304             "reserve": 20,
 305             "shares": 30,
 306         }
 307         quota_descriptor = {
 308             "limit": "10",
 309             "reserve": "20",
 310             "shares": "30",
 311         }
 312
 313         result = Ns._get_resource_allocation_params(
 314             quota_descriptor=quota_descriptor,
 315         )
 316
 317         self.assertDictEqual(expected_result, result)