1 # Copyright 2021 Selvi Jayaraman (Tata Elxsi)
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 __author__
= "Selvi Jayaraman <selvi.j@tataelxsi.co.in>"
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 (
37 class TestVnfInstances(unittest
.TestCase
):
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("vim_accounts", yaml
.safe_load(db_vim_accounts_text
))
45 self
.db
.create_list("vnfds", yaml
.safe_load(db_vnfm_vnfd_text
))
46 self
.vnfd
= self
.db
.get_list("vnfds")[0]
47 self
.vnfd_id
= self
.vnfd
["id"]
48 self
.vnfd_project
= self
.vnfd
["_admin"]["projects_read"][0]
50 self
.vim
= self
.db
.get_list("vim_accounts")[0]
51 self
.vim_id
= self
.vim
["_id"]
53 @patch("osm_nbi.descriptor_topics.shutil")
54 @patch("osm_nbi.descriptor_topics.os.rename")
55 def test_create_identifier(self
, mock_rename
, mock_shutil
):
60 "project_id": [self
.vnfd_project
],
64 "vnfdId": self
.vnfd_id
,
65 "vnfInstanceName": "vnf_instance_name",
66 "vnfInstanceDescription": "vnf instance description",
67 "vimAccountId": self
.vim_id
,
69 "virtual-link-desc": [{"id": "mgmt-net", "mgmt-network": True}],
70 "constituent-cpd-id": "vnf-cp0-ext",
71 "virtual-link-profile-id": "mgmt-net",
76 self
.fs
.get_params
.return_value
= {}
77 self
.fs
.file_exists
.return_value
= False
78 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
79 "/tmp/" + str(uuid4()), "a+b"
82 vnfr_id
, _
= self
.vnfinstances
.new(
83 rollback
, session
, indata
, {}, headers
={"Content-Type": []}
85 vnfr
= self
.db
.get_one("vnfrs")
87 vnfr_id
, vnfr
["id"], "Mismatch between return id and database id"
90 "nsState" not in vnfr
["_admin"],
91 "Database VNF record must not contain 'nsState'",
96 "vnfr record is not properly created for the given vnfd",
99 def test_show_vnfinstance(self
):
104 "project_id": [self
.vnfd_project
],
108 self
.db
.create_list("vnfrs", yaml
.safe_load(db_vnfrs_text
))
109 actual_vnfr
= self
.db
.get_list("vnfrs")[0]
110 id = actual_vnfr
["_id"]
111 expected_vnfr
= self
.vnfinstances
.show(session
, id, filter_q
)
114 expected_vnfr
["_id"],
115 "Mismatch between return vnfr Id and database vnfr Id",
118 def test_delete_vnfinstance(self
):
123 "project_id": [self
.vnfd_project
],
126 self
.db
.create_list("vnfrs", yaml
.safe_load(db_vnfrs_text
))
127 self
.db
.create_list("nsrs", yaml
.safe_load(db_nsrs_text
))
128 self
.db
.create_list("nsds", yaml
.safe_load(db_nsds_text
))
130 self
.vnfr
= self
.db
.get_list("vnfrs")[0]
131 self
.vnfr_id
= self
.vnfr
["_id"]
132 self
.db
.set_one
= self
.db
.set_one
133 self
.db
.set_one
= Mock()
135 self
.vnfinstances
.delete(session
, self
.vnfr_id
)
136 msg_args
= self
.msg
.write
.call_args
[0]
137 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
140 class TestVnfLcmOpTopic(unittest
.TestCase
):
143 self
.fs
= Mock(FsBase())
144 self
.fs
.get_params
.return_value
= {"./fake/folder"}
145 self
.fs
.file_open
= mock_open()
146 self
.msg
= Mock(MsgBase())
148 self
.vnflcmop_topic
= VnfLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
149 self
.vnflcmop_topic
.check_quota
= Mock(return_value
=None) # skip quota
151 self
.db
.create_list("vim_accounts", yaml
.safe_load(db_vim_accounts_text
))
152 self
.db
.create_list("nsds", yaml
.safe_load(db_nsds_text
))
153 self
.db
.create_list("vnfds", yaml
.safe_load(db_vnfm_vnfd_text
))
154 self
.db
.create_list("vnfrs", yaml
.safe_load(db_vnfrs_text
))
155 self
.db
.create_list("nsrs", yaml
.safe_load(db_nsrs_text
))
157 self
.vnfd
= self
.db
.get_list("vnfds")[0]
158 self
.vnfd_id
= self
.vnfd
["_id"]
159 self
.vnfr
= self
.db
.get_list("vnfrs")[0]
160 self
.vnfr_id
= self
.vnfr
["_id"]
162 self
.vnfd_project
= self
.vnfd
["_admin"]["projects_read"][0]
164 self
.vim
= self
.db
.get_list("vim_accounts")[0]
165 self
.vim_id
= self
.vim
["_id"]
167 def test_create_vnf_instantiate(self
):
172 "project_id": [self
.vnfd_project
],
176 "vnfInstanceId": self
.vnfr_id
,
177 "lcmOperationType": "instantiate",
178 "vnfName": "vnf_instance_name",
179 "vnfDescription": "vnf instance description",
180 "vnfId": self
.vnfd_id
,
181 "vimAccountId": self
.vim_id
,
185 vnflcmop_id
, _
= self
.vnflcmop_topic
.new(
186 rollback
, session
, indata
, kwargs
=None, headers
=headers
188 vnflcmop_info
= self
.db
.get_one("nslcmops")
191 vnflcmop_info
["_id"],
192 "Mismatch between return id and database '_id'",
195 vnflcmop_info
["lcmOperationType"] == "instantiate",
196 "Database record must contain 'lcmOperationType=instantiate'",
199 def test_show_vnflmcop(self
):
204 "project_id": [self
.vnfd_project
],
207 self
.db
.create_list("nslcmops", yaml
.safe_load(db_nslcmops_text
))
209 actual_lcmop
= self
.db
.get_list("nslcmops")[0]
210 id = actual_lcmop
["_id"]
211 vnfr
= self
.db
.get_list("vnfrs")[0]
212 vnfr_id
= vnfr
["_id"]
213 vnflcmop
= self
.vnflcmop_topic
.show(session
, id, filter_q
)
214 _id
= vnflcmop
["vnfInstanceId"]
218 "Mismatch between vnflcmop's vnfInstanceId and database vnfr's id",