osm_nbi/tests/test_osm_vnfm.py

   1 # Copyright 2021 Selvi Jayaraman (Tata Elxsi)
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 __author__ = "Selvi Jayaraman <selvi.j@tataelxsi.co.in>"
  17
  18 import unittest
  19 from uuid import uuid4
  20 from unittest.mock import Mock, patch, mock_open
  21 from osm_common.dbmemory import DbMemory
  22 from osm_common.fsbase import FsBase
  23 from osm_common.msgbase import MsgBase
  24 from osm_nbi.vnf_instance_topics import VnfInstances, VnfLcmOpTopic
  25 from osm_nbi.instance_topics import NsrTopic
  26 from osm_nbi.tests.test_db_descriptors import (
  27     db_vim_accounts_text,
  28     db_vnfm_vnfd_text,
  29     db_nsds_text,
  30     db_nsrs_text,
  31     db_vnfrs_text,
  32     db_nslcmops_text
  33 )
  34 import yaml
  35
  36
  37 class TestVnfInstances(unittest.TestCase):
  38     def setUp(self):
  39         self.db = DbMemory()
  40         self.fs = Mock(FsBase())
  41         self.msg = Mock(MsgBase())
  42         self.vnfinstances = VnfInstances(self.db, self.fs, self.msg, None)
  43         self.nsrtopic = NsrTopic(self.db, self.fs, self.msg, None)
  44         self.db.create_list(
  45             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
  46         )
  47         self.db.create_list("vnfds", yaml.load(db_vnfm_vnfd_text, Loader=yaml.Loader))
  48         self.vnfd = self.db.get_list("vnfds")[0]
  49         self.vnfd_id = self.vnfd["id"]
  50         self.vnfd_project = self.vnfd["_admin"]["projects_read"][0]
  51
  52         self.vim = self.db.get_list("vim_accounts")[0]
  53         self.vim_id = self.vim["_id"]
  54
  55     @patch("osm_nbi.descriptor_topics.shutil")
  56     @patch("osm_nbi.descriptor_topics.os.rename")
  57     def test_create_identifier(self, mock_rename, mock_shutil):
  58         session = {
  59             "force": True,
  60             "admin": False,
  61             "public": False,
  62             "project_id": [self.vnfd_project],
  63             "method": "write",
  64         }
  65         indata = {
  66             "vnfdId": self.vnfd_id,
  67             "vnfInstanceName": "vnf_instance_name",
  68             "vnfInstanceDescription": "vnf instance description",
  69             "vimAccountId": self.vim_id,
  70             "additionalParams": {
  71                 "virtual-link-desc": [
  72                     {
  73                         "id": "mgmt-net",
  74                         "mgmt-network": True
  75                     }
  76                 ],
  77                 "constituent-cpd-id": "vnf-cp0-ext",
  78                 "virtual-link-profile-id": "mgmt-net"
  79             }
  80         }
  81         rollback = []
  82         headers = {}
  83         self.fs.path = ""
  84         self.fs.get_params.return_value = {}
  85         self.fs.file_exists.return_value = False
  86         self.fs.file_open.side_effect = lambda path, mode: open(
  87                 "/tmp/" + str(uuid4()), "a+b"
  88                 )
  89
  90         vnfr_id, _ = self.vnfinstances.new(
  91             rollback, session, indata, {}, headers={"Content-Type": []}
  92         )
  93         vnfr = self.db.get_one("vnfrs")
  94         self.assertEqual(
  95                 vnfr_id,
  96                 vnfr["id"],
  97                 "Mismatch between return id and database id"
  98                 )
  99         self.assertEqual(
 100                 "NOT_INSTANTIATED",
 101                 vnfr["_admin"]["nsState"],
 102                 "Database record must contain 'nsState' NOT_INSTANTIATED"
 103                 )
 104         self.assertEqual(
 105                 self.vnfd_id,
 106                 vnfr["vnfd-ref"],
 107                 "vnfr record is not properly created for the given vnfd")
 108
 109     def test_show_vnfinstance(self):
 110         session = {
 111             "force": False,
 112             "admin": False,
 113             "public": False,
 114             "project_id": [self.vnfd_project],
 115             "method": "write",
 116         }
 117         filter_q = {}
 118         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
 119         actual_vnfr = self.db.get_list("vnfrs")[0]
 120         id = actual_vnfr["_id"]
 121         expected_vnfr = self.vnfinstances.show(session, id, filter_q)
 122         self.assertEqual(
 123             actual_vnfr["_id"],
 124             expected_vnfr["_id"],
 125             "Mismatch between return vnfr Id and database vnfr Id"
 126         )
 127
 128     def test_delete_vnfinstance(self):
 129         session = {
 130             "force": False,
 131             "admin": False,
 132             "public": False,
 133             "project_id": [self.vnfd_project],
 134             "method": "delete",
 135         }
 136         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
 137         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 138         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
 139
 140         self.vnfr = self.db.get_list("vnfrs")[0]
 141         self.vnfr_id = self.vnfr["_id"]
 142         self.db.set_one = self.db.set_one
 143         self.db.set_one = Mock()
 144
 145         self.vnfinstances.delete(session, self.vnfr_id)
 146         msg_args = self.msg.write.call_args[0]
 147         self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 148
 149
 150 class TestVnfLcmOpTopic(unittest.TestCase):
 151     def setUp(self):
 152         self.db = DbMemory()
 153         self.fs = Mock(FsBase())
 154         self.fs.get_params.return_value = {"./fake/folder"}
 155         self.fs.file_open = mock_open()
 156         self.msg = Mock(MsgBase())
 157
 158         self.vnflcmop_topic = VnfLcmOpTopic(self.db, self.fs, self.msg, None)
 159         self.vnflcmop_topic.check_quota = Mock(return_value=None)  # skip quota
 160
 161         self.db.create_list(
 162             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
 163         )
 164         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
 165         self.db.create_list("vnfds", yaml.load(db_vnfm_vnfd_text, Loader=yaml.Loader))
 166         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
 167         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 168
 169         self.vnfd = self.db.get_list("vnfds")[0]
 170         self.vnfd_id = self.vnfd["_id"]
 171         self.vnfr = self.db.get_list("vnfrs")[0]
 172         self.vnfr_id = self.vnfr["_id"]
 173
 174         self.vnfd_project = self.vnfd["_admin"]["projects_read"][0]
 175
 176         self.vim = self.db.get_list("vim_accounts")[0]
 177         self.vim_id = self.vim["_id"]
 178
 179     def test_create_vnf_instantiate(self):
 180         session = {
 181             "force": False,
 182             "admin": False,
 183             "public": False,
 184             "project_id": [self.vnfd_project],
 185             "method": "write",
 186         }
 187         indata = {
 188             "vnfInstanceId": self.vnfr_id,
 189             "lcmOperationType": "instantiate",
 190             "vnfName": "vnf_instance_name",
 191             "vnfDescription": "vnf instance description",
 192             "vnfId": self.vnfd_id,
 193             "vimAccountId": self.vim_id
 194         }
 195         rollback = []
 196         headers = {}
 197         vnflcmop_id, _ = self.vnflcmop_topic.new(
 198             rollback, session, indata, kwargs=None, headers=headers
 199         )
 200         vnflcmop_info = self.db.get_one("nslcmops")
 201         self.assertEqual(
 202             vnflcmop_id,
 203             vnflcmop_info["_id"],
 204             "Mismatch between return id and database '_id'",
 205         )
 206         self.assertTrue(
 207             vnflcmop_info["lcmOperationType"] == "instantiate",
 208             "Database record must contain 'lcmOperationType=instantiate'",
 209         )
 210
 211     def test_show_vnflmcop(self):
 212         session = {
 213             "force": False,
 214             "admin": False,
 215             "public": False,
 216             "project_id": [self.vnfd_project],
 217             "method": "write",
 218         }
 219         self.db.create_list("nslcmops", yaml.load(db_nslcmops_text, Loader=yaml.Loader))
 220         filter_q = {}
 221         actual_lcmop = self.db.get_list("nslcmops")[0]
 222         id = actual_lcmop["_id"]
 223         vnfr = self.db.get_list("vnfrs")[0]
 224         vnfr_id = vnfr["_id"]
 225         vnflcmop = self.vnflcmop_topic.show(session, id, filter_q)
 226         _id = vnflcmop["vnfInstanceId"]
 227         self.assertEqual(_id, vnfr_id, "Mismatch between vnflcmop's vnfInstanceId and database vnfr's id")