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)
45 "vim_accounts", yaml
.load(db_vim_accounts_text
, Loader
=yaml
.Loader
)
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]
52 self
.vim
= self
.db
.get_list("vim_accounts")[0]
53 self
.vim_id
= self
.vim
["_id"]
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
):
62 "project_id": [self
.vnfd_project
],
66 "vnfdId": self
.vnfd_id
,
67 "vnfInstanceName": "vnf_instance_name",
68 "vnfInstanceDescription": "vnf instance description",
69 "vimAccountId": self
.vim_id
,
71 "virtual-link-desc": [
77 "constituent-cpd-id": "vnf-cp0-ext",
78 "virtual-link-profile-id": "mgmt-net"
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"
90 vnfr_id
, _
= self
.vnfinstances
.new(
91 rollback
, session
, indata
, {}, headers
={"Content-Type": []}
93 vnfr
= self
.db
.get_one("vnfrs")
97 "Mismatch between return id and database id"
101 vnfr
["_admin"]["nsState"],
102 "Database record must contain 'nsState' NOT_INSTANTIATED"
107 "vnfr record is not properly created for the given vnfd")
109 def test_show_vnfinstance(self
):
114 "project_id": [self
.vnfd_project
],
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
)
124 expected_vnfr
["_id"],
125 "Mismatch between return vnfr Id and database vnfr Id"
128 def test_delete_vnfinstance(self
):
133 "project_id": [self
.vnfd_project
],
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
))
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()
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")
150 class TestVnfLcmOpTopic(unittest
.TestCase
):
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())
158 self
.vnflcmop_topic
= VnfLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
159 self
.vnflcmop_topic
.check_quota
= Mock(return_value
=None) # skip quota
162 "vim_accounts", yaml
.load(db_vim_accounts_text
, Loader
=yaml
.Loader
)
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
))
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"]
174 self
.vnfd_project
= self
.vnfd
["_admin"]["projects_read"][0]
176 self
.vim
= self
.db
.get_list("vim_accounts")[0]
177 self
.vim_id
= self
.vim
["_id"]
179 def test_create_vnf_instantiate(self
):
184 "project_id": [self
.vnfd_project
],
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
197 vnflcmop_id
, _
= self
.vnflcmop_topic
.new(
198 rollback
, session
, indata
, kwargs
=None, headers
=headers
200 vnflcmop_info
= self
.db
.get_one("nslcmops")
203 vnflcmop_info
["_id"],
204 "Mismatch between return id and database '_id'",
207 vnflcmop_info
["lcmOperationType"] == "instantiate",
208 "Database record must contain 'lcmOperationType=instantiate'",
211 def test_show_vnflmcop(self
):
216 "project_id": [self
.vnfd_project
],
219 self
.db
.create_list("nslcmops", yaml
.load(db_nslcmops_text
, Loader
=yaml
.Loader
))
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")