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"
91 vnfr
["_admin"]["nsState"],
92 "Database record must contain 'nsState' NOT_INSTANTIATED",
97 "vnfr record is not properly created for the given vnfd",
100 def test_show_vnfinstance(self
):
105 "project_id": [self
.vnfd_project
],
109 self
.db
.create_list("vnfrs", yaml
.safe_load(db_vnfrs_text
))
110 actual_vnfr
= self
.db
.get_list("vnfrs")[0]
111 id = actual_vnfr
["_id"]
112 expected_vnfr
= self
.vnfinstances
.show(session
, id, filter_q
)
115 expected_vnfr
["_id"],
116 "Mismatch between return vnfr Id and database vnfr Id",
119 def test_delete_vnfinstance(self
):
124 "project_id": [self
.vnfd_project
],
127 self
.db
.create_list("vnfrs", yaml
.safe_load(db_vnfrs_text
))
128 self
.db
.create_list("nsrs", yaml
.safe_load(db_nsrs_text
))
129 self
.db
.create_list("nsds", yaml
.safe_load(db_nsds_text
))
131 self
.vnfr
= self
.db
.get_list("vnfrs")[0]
132 self
.vnfr_id
= self
.vnfr
["_id"]
133 self
.db
.set_one
= self
.db
.set_one
134 self
.db
.set_one
= Mock()
136 self
.vnfinstances
.delete(session
, self
.vnfr_id
)
137 msg_args
= self
.msg
.write
.call_args
[0]
138 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
141 class TestVnfLcmOpTopic(unittest
.TestCase
):
144 self
.fs
= Mock(FsBase())
145 self
.fs
.get_params
.return_value
= {"./fake/folder"}
146 self
.fs
.file_open
= mock_open()
147 self
.msg
= Mock(MsgBase())
149 self
.vnflcmop_topic
= VnfLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
150 self
.vnflcmop_topic
.check_quota
= Mock(return_value
=None) # skip quota
152 self
.db
.create_list("vim_accounts", yaml
.safe_load(db_vim_accounts_text
))
153 self
.db
.create_list("nsds", yaml
.safe_load(db_nsds_text
))
154 self
.db
.create_list("vnfds", yaml
.safe_load(db_vnfm_vnfd_text
))
155 self
.db
.create_list("vnfrs", yaml
.safe_load(db_vnfrs_text
))
156 self
.db
.create_list("nsrs", yaml
.safe_load(db_nsrs_text
))
158 self
.vnfd
= self
.db
.get_list("vnfds")[0]
159 self
.vnfd_id
= self
.vnfd
["_id"]
160 self
.vnfr
= self
.db
.get_list("vnfrs")[0]
161 self
.vnfr_id
= self
.vnfr
["_id"]
163 self
.vnfd_project
= self
.vnfd
["_admin"]["projects_read"][0]
165 self
.vim
= self
.db
.get_list("vim_accounts")[0]
166 self
.vim_id
= self
.vim
["_id"]
168 def test_create_vnf_instantiate(self
):
173 "project_id": [self
.vnfd_project
],
177 "vnfInstanceId": self
.vnfr_id
,
178 "lcmOperationType": "instantiate",
179 "vnfName": "vnf_instance_name",
180 "vnfDescription": "vnf instance description",
181 "vnfId": self
.vnfd_id
,
182 "vimAccountId": self
.vim_id
,
186 vnflcmop_id
, _
= self
.vnflcmop_topic
.new(
187 rollback
, session
, indata
, kwargs
=None, headers
=headers
189 vnflcmop_info
= self
.db
.get_one("nslcmops")
192 vnflcmop_info
["_id"],
193 "Mismatch between return id and database '_id'",
196 vnflcmop_info
["lcmOperationType"] == "instantiate",
197 "Database record must contain 'lcmOperationType=instantiate'",
200 def test_show_vnflmcop(self
):
205 "project_id": [self
.vnfd_project
],
208 self
.db
.create_list("nslcmops", yaml
.safe_load(db_nslcmops_text
))
210 actual_lcmop
= self
.db
.get_list("nslcmops")[0]
211 id = actual_lcmop
["_id"]
212 vnfr
= self
.db
.get_list("vnfrs")[0]
213 vnfr_id
= vnfr
["_id"]
214 vnflcmop
= self
.vnflcmop_topic
.show(session
, id, filter_q
)
215 _id
= vnflcmop
["vnfInstanceId"]
219 "Mismatch between vnflcmop's vnfInstanceId and database vnfr's id",