Coverage for osm_nbi/tests/test_osm_vnfm.py: 100%

101 statements  

« prev     ^ index     » next       coverage.py v7.3.1, created at 2024-06-30 10:14 +0000

1# Copyright 2021 Selvi Jayaraman (Tata Elxsi) 

2# 

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 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

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 

12# implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15 

16__author__ = "Selvi Jayaraman <selvi.j@tataelxsi.co.in>" 

17 

18import unittest 

19from uuid import uuid4 

20from unittest.mock import Mock, patch, mock_open 

21from osm_common.dbmemory import DbMemory 

22from osm_common.fsbase import FsBase 

23from osm_common.msgbase import MsgBase 

24from osm_nbi.vnf_instance_topics import VnfInstances, VnfLcmOpTopic 

25from osm_nbi.instance_topics import NsrTopic 

26from osm_nbi.tests.test_db_descriptors import ( 

27 db_vim_accounts_text, 

28 db_vnfm_vnfd_text, 

29 db_nsds_text, 

30 db_nsrs_text, 

31 db_vnfrs_text, 

32 db_nslcmops_text, 

33) 

34import yaml 

35 

36 

37class TestVnfInstances(unittest.TestCase): 

38 def setUp(self): 

39 self.db = DbMemory() 

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] 

49 

50 self.vim = self.db.get_list("vim_accounts")[0] 

51 self.vim_id = self.vim["_id"] 

52 

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): 

56 session = { 

57 "force": True, 

58 "admin": False, 

59 "public": False, 

60 "project_id": [self.vnfd_project], 

61 "method": "write", 

62 } 

63 indata = { 

64 "vnfdId": self.vnfd_id, 

65 "vnfInstanceName": "vnf_instance_name", 

66 "vnfInstanceDescription": "vnf instance description", 

67 "vimAccountId": self.vim_id, 

68 "additionalParams": { 

69 "virtual-link-desc": [{"id": "mgmt-net", "mgmt-network": True}], 

70 "constituent-cpd-id": "vnf-cp0-ext", 

71 "virtual-link-profile-id": "mgmt-net", 

72 }, 

73 } 

74 rollback = [] 

75 self.fs.path = "" 

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" 

80 ) 

81 

82 vnfr_id, _ = self.vnfinstances.new( 

83 rollback, session, indata, {}, headers={"Content-Type": []} 

84 ) 

85 vnfr = self.db.get_one("vnfrs") 

86 self.assertEqual( 

87 vnfr_id, vnfr["id"], "Mismatch between return id and database id" 

88 ) 

89 self.assertEqual( 

90 "NOT_INSTANTIATED", 

91 vnfr["_admin"]["nsState"], 

92 "Database record must contain 'nsState' NOT_INSTANTIATED", 

93 ) 

94 self.assertEqual( 

95 self.vnfd_id, 

96 vnfr["vnfd-ref"], 

97 "vnfr record is not properly created for the given vnfd", 

98 ) 

99 

100 def test_show_vnfinstance(self): 

101 session = { 

102 "force": False, 

103 "admin": False, 

104 "public": False, 

105 "project_id": [self.vnfd_project], 

106 "method": "write", 

107 } 

108 filter_q = {} 

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) 

113 self.assertEqual( 

114 actual_vnfr["_id"], 

115 expected_vnfr["_id"], 

116 "Mismatch between return vnfr Id and database vnfr Id", 

117 ) 

118 

119 def test_delete_vnfinstance(self): 

120 session = { 

121 "force": False, 

122 "admin": False, 

123 "public": False, 

124 "project_id": [self.vnfd_project], 

125 "method": "delete", 

126 } 

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)) 

130 

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() 

135 

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") 

139 

140 

141class TestVnfLcmOpTopic(unittest.TestCase): 

142 def setUp(self): 

143 self.db = DbMemory() 

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()) 

148 

149 self.vnflcmop_topic = VnfLcmOpTopic(self.db, self.fs, self.msg, None) 

150 self.vnflcmop_topic.check_quota = Mock(return_value=None) # skip quota 

151 

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)) 

157 

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"] 

162 

163 self.vnfd_project = self.vnfd["_admin"]["projects_read"][0] 

164 

165 self.vim = self.db.get_list("vim_accounts")[0] 

166 self.vim_id = self.vim["_id"] 

167 

168 def test_create_vnf_instantiate(self): 

169 session = { 

170 "force": False, 

171 "admin": False, 

172 "public": False, 

173 "project_id": [self.vnfd_project], 

174 "method": "write", 

175 } 

176 indata = { 

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, 

183 } 

184 rollback = [] 

185 headers = {} 

186 vnflcmop_id, _ = self.vnflcmop_topic.new( 

187 rollback, session, indata, kwargs=None, headers=headers 

188 ) 

189 vnflcmop_info = self.db.get_one("nslcmops") 

190 self.assertEqual( 

191 vnflcmop_id, 

192 vnflcmop_info["_id"], 

193 "Mismatch between return id and database '_id'", 

194 ) 

195 self.assertTrue( 

196 vnflcmop_info["lcmOperationType"] == "instantiate", 

197 "Database record must contain 'lcmOperationType=instantiate'", 

198 ) 

199 

200 def test_show_vnflmcop(self): 

201 session = { 

202 "force": False, 

203 "admin": False, 

204 "public": False, 

205 "project_id": [self.vnfd_project], 

206 "method": "write", 

207 } 

208 self.db.create_list("nslcmops", yaml.safe_load(db_nslcmops_text)) 

209 filter_q = {} 

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"] 

216 self.assertEqual( 

217 _id, 

218 vnfr_id, 

219 "Mismatch between vnflcmop's vnfInstanceId and database vnfr's id", 

220 )