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

90 statements  

« prev     ^ index     » next       coverage.py v7.3.1, created at 2024-07-02 10:15 +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, _ = self.nslcmoptopic.new(rollback, session, indata, kwargs, headers) 

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

181 return op_id, _ 

182 

183 

184class ListVnfLcmOp(BaseMethod): 

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

186 """ 

187 Constructor call for listing vnf lcm operations 

188 """ 

189 super().__init__() 

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

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

192 

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

194 """ 

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

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

197 :param filter_q: filter of data to be applied 

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

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

200 """ 

201 list = [] 

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

203 for record in records: 

204 ns_id = record.get("nsInstanceId") 

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

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

207 outdata = sol003_projection(record, vnfInstance_id) 

208 list.append(outdata) 

209 return list 

210 

211 

212class ShowVnfLcmOp(BaseMethod): 

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

214 """ 

215 Constructor call for showing vnf lcm operation 

216 """ 

217 super().__init__() 

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

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

220 

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

222 """ 

223 Get complete information on an Vnf Lcm Operation. 

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

225 :param _id: Vnf Lcm operation id 

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

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

228 """ 

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

230 ns_id = record.get("nsInstanceId") 

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

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

233 outdata = sol003_projection(record, vnfinstance_id) 

234 return outdata 

235 

236 

237def sol003_projection(data, vnfinstance_id): 

238 """ 

239 Returns SOL003 formatted data 

240 :param data: contains Lcm Operation information 

241 :param vnfinstance_id: id of vnf_instance 

242 :return: SOL003 formatted data of vnf lcm op 

243 """ 

244 data.pop("nsInstanceId") 

245 data.pop("operationParams") 

246 data.pop("links") 

247 links = { 

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

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

250 } 

251 data["_links"] = links 

252 data["vnfInstanceId"] = vnfinstance_id 

253 return data