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"
92 "nsState" not in vnfr
["_admin"],
93 "Database VNF record must not contain 'nsState'",
98 "vnfr record is not properly created for the given vnfd",
101 def test_show_vnfinstance(self
):
106 "project_id": [self
.vnfd_project
],
110 self
.db
.create_list("vnfrs", yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
))
111 actual_vnfr
= self
.db
.get_list("vnfrs")[0]
112 id = actual_vnfr
["_id"]
113 expected_vnfr
= self
.vnfinstances
.show(session
, id, filter_q
)
116 expected_vnfr
["_id"],
117 "Mismatch between return vnfr Id and database vnfr Id",
120 def test_delete_vnfinstance(self
):
125 "project_id": [self
.vnfd_project
],
128 self
.db
.create_list("vnfrs", yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
))
129 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
130 self
.db
.create_list("nsds", yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
))
132 self
.vnfr
= self
.db
.get_list("vnfrs")[0]
133 self
.vnfr_id
= self
.vnfr
["_id"]
134 self
.db
.set_one
= self
.db
.set_one
135 self
.db
.set_one
= Mock()
137 self
.vnfinstances
.delete(session
, self
.vnfr_id
)
138 msg_args
= self
.msg
.write
.call_args
[0]
139 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
142 class TestVnfLcmOpTopic(unittest
.TestCase
):
145 self
.fs
= Mock(FsBase())
146 self
.fs
.get_params
.return_value
= {"./fake/folder"}
147 self
.fs
.file_open
= mock_open()
148 self
.msg
= Mock(MsgBase())
150 self
.vnflcmop_topic
= VnfLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
151 self
.vnflcmop_topic
.check_quota
= Mock(return_value
=None) # skip quota
154 "vim_accounts", yaml
.load(db_vim_accounts_text
, Loader
=yaml
.Loader
)
156 self
.db
.create_list("nsds", yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
))
157 self
.db
.create_list("vnfds", yaml
.load(db_vnfm_vnfd_text
, Loader
=yaml
.Loader
))
158 self
.db
.create_list("vnfrs", yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
))
159 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
161 self
.vnfd
= self
.db
.get_list("vnfds")[0]
162 self
.vnfd_id
= self
.vnfd
["_id"]
163 self
.vnfr
= self
.db
.get_list("vnfrs")[0]
164 self
.vnfr_id
= self
.vnfr
["_id"]
166 self
.vnfd_project
= self
.vnfd
["_admin"]["projects_read"][0]
168 self
.vim
= self
.db
.get_list("vim_accounts")[0]
169 self
.vim_id
= self
.vim
["_id"]
171 def test_create_vnf_instantiate(self
):
176 "project_id": [self
.vnfd_project
],
180 "vnfInstanceId": self
.vnfr_id
,
181 "lcmOperationType": "instantiate",
182 "vnfName": "vnf_instance_name",
183 "vnfDescription": "vnf instance description",
184 "vnfId": self
.vnfd_id
,
185 "vimAccountId": self
.vim_id
,
189 vnflcmop_id
, _
= self
.vnflcmop_topic
.new(
190 rollback
, session
, indata
, kwargs
=None, headers
=headers
192 vnflcmop_info
= self
.db
.get_one("nslcmops")
195 vnflcmop_info
["_id"],
196 "Mismatch between return id and database '_id'",
199 vnflcmop_info
["lcmOperationType"] == "instantiate",
200 "Database record must contain 'lcmOperationType=instantiate'",
203 def test_show_vnflmcop(self
):
208 "project_id": [self
.vnfd_project
],
211 self
.db
.create_list("nslcmops", yaml
.load(db_nslcmops_text
, Loader
=yaml
.Loader
))
213 actual_lcmop
= self
.db
.get_list("nslcmops")[0]
214 id = actual_lcmop
["_id"]
215 vnfr
= self
.db
.get_list("vnfrs")[0]
216 vnfr_id
= vnfr
["_id"]
217 vnflcmop
= self
.vnflcmop_topic
.show(session
, id, filter_q
)
218 _id
= vnflcmop
["vnfInstanceId"]
222 "Mismatch between vnflcmop's vnfInstanceId and database vnfr's id",