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 = {"no_image": "to_see_here"}
 166
 167         result = Ns._process_image_params(
 168             target_image=target_image,
 169             vim_info=None,
 170             target_record_id=None,
 171         )
 172
 173         self.assertDictEqual(expected_result, result)
 174
 175     def test__process_image_params_with_image(self):
 176         expected_result = {
 177             "find_params": {
 178                 "filter_dict": {
 179                     "name": "cirros",
 180                 },
 181             },
 182         }
 183         target_image = {"image": "cirros"}
 184
 185         result = Ns._process_image_params(
 186             target_image=target_image,
 187             vim_info=None,
 188             target_record_id=None,
 189         )
 190
 191         self.assertDictEqual(expected_result, result)
 192
 193     def test__process_image_params_with_vim_image_id(self):
 194         expected_result = {
 195             "find_params": {
 196                 "filter_dict": {
 197                     "id": "123456",
 198                 },
 199             },
 200         }
 201         target_image = {"vim_image_id": "123456"}
 202
 203         result = Ns._process_image_params(
 204             target_image=target_image,
 205             vim_info=None,
 206             target_record_id=None,
 207         )
 208
 209         self.assertDictEqual(expected_result, result)
 210
 211     def test__process_image_params_with_image_checksum(self):
 212         expected_result = {
 213             "find_params": {
 214                 "filter_dict": {
 215                     "checksum": "e3fc50a88d0a364313df4b21ef20c29e",
 216                 },
 217             },
 218         }
 219         target_image = {"image_checksum": "e3fc50a88d0a364313df4b21ef20c29e"}
 220
 221         result = Ns._process_image_params(
 222             target_image=target_image,
 223             vim_info=None,
 224             target_record_id=None,
 225         )
 226
 227         self.assertDictEqual(expected_result, result)