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": [{"id": "mgmt-net", "mgmt-network": True}],
72 "constituent-cpd-id": "vnf-cp0-ext",
73 "virtual-link-profile-id": "mgmt-net",
78 self
.fs
.get_params
.return_value
= {}
79 self
.fs
.file_exists
.return_value
= False
80 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
81 "/tmp/" + str(uuid4()), "a+b"
84 vnfr_id
, _
= self
.vnfinstances
.new(
85 rollback
, session
, indata
, {}, headers
={"Content-Type": []}
87 vnfr
= self
.db
.get_one("vnfrs")
89 vnfr_id
, vnfr
["id"], "Mismatch between return id and database id"
93 vnfr
["_admin"]["nsState"],
94 "Database record must contain 'nsState' NOT_INSTANTIATED",
99 "vnfr record is not properly created for the given vnfd",
102 def test_show_vnfinstance(self
):
107 "project_id": [self
.vnfd_project
],
111 self
.db
.create_list("vnfrs", yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
))
112 actual_vnfr
= self
.db
.get_list("vnfrs")[0]
113 id = actual_vnfr
["_id"]
114 expected_vnfr
= self
.vnfinstances
.show(session
, id, filter_q
)
117 expected_vnfr
["_id"],
118 "Mismatch between return vnfr Id and database vnfr Id",
121 def test_delete_vnfinstance(self
):
126 "project_id": [self
.vnfd_project
],
129 self
.db
.create_list("vnfrs", yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
))
130 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
131 self
.db
.create_list("nsds", yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
))
133 self
.vnfr
= self
.db
.get_list("vnfrs")[0]
134 self
.vnfr_id
= self
.vnfr
["_id"]
135 self
.db
.set_one
= self
.db
.set_one
136 self
.db
.set_one
= Mock()
138 self
.vnfinstances
.delete(session
, self
.vnfr_id
)
139 msg_args
= self
.msg
.write
.call_args
[0]
140 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
143 class TestVnfLcmOpTopic(unittest
.TestCase
):
146 self
.fs
= Mock(FsBase())
147 self
.fs
.get_params
.return_value
= {"./fake/folder"}
148 self
.fs
.file_open
= mock_open()
149 self
.msg
= Mock(MsgBase())
151 self
.vnflcmop_topic
= VnfLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
152 self
.vnflcmop_topic
.check_quota
= Mock(return_value
=None) # skip quota
155 "vim_accounts", yaml
.load(db_vim_accounts_text
, Loader
=yaml
.Loader
)
157 self
.db
.create_list("nsds", yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
))
158 self
.db
.create_list("vnfds", yaml
.load(db_vnfm_vnfd_text
, Loader
=yaml
.Loader
))
159 self
.db
.create_list("vnfrs", yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
))
160 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
162 self
.vnfd
= self
.db
.get_list("vnfds")[0]
163 self
.vnfd_id
= self
.vnfd
["_id"]
164 self
.vnfr
= self
.db
.get_list("vnfrs")[0]
165 self
.vnfr_id
= self
.vnfr
["_id"]
167 self
.vnfd_project
= self
.vnfd
["_admin"]["projects_read"][0]
169 self
.vim
= self
.db
.get_list("vim_accounts")[0]
170 self
.vim_id
= self
.vim
["_id"]
172 def test_create_vnf_instantiate(self
):
177 "project_id": [self
.vnfd_project
],
181 "vnfInstanceId": self
.vnfr_id
,
182 "lcmOperationType": "instantiate",
183 "vnfName": "vnf_instance_name",
184 "vnfDescription": "vnf instance description",
185 "vnfId": self
.vnfd_id
,
186 "vimAccountId": self
.vim_id
,
190 vnflcmop_id
, _
= self
.vnflcmop_topic
.new(
191 rollback
, session
, indata
, kwargs
=None, headers
=headers
193 vnflcmop_info
= self
.db
.get_one("nslcmops")
196 vnflcmop_info
["_id"],
197 "Mismatch between return id and database '_id'",
200 vnflcmop_info
["lcmOperationType"] == "instantiate",
201 "Database record must contain 'lcmOperationType=instantiate'",
204 def test_show_vnflmcop(self
):
209 "project_id": [self
.vnfd_project
],
212 self
.db
.create_list("nslcmops", yaml
.load(db_nslcmops_text
, Loader
=yaml
.Loader
))
214 actual_lcmop
= self
.db
.get_list("nslcmops")[0]
215 id = actual_lcmop
["_id"]
216 vnfr
= self
.db
.get_list("vnfrs")[0]
217 vnfr_id
= vnfr
["_id"]
218 vnflcmop
= self
.vnflcmop_topic
.show(session
, id, filter_q
)
219 _id
= vnflcmop
["vnfInstanceId"]
223 "Mismatch between vnflcmop's vnfInstanceId and database vnfr's id",