Coverage for osm_nbi/osm_vnfm/vnf_instances.py: 97%
99 statements
« prev ^ index » next coverage.py v7.3.1, created at 2024-06-29 09:05 +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.
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$"
19from copy import deepcopy
20from osm_nbi.descriptor_topics import NsdTopic
21from .base_methods import BaseMethod
23from osm_nbi.instance_topics import NsrTopic, VnfrTopic
26class VnfInstances2NsInstances:
27 def __init__(self, db, fs, msg, auth):
28 """
29 Constructor of Vnf Instances to Ns Instances
30 """
31 self.new_vnf_instance = NewVnfInstance(db, fs, msg, auth)
32 self.list_vnf_instance = ListVnfInstance(db, fs, msg, auth)
33 self.show_vnf_instance = ShowVnfInstance(db, fs, msg, auth)
34 self.delete_vnf_instance = DeleteVnfInstance(db, fs, msg, auth)
36 def new(self, rollback, session, indata=None, kwargs=None, headers=None):
37 """
38 Creates a new entry into database.
39 :param rollback: list to append created items at database in case a rollback may have to be done
40 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
41 :param indata: data to be inserted
42 :param kwargs: used to override the indata descriptor
43 :param headers: http request headers
44 :return: _id, op_id:
45 _id: identity of the inserted data.
46 op_id: operation id if this is asynchronous, None otherwise
47 """
48 return self.new_vnf_instance.action(rollback, session, indata, kwargs, headers)
50 def list(self, session, filter_q=None, api_req=False):
51 """
52 Get a list of the Vnfs that match a filter
53 :param session: contains the used login username and working project
54 :param filter_q: filter of data to be applied
55 :param api_req: True if this call is serving an external API request. False if serving internal request.
56 :return: The list, it can be empty if no one match the filter.
57 """
58 return self.list_vnf_instance.action(session, filter_q, api_req)
60 def show(self, session, _id, api_req=False):
61 """
62 Get complete information on an Vnf
63 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
64 :param _id: server internal id
65 :param api_req: True if this call is serving an external API request. False if serving internal request.
66 :return: dictionary, raise exception if not found.
67 """
68 return self.show_vnf_instance.action(session, _id, api_req)
70 def delete(self, session, _id, dry_run=False, not_send_msg=None):
71 """
72 Delete item by its internal _id
73 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
74 :param _id: server internal id
75 :param dry_run: make checking but do not delete
76 :param not_send_msg: To not send message (False) or store content (list) instead
77 :return: operation id (None if there is not operation), raise exception if error or not found, conflict, ...
78 """
79 return self.delete_vnf_instance.action(session, _id, dry_run, not_send_msg)
82class NewVnfInstance(BaseMethod):
83 # sample ns descriptor
84 sample_nsd = {
85 "nsd": {
86 "nsd": [
87 {
88 "df": [
89 {
90 "id": "default-df",
91 "vnf-profile": [
92 {
93 "id": 1,
94 "virtual-link-connectivity": [
95 {
96 "constituent-cpd-id": [
97 {
98 "constituent-base-element-id": 1,
99 "constituent-cpd-id": "eth0-ext",
100 }
101 ],
102 "virtual-link-profile-id": "mgmtnet",
103 }
104 ],
105 "vnfd-id": "cirros_vnfd",
106 }
107 ],
108 }
109 ],
110 "vnfd-id": ["cirros_vnfd"],
111 "description": "Generated by OSM pacakage generator",
112 "id": "cirros_2vnf_nsd",
113 "name": "cirros_2vnf_ns",
114 "short-name": "cirros_2vnf_ns",
115 "vendor": "OSM",
116 "version": "1.0",
117 }
118 ]
119 }
120 }
122 @staticmethod
123 def __get_formatted_indata(indata, nsd_id):
124 """
125 Create indata for nsd_id
126 :param indata: Contains unformatted data for new vnf instance
127 :param nsd_id: Id of nsd
128 :return: formatted indata for nsd_id
129 """
130 formatted_indata = deepcopy(indata)
131 formatted_indata["nsdId"] = nsd_id
132 formatted_indata["nsName"] = indata["vnfInstanceName"] + "-ns"
133 for invalid_key in (
134 "vnfdId",
135 "vnfInstanceName",
136 "vnfInstanceDescription",
137 "additionalParams",
138 ):
139 formatted_indata.pop(invalid_key)
140 return formatted_indata
142 def __init__(self, db, fs, msg, auth):
143 """
144 Constructor for new vnf instance
145 """
146 super().__init__()
147 self.msg = msg
148 self.nsdtopic = NsdTopic(db, fs, msg, auth)
149 self.nsrtopic = NsrTopic(db, fs, msg, auth)
151 def __get_vnfd(self):
152 # get vnfd from nfvo
153 pass
155 def __onboard_vnfd(self):
156 self.__get_vnfd()
157 pass
159 def __create_nsd(self, rollback, session, indata=None, kwargs=None, headers=None):
160 """
161 Creates new ns descriptor from a vnfd.
162 :param rollback: list to append the created items at database in case a rollback must be done
163 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
164 :param indata: params to be used for the nsr
165 :param kwargs: used to override the indata
166 :param headers: http request headers
167 :return: id of new nsd created
168 """
169 _id, *others = self.nsdtopic.new(rollback, session, {}, None, headers)
170 new_nsd = deepcopy(NewVnfInstance.sample_nsd)
171 vnf_content = {
172 "id": "default-df",
173 "vnf-profile": [
174 {
175 "id": "1",
176 "virtual-link-connectivity": [
177 {
178 "constituent-cpd-id": [
179 {
180 "constituent-base-element-id": "1",
181 "constituent-cpd-id": indata["additionalParams"][
182 "constituent-cpd-id"
183 ],
184 }
185 ],
186 "virtual-link-profile-id": indata["additionalParams"][
187 "virtual-link-profile-id"
188 ],
189 }
190 ],
191 "vnfd-id": indata["vnfdId"],
192 }
193 ],
194 }
195 vnf_profile = vnf_content["vnf-profile"][0]
196 virtual_link_connectivity = vnf_profile["virtual-link-connectivity"][0]
197 constituent_cpd_id = virtual_link_connectivity["constituent-cpd-id"][0]
198 if "ip-address" in indata["additionalParams"]:
199 constituent_cpd_id["ip-address"] = indata["additionalParams"]["ip-address"]
200 new_nsd["nsd"]["nsd"][0] = {
201 "description": indata["vnfInstanceDescription"],
202 "designer": "OSM",
203 "id": indata["vnfdId"] + "-ns",
204 "name": indata["vnfdId"] + "-ns",
205 "version": "1.0",
206 "df": [
207 vnf_content,
208 ],
209 "virtual-link-desc": indata["additionalParams"]["virtual-link-desc"],
210 "vnfd-id": [indata["vnfdId"]],
211 }
212 return _id, new_nsd
214 def __create_nsr(self, rollback, session, indata=None, kwargs=None, headers=None):
215 """
216 Creates an new ns record in database
217 :param rollback: list to append the created items at database in case a rollback must be done
218 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
219 :param indata: params to be used for the nsr
220 :param kwargs: used to override the indata
221 :param headers: http request headers
222 :return: id of new nsr
223 """
224 return self.nsrtopic.new(rollback, session, indata, kwargs, headers)
226 def __action_pre_processing(
227 self, rollback, session, indata=None, kwargs=None, headers=None
228 ):
229 """
230 Pre process for creating new vnf instance
231 :param rollback: list to append the created items at database in case a rollback must be done
232 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
233 :param indata: params to be used for the nsr
234 :param kwargs: used to override the indata
235 :param headers: http request headers
236 :return: id nsr
237 """
238 self.__onboard_vnfd()
239 nsd_id, nsd = self.__create_nsd(rollback, session, indata, kwargs, headers)
240 self.nsdtopic.upload_content(session, nsd_id, nsd, kwargs, headers)
241 formatted_indata = NewVnfInstance.__get_formatted_indata(indata, nsd_id)
242 nsr_id, _ = self.__create_nsr(
243 rollback, session, formatted_indata, kwargs, headers
244 )
245 nsr = self.nsrtopic.show(session, nsr_id)
246 vnfr_id = nsr["constituent-vnfr-ref"][0]
247 if vnfr_id:
248 links = {"vnfInstance": "/osm/vnflcm/v1/vnf_instances/" + vnfr_id}
249 indata["vnfInstanceId"] = vnfr_id
250 indata["_links"] = links
251 self.msg.write("vnf", "create", indata)
252 return vnfr_id, None
254 def action(self, rollback, session, indata=None, kwargs=None, headers=None):
255 """
256 Creates an new vnf instance
257 :param rollback: list to append the created items at database in case a rollback must be done
258 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
259 :param indata: params to be used for the nsr
260 :param kwargs: used to override the indata
261 :param headers: http request headers
262 :return: id of new vnf instance
263 """
264 return self.__action_pre_processing(rollback, session, indata, kwargs, headers)
267class ListVnfInstance(BaseMethod):
268 def __init__(self, db, fs, msg, auth):
269 """
270 Constructor call for listing vnfs
271 """
272 super().__init__()
273 self.vnfrtopic = VnfrTopic(db, fs, msg, auth)
275 def action(self, session, filter_q=None, api_req=False):
276 """
277 To get list of vnfs that matches a filter
278 :param session: contains the used login username and working project
279 :param filter_q: filter of data to be applied
280 :param api_req: True if this call is serving an external API request. False if serving internal request.
281 :return: The list, it can be empty if no one match the filter.
282 """
283 return self.vnfrtopic.list(session, filter_q, api_req)
286class ShowVnfInstance(BaseMethod):
287 def __init__(self, db, fs, msg, auth):
288 """
289 Constructor call for showing vnf lcm operation
290 """
291 super().__init__()
292 self.vnfrtopic = VnfrTopic(db, fs, msg, auth)
294 def action(self, session, _id, api_req=False):
295 """
296 Get complete information on an Vnf Lcm Operation.
297 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
298 :param _id: Vnf Lcm operation id
299 :param api_req: True if this call is serving an external API request. False if serving internal request.
300 :return: dictionary, raise exception if not found.
301 """
302 return self.vnfrtopic.show(session, _id, api_req)
305class DeleteVnfInstance(BaseMethod):
306 def __init__(self, db, fs, msg, auth):
307 """
308 Constructor call for deleting vnf
309 """
310 super().__init__()
311 self.msg = msg
312 self.nsrtopic = NsrTopic(db, fs, msg, auth)
313 self.nsdtopic = NsdTopic(db, fs, msg, auth)
314 self.vnfrtopic = VnfrTopic(db, fs, msg, auth)
316 def action(self, session, _id, dry_run=False, not_send_msg=None):
317 """
318 Delete vnf instance by its internal _id
319 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
320 :param _id: server internal id
321 :param dry_run: make checking but do not delete
322 :param not_send_msg: To not send message (False) or store content (list) instead
323 :return: operation id (None if there is not operation), raise exception if error or not found, conflict, ...
324 """
325 vnfInstanceId = _id
326 vnfr = self.vnfrtopic.show(session, vnfInstanceId)
327 ns_id = vnfr.get("nsr-id-ref")
328 nsr = self.nsrtopic.show(session, ns_id)
329 nsd_to_del = nsr["nsd"]["_id"]
330 links = {"vnfInstance": "/osm/vnflcm/v1/vnf_instances/" + _id}
331 params = {"vnfdId": vnfr["vnfd-ref"], "vnfInstanceId": _id, "_links": links}
332 self.msg.write("vnf", "delete", params)
333 self.nsrtopic.delete(session, ns_id, dry_run, not_send_msg)
334 return self.nsdtopic.delete(session, nsd_to_del, dry_run, not_send_msg)