+++ /dev/null
-# Copyright 2021 Selvi Jayaraman (Tata Elxsi)
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-# implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-__author__ = "Selvi Jayaraman <selvi.j@tataelxsi.co.in>"
-
-import unittest
-from uuid import uuid4
-from unittest.mock import Mock, patch, mock_open
-from osm_common.dbmemory import DbMemory
-from osm_common.fsbase import FsBase
-from osm_common.msgbase import MsgBase
-from osm_nbi.vnf_instance_topics import VnfInstances, VnfLcmOpTopic
-from osm_nbi.instance_topics import NsrTopic
-from osm_nbi.tests.test_db_descriptors import (
- db_vim_accounts_text,
- db_vnfm_vnfd_text,
- db_nsds_text,
- db_nsrs_text,
- db_vnfrs_text,
- db_nslcmops_text
-)
-import yaml
-
-
-class TestVnfInstances(unittest.TestCase):
- def setUp(self):
- self.db = DbMemory()
- self.fs = Mock(FsBase())
- self.msg = Mock(MsgBase())
- self.vnfinstances = VnfInstances(self.db, self.fs, self.msg, None)
- self.nsrtopic = NsrTopic(self.db, self.fs, self.msg, None)
- self.db.create_list(
- "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
- )
- self.db.create_list("vnfds", yaml.load(db_vnfm_vnfd_text, Loader=yaml.Loader))
- self.vnfd = self.db.get_list("vnfds")[0]
- self.vnfd_id = self.vnfd["id"]
- self.vnfd_project = self.vnfd["_admin"]["projects_read"][0]
-
- self.vim = self.db.get_list("vim_accounts")[0]
- self.vim_id = self.vim["_id"]
-
- @patch("osm_nbi.descriptor_topics.shutil")
- @patch("osm_nbi.descriptor_topics.os.rename")
- def test_create_identifier(self, mock_rename, mock_shutil):
- session = {
- "force": True,
- "admin": False,
- "public": False,
- "project_id": [self.vnfd_project],
- "method": "write",
- }
- indata = {
- "vnfdId": self.vnfd_id,
- "vnfInstanceName": "vnf_instance_name",
- "vnfInstanceDescription": "vnf instance description",
- "vimAccountId": self.vim_id,
- "additionalParams": {
- "virtual-link-desc": [
- {
- "id": "mgmt-net",
- "mgmt-network": True
- }
- ],
- "constituent-cpd-id": "vnf-cp0-ext",
- "virtual-link-profile-id": "mgmt-net"
- }
- }
- rollback = []
- headers = {}
- self.fs.path = ""
- self.fs.get_params.return_value = {}
- self.fs.file_exists.return_value = False
- self.fs.file_open.side_effect = lambda path, mode: open(
- "/tmp/" + str(uuid4()), "a+b"
- )
-
- vnfr_id, _ = self.vnfinstances.new(
- rollback, session, indata, {}, headers={"Content-Type": []}
- )
- vnfr = self.db.get_one("vnfrs")
- self.assertEqual(
- vnfr_id,
- vnfr["id"],
- "Mismatch between return id and database id"
- )
- self.assertEqual(
- "NOT_INSTANTIATED",
- vnfr["_admin"]["nsState"],
- "Database record must contain 'nsState' NOT_INSTANTIATED"
- )
- self.assertEqual(
- self.vnfd_id,
- vnfr["vnfd-ref"],
- "vnfr record is not properly created for the given vnfd")
-
- def test_show_vnfinstance(self):
- session = {
- "force": False,
- "admin": False,
- "public": False,
- "project_id": [self.vnfd_project],
- "method": "write",
- }
- filter_q = {}
- self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
- actual_vnfr = self.db.get_list("vnfrs")[0]
- id = actual_vnfr["_id"]
- expected_vnfr = self.vnfinstances.show(session, id, filter_q)
- self.assertEqual(
- actual_vnfr["_id"],
- expected_vnfr["_id"],
- "Mismatch between return vnfr Id and database vnfr Id"
- )
-
- def test_delete_vnfinstance(self):
- session = {
- "force": False,
- "admin": False,
- "public": False,
- "project_id": [self.vnfd_project],
- "method": "delete",
- }
- self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
- self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
- self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
-
- self.vnfr = self.db.get_list("vnfrs")[0]
- self.vnfr_id = self.vnfr["_id"]
- self.db.set_one = self.db.set_one
- self.db.set_one = Mock()
-
- self.vnfinstances.delete(session, self.vnfr_id)
- msg_args = self.msg.write.call_args[0]
- self.assertEqual(msg_args[1], "deleted", "Wrong message action")
-
-
-class TestVnfLcmOpTopic(unittest.TestCase):
- def setUp(self):
- self.db = DbMemory()
- self.fs = Mock(FsBase())
- self.fs.get_params.return_value = {"./fake/folder"}
- self.fs.file_open = mock_open()
- self.msg = Mock(MsgBase())
-
- self.vnflcmop_topic = VnfLcmOpTopic(self.db, self.fs, self.msg, None)
- self.vnflcmop_topic.check_quota = Mock(return_value=None) # skip quota
-
- self.db.create_list(
- "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
- )
- self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
- self.db.create_list("vnfds", yaml.load(db_vnfm_vnfd_text, Loader=yaml.Loader))
- self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
- self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
-
- self.vnfd = self.db.get_list("vnfds")[0]
- self.vnfd_id = self.vnfd["_id"]
- self.vnfr = self.db.get_list("vnfrs")[0]
- self.vnfr_id = self.vnfr["_id"]
-
- self.vnfd_project = self.vnfd["_admin"]["projects_read"][0]
-
- self.vim = self.db.get_list("vim_accounts")[0]
- self.vim_id = self.vim["_id"]
-
- def test_create_vnf_instantiate(self):
- session = {
- "force": False,
- "admin": False,
- "public": False,
- "project_id": [self.vnfd_project],
- "method": "write",
- }
- indata = {
- "vnfInstanceId": self.vnfr_id,
- "lcmOperationType": "instantiate",
- "vnfName": "vnf_instance_name",
- "vnfDescription": "vnf instance description",
- "vnfId": self.vnfd_id,
- "vimAccountId": self.vim_id
- }
- rollback = []
- headers = {}
- vnflcmop_id, _ = self.vnflcmop_topic.new(
- rollback, session, indata, kwargs=None, headers=headers
- )
- vnflcmop_info = self.db.get_one("nslcmops")
- self.assertEqual(
- vnflcmop_id,
- vnflcmop_info["_id"],
- "Mismatch between return id and database '_id'",
- )
- self.assertTrue(
- vnflcmop_info["lcmOperationType"] == "instantiate",
- "Database record must contain 'lcmOperationType=instantiate'",
- )
-
- def test_show_vnflmcop(self):
- session = {
- "force": False,
- "admin": False,
- "public": False,
- "project_id": [self.vnfd_project],
- "method": "write",
- }
- self.db.create_list("nslcmops", yaml.load(db_nslcmops_text, Loader=yaml.Loader))
- filter_q = {}
- actual_lcmop = self.db.get_list("nslcmops")[0]
- id = actual_lcmop["_id"]
- vnfr = self.db.get_list("vnfrs")[0]
- vnfr_id = vnfr["_id"]
- vnflcmop = self.vnflcmop_topic.show(session, id, filter_q)
- _id = vnflcmop["vnfInstanceId"]
- self.assertEqual(_id, vnfr_id, "Mismatch between vnflcmop's vnfInstanceId and database vnfr's id")
projects / osm / NBI.git / blobdiff