Coverage for osm_nbi/osm_vnfm/vnf_instance_actions.py: 82%

90 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-04-12 20:04 +0000

1# Copyright 2021 K Sai Kiran (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__ = "K Sai Kiran <saikiran.k@tataelxsi.co.in>, Selvi Jayaraman <selvi.j@tataelxsi.co.in>" 

17__date__ = "$12-June-2021 8:30:59$" 

18 

19from osm_nbi.instance_topics import NsrTopic, NsLcmOpTopic, VnfrTopic 

20from .base_methods import BaseMethod 

21 

22 

23class VnfLcmOp2NsLcmOp: 

24 def __init__(self, db, fs, msg, auth): 

25 """ 

26 Constructor of Vnf lcm op to Ns lcm op 

27 """ 

28 self.new_vnf_lcmop = NewVnfLcmOp(db, fs, msg, auth) 

29 self.list_vnf_lcmop = ListVnfLcmOp(db, fs, msg, auth) 

30 self.show_vnf_lcmop = ShowVnfLcmOp(db, fs, msg, auth) 

31 

32 def new(self, rollback, session, indata=None, kwargs=None, headers=None): 

33 """ 

34 Creates a new entry into database. 

35 :param rollback: list to append created items at database in case a rollback may to be done 

36 :param session: contains "username", "admin", "force", "public", "project_id", "set_project" 

37 :param indata: data to be inserted 

38 :param kwargs: used to override the indata descriptor 

39 :param headers: http request headers 

40 :return: _id, op_id: 

41 _id: identity of the inserted data. 

42 op_id: operation id if this is asynchronous, None otherwise 

43 """ 

44 return self.new_vnf_lcmop.action(rollback, session, indata, kwargs, headers) 

45 

46 def list(self, session, filter_q=None, api_req=False): 

47 """ 

48 Get a list of the Vnf Lcm Operation that matches a filter 

49 :param session: contains the used login username and working project 

50 :param filter_q: filter of data to be applied 

51 :param api_req: True if this call is serving an external API request. False if serving internal request. 

52 :return: The list, it can be empty if no one match the filter. 

53 """ 

54 return self.list_vnf_lcmop.action(session, filter_q, api_req) 

55 

56 def show(self, session, _id, api_req=False): 

57 """ 

58 Get complete information on an Vnf Lcm Operation 

59 :param session: contains "username", "admin", "force", "public", "project_id", "set_project" 

60 :param _id: server internal id 

61 :param api_req: True if this call is serving an external API request. False if serving internal request. 

62 :return: dictionary, raise exception if not found. 

63 """ 

64 return self.show_vnf_lcmop.action(session, _id, api_req) 

65 

66 

67class NewVnfLcmOp(BaseMethod): 

68 def __init__(self, db, fs, msg, auth): 

69 """ 

70 Constructor of new Vnf Lcm Op 

71 """ 

72 super().__init__() 

73 self.msg = msg 

74 self.nslcmoptopic = NsLcmOpTopic(db, fs, msg, auth) 

75 self.nsrtopic = NsrTopic(db, fs, msg, auth) 

76 self.vnfrtopic = VnfrTopic(db, fs, msg, auth) 

77 

78 def __get_nsdid(self, session, vnf_instance_id): 

79 """ 

80 Returns a nsd id from vnf instance id. 

81 :param session: contains the used login username and working project 

82 :param vnf_instance_id: id of vnf instance 

83 :return: id of nsd id 

84 """ 

85 nsr = self.nsrtopic.show(session, vnf_instance_id) 

86 return nsr["nsd"]["_id"] 

87 

88 def __get_formatted_indata(self, session, indata): 

89 """ 

90 Returns formatted data for new vnf lcm op 

91 :param session: contains the used login username and working project 

92 :param indata: contains information for new lcm operation. 

93 :return: formatted indata for new lcm op. 

94 """ 

95 formatted_indata = {} 

96 if indata["lcmOperationType"] == "instantiate": 

97 formatted_indata = { 

98 "nsName": indata["vnfName"], 

99 "nsDescription": indata["vnfDescription"], 

100 "nsdId": self.__get_nsdid(session, indata["vnfInstanceId"]), 

101 "vimAccountId": indata["vimAccountId"], 

102 "nsr_id": indata["vnfInstanceId"], 

103 "lcmOperationType": indata["lcmOperationType"], 

104 "nsInstanceId": indata["vnfInstanceId"], 

105 } 

106 elif indata["lcmOperationType"] == "terminate": 

107 formatted_indata = { 

108 "lcmOperationType": indata["lcmOperationType"], 

109 "nsInstanceId": indata["vnfInstanceId"], 

110 } 

111 elif indata["lcmOperationType"] == "scale": 

112 formatted_indata = { 

113 "lcmOperationType": indata["lcmOperationType"], 

114 "nsInstanceId": indata["vnfInstanceId"], 

115 "scaleType": "SCALE_VNF", 

116 "scaleVnfData": { 

117 "scaleVnfType": indata["type"], 

118 "scaleByStepData": { 

119 "scaling-group-descriptor": indata["aspectId"], 

120 "member-vnf-index": indata["additionalParams"][ 

121 "member-vnf-index" 

122 ], 

123 }, 

124 }, 

125 } 

126 elif indata["lcmOperationType"] == "action": 

127 formatted_indata = { 

128 "lcmOperationType": indata["lcmOperationType"], 

129 "nsInstanceId": indata["vnfInstanceId"], 

130 "member_vnf_index": indata["member_vnf_index"], 

131 "primitive": indata["primitive"], 

132 "primitive_params": indata["primitive_params"], 

133 } 

134 return formatted_indata 

135 

136 def notify_operation(self, session, _id, lcm_operation, op_id): 

137 """ 

138 Formats the operation message params and sends to kafka 

139 :param session: contains "username", "admin", "force", "public", "project_id", "set_project" 

140 :param _id: vnf instance id 

141 :param lcm_operation: lcm operation type of a VNF (instantiate, scale, terminate) 

142 :param op_id: lcm operation id of a VNF 

143 :return: None 

144 """ 

145 vnfInstanceId = _id 

146 operation = lcm_operation 

147 nslcmop_rec = self.nslcmoptopic.show(session, op_id) 

148 operation_status = nslcmop_rec["operationState"] 

149 vnfr = self.vnfrtopic.show(session, vnfInstanceId) 

150 links = { 

151 "self": "/osm/vnflcm/v1/vnf_lcm_op_occs/" + op_id, 

152 "vnfInstance": "/osm/vnflcm/v1/vnf_instances/" + vnfInstanceId, 

153 } 

154 params = { 

155 "vnfdId": vnfr["vnfd-ref"], 

156 "vnfInstanceId": vnfInstanceId, 

157 "operationState": operation_status, 

158 "vnfLcmOpOccId": op_id, 

159 "_links": links, 

160 } 

161 self.msg.write("vnf", operation, params) 

162 return None 

163 

164 def action(self, rollback, session, indata=None, kwargs=None, headers=None): 

165 """ 

166 Creates an new lcm operation. 

167 :param rollback: list to append the created items at database in case a rollback must be done 

168 :param session: contains "username", "admin", "force", "public", "project_id", "set_project" 

169 :param indata: params to be used for the nsr 

170 :param kwargs: used to override the indata 

171 :param headers: http request headers 

172 :return: id of new lcm operation. 

173 """ 

174 vnfInstanceId = indata["vnfInstanceId"] 

175 lcm_operation = indata["lcmOperationType"] 

176 vnfr = self.vnfrtopic.show(session, vnfInstanceId) 

177 indata["vnfInstanceId"] = vnfr.get("nsr-id-ref") 

178 indata = self.__get_formatted_indata(session, indata) 

179 op_id, nsName, _ = self.nslcmoptopic.new( 

180 rollback, session, indata, kwargs, headers 

181 ) 

182 self.notify_operation(session, vnfInstanceId, lcm_operation, op_id) 

183 return op_id, nsName, _ 

184 

185 

186class ListVnfLcmOp(BaseMethod): 

187 def __init__(self, db, fs, msg, auth): 

188 """ 

189 Constructor call for listing vnf lcm operations 

190 """ 

191 super().__init__() 

192 self.nslcmoptopic = NsLcmOpTopic(db, fs, msg, auth) 

193 self.nsrtopic = NsrTopic(db, fs, msg, auth) 

194 

195 def action(self, session, filter_q=None, api_req=False): 

196 """ 

197 To get list of vnf lcm operations that matches a filter 

198 :param session: contains the used login username and working project 

199 :param filter_q: filter of data to be applied 

200 :param api_req: True if this call is serving an external API request. False if serving internal request. 

201 :return: The list, it can be empty if no one match the filter. 

202 """ 

203 list = [] 

204 records = self.nslcmoptopic.list(session, filter_q, api_req) 

205 for record in records: 

206 ns_id = record.get("nsInstanceId") 

207 nsr = self.nsrtopic.show(session, ns_id) 

208 vnfInstance_id = nsr["constituent-vnfr-ref"][0] 

209 outdata = sol003_projection(record, vnfInstance_id) 

210 list.append(outdata) 

211 return list 

212 

213 

214class ShowVnfLcmOp(BaseMethod): 

215 def __init__(self, db, fs, msg, auth): 

216 """ 

217 Constructor call for showing vnf lcm operation 

218 """ 

219 super().__init__() 

220 self.nslcmoptopic = NsLcmOpTopic(db, fs, msg, auth) 

221 self.nsrtopic = NsrTopic(db, fs, msg, auth) 

222 

223 def action(self, session, _id, api_req=False): 

224 """ 

225 Get complete information on an Vnf Lcm Operation. 

226 :param session: contains "username", "admin", "force", "public", "project_id", "set_project" 

227 :param _id: Vnf Lcm operation id 

228 :param api_req: True if this call is serving an external API request. False if serving internal request. 

229 :return: dictionary, raise exception if not found. 

230 """ 

231 record = self.nslcmoptopic.show(session, _id, api_req) 

232 ns_id = record.get("nsInstanceId") 

233 nsr = self.nsrtopic.show(session, ns_id) 

234 vnfinstance_id = nsr["constituent-vnfr-ref"][0] 

235 outdata = sol003_projection(record, vnfinstance_id) 

236 return outdata 

237 

238 

239def sol003_projection(data, vnfinstance_id): 

240 """ 

241 Returns SOL003 formatted data 

242 :param data: contains Lcm Operation information 

243 :param vnfinstance_id: id of vnf_instance 

244 :return: SOL003 formatted data of vnf lcm op 

245 """ 

246 data.pop("nsInstanceId") 

247 data.pop("operationParams") 

248 data.pop("links") 

249 links = { 

250 "self": "/osm/vnflcm/v1/vnf_lcm_op_occs/" + data["_id"], 

251 "vnfInstance": "/osm/vnflcm/v1/vnf_instances/" + vnfinstance_id, 

252 } 

253 data["_links"] = links 

254 data["vnfInstanceId"] = vnfinstance_id 

255 return data