X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_nbi%2Ftests%2Ftest_osm_vnfm.py;fp=osm_nbi%2Ftests%2Ftest_osm_vnfm.py;h=0000000000000000000000000000000000000000;hb=b1186b0b9ea287010c9f37ba6062512ea93f0ed4;hp=a7ea8a3fc29670e38cff7f3ca5b8a1ed3eb8f47d;hpb=59e9d88971c2f513c85ca7095a1e8493f9ebb513;p=osm%2FNBI.git diff --git a/osm_nbi/tests/test_osm_vnfm.py b/osm_nbi/tests/test_osm_vnfm.py deleted file mode 100644 index a7ea8a3..0000000 --- a/osm_nbi/tests/test_osm_vnfm.py +++ /dev/null @@ -1,227 +0,0 @@ -# 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 " - -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")