Coverage for RO-SDN-juniper_contrail/osm_rosdn_juniper_contrail/sdn_api.py: 46%
154 statements
« prev ^ index » next coverage.py v6.4.1, created at 2024-06-28 08:30 +0000
1# Copyright 2020 ETSI
2#
3# All Rights Reserved.
4#
5# Licensed under the Apache License, Version 2.0 (the "License"); you may
6# not use this file except in compliance with the License. You may obtain
7# a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14# License for the specific language governing permissions and limitations
15# under the License.
17import json
18import logging
20from osm_ro_plugin.sdnconn import SdnConnectorError
21from osm_rosdn_juniper_contrail.rest_lib import ContrailHttp
22from osm_rosdn_juniper_contrail.rest_lib import NotFound
24# from osm_rosdn_juniper_contrail.rest_lib import DuplicateFound
25# from osm_rosdn_juniper_contrail.rest_lib import HttpException
28class UnderlayApi:
29 """Class with CRUD operations for the underlay API"""
31 def __init__(self, url, config=None, user=None, password=None, logger=None):
32 self.logger = logger or logging.getLogger("ro.sdn.junipercontrail.sdnapi")
33 self.controller_url = url
35 if not url:
36 raise SdnConnectorError("'url' must be provided")
38 if not url.startswith("http"):
39 url = "http://" + url
41 if not url.endswith("/"):
42 url = url + "/"
44 self.url = url
45 self.auth_url = None
46 self.project = None
47 self.domain = None
48 self.asn = None
49 self.fabric = None
51 if config:
52 self.auth_url = config.get("auth_url")
53 self.project = config.get("project")
54 self.domain = config.get("domain")
55 self.asn = config.get("asn")
56 self.fabric = config.get("fabric")
57 self.verify = config.get("verify")
59 # Init http headers for all requests
60 self.http_header = {"Content-Type": "application/json"}
62 if user:
63 self.user = user
65 if password:
66 self.password = password
68 self.logger.debug(
69 "Config parameters for the underlay controller: auth_url: {}, project: {},"
70 " domain: {}, user: {}, password: {}".format(
71 self.auth_url, self.project, self.domain, self.user, self.password
72 )
73 )
75 auth_dict = {}
76 auth_dict["auth"] = {}
77 auth_dict["auth"]["scope"] = {}
78 auth_dict["auth"]["scope"]["project"] = {}
79 auth_dict["auth"]["scope"]["project"]["domain"] = {}
80 auth_dict["auth"]["scope"]["project"]["domain"]["id"] = self.domain
81 auth_dict["auth"]["scope"]["project"]["name"] = self.project
82 auth_dict["auth"]["identity"] = {}
83 auth_dict["auth"]["identity"]["methods"] = ["password"]
84 auth_dict["auth"]["identity"]["password"] = {}
85 auth_dict["auth"]["identity"]["password"]["user"] = {}
86 auth_dict["auth"]["identity"]["password"]["user"]["name"] = self.user
87 auth_dict["auth"]["identity"]["password"]["user"]["password"] = self.password
88 auth_dict["auth"]["identity"]["password"]["user"]["domain"] = {}
89 auth_dict["auth"]["identity"]["password"]["user"]["domain"]["id"] = self.domain
90 self.auth_dict = auth_dict
92 # Init http lib
93 auth_info = {"auth_url": self.auth_url, "auth_dict": auth_dict}
94 self.http = ContrailHttp(auth_info, self.logger, self.verify)
96 def check_auth(self):
97 response = self.http.get_cmd(url=self.auth_url, headers=self.http_header)
99 return response
101 # Helper methods for CRUD operations
102 def get_all_by_type(self, controller_url, type):
103 endpoint = controller_url + type
104 response = self.http.get_cmd(url=endpoint, headers=self.http_header)
106 return response.get(type)
108 def get_by_uuid(self, type, uuid):
109 try:
110 endpoint = self.controller_url + type + "/{}".format(uuid)
111 response = self.http.get_cmd(url=endpoint, headers=self.http_header)
113 return response.get(type)
114 except NotFound:
115 return None
117 def delete_by_uuid(self, controller_url, type, uuid):
118 endpoint = controller_url + type + "/{}".format(uuid)
119 self.http.delete_cmd(url=endpoint, headers=self.http_header)
121 def get_uuid_from_fqname(self, type, fq_name):
122 """
123 Obtain uuid from fqname
124 Returns: If resource not found returns None
125 In case of error raises an Exception
126 """
127 payload = {"type": type, "fq_name": fq_name}
129 try:
130 endpoint = self.controller_url + "fqname-to-id"
131 resp = self.http.post_cmd(
132 url=endpoint, headers=self.http_header, post_fields_dict=payload
133 )
135 return json.loads(resp).get("uuid")
136 except NotFound:
137 return None
139 def get_by_fq_name(self, type, fq_name):
140 # Obtain uuid by fqdn and then get data by uuid
141 uuid = self.get_uuid_from_fqname(type, fq_name)
143 if uuid:
144 return self.get_by_uuid(type, uuid)
145 else:
146 return None
148 def delete_ref(self, type, uuid, ref_type, ref_uuid, ref_fq_name):
149 payload = {
150 "type": type,
151 "uuid": uuid,
152 "ref-type": ref_type,
153 "ref-fq-name": ref_fq_name,
154 "operation": "DELETE",
155 }
156 endpoint = self.controller_url + "ref-update"
157 resp = self.http.post_cmd(
158 url=endpoint, headers=self.http_header, post_fields_dict=payload
159 )
161 return resp
163 # Aux methods to avoid code duplication of name conventions
164 def get_vpg_name(self, switch_id, switch_port):
165 return "{}_{}".format(switch_id, switch_port).replace(":", "_")
167 def get_vmi_name(self, switch_id, switch_port, vlan):
168 return "{}_{}-{}".format(switch_id, switch_port, vlan).replace(":", "_")
170 # Virtual network operations
172 def create_virtual_network(self, name, vni):
173 self.logger.debug("create vname, name: {}, vni: {}".format(name, vni))
174 routetarget = "{}:{}".format(self.asn, vni)
175 vnet_dict = {
176 "virtual-network": {
177 "virtual_network_properties": {
178 "vxlan_network_identifier": vni,
179 },
180 "parent_type": "project",
181 "fq_name": [self.domain, self.project, name],
182 "route_target_list": {"route_target": ["target:" + routetarget]},
183 }
184 }
185 endpoint = self.controller_url + "virtual-networks"
186 resp = self.http.post_cmd(
187 url=endpoint, headers=self.http_header, post_fields_dict=vnet_dict
188 )
190 if not resp:
191 raise SdnConnectorError("Error creating virtual network: empty response")
193 vnet_info = json.loads(resp)
194 self.logger.debug("created vnet, vnet_info: {}".format(vnet_info))
196 return vnet_info.get("virtual-network").get("uuid"), vnet_info.get(
197 "virtual-network"
198 )
200 def get_virtual_networks(self):
201 return self.get_all_by_type("virtual-networks")
203 def get_virtual_network(self, network_id):
204 return self.get_by_uuid("virtual-network", network_id)
206 def delete_virtual_network(self, network_id):
207 self.logger.debug("delete vnet uuid: {}".format(network_id))
208 self.delete_by_uuid(self.controller_url, "virtual-network", network_id)
209 self.logger.debug("deleted vnet uuid: {}".format(network_id))
211 # Vpg operations
213 def create_vpg(self, switch_id, switch_port):
214 self.logger.debug(
215 "create vpg, switch_id: {}, switch_port: {}".format(switch_id, switch_port)
216 )
217 vpg_name = self.get_vpg_name(switch_id, switch_port)
218 vpg_dict = {
219 "virtual-port-group": {
220 "parent_type": "fabric",
221 "fq_name": ["default-global-system-config", self.fabric, vpg_name],
222 }
223 }
224 endpoint = self.controller_url + "virtual-port-groups"
225 resp = self.http.post_cmd(
226 url=endpoint, headers=self.http_header, post_fields_dict=vpg_dict
227 )
229 if not resp:
230 raise SdnConnectorError("Error creating virtual port group: empty response")
232 vpg_info = json.loads(resp)
233 self.logger.debug("created vpg, vpg_info: {}".format(vpg_info))
235 return vpg_info.get("virtual-port-group").get("uuid"), vpg_info.get(
236 "virtual-port-group"
237 )
239 def get_vpgs(self):
240 return self.get_all_by_type(self.controller_url, "virtual-port-groups")
242 def get_vpg(self, vpg_id):
243 return self.get_by_uuid(self.controller_url, "virtual-port-group", vpg_id)
245 def get_vpg_by_name(self, vpg_name):
246 fq_name = ["default-global-system-config", self.fabric, vpg_name]
248 return self.get_by_fq_name("virtual-port-group", fq_name)
250 def delete_vpg(self, vpg_id):
251 self.logger.debug("delete vpg, uuid: {}".format(vpg_id))
252 self.delete_by_uuid(self.controller_url, "virtual-port-group", vpg_id)
253 self.logger.debug("deleted vpg, uuid: {}".format(vpg_id))
255 def create_vmi(self, switch_id, switch_port, network, vlan):
256 self.logger.debug(
257 "create vmi, switch_id: {}, switch_port: {}, network: {}, vlan: {}".format(
258 switch_id, switch_port, network, vlan
259 )
260 )
261 vmi_name = self.get_vmi_name(switch_id, switch_port, vlan)
262 vpg_name = self.get_vpg_name(switch_id, switch_port)
263 profile_dict = {
264 "local_link_information": [
265 {
266 "port_id": switch_port.replace(":", "_"),
267 "switch_id": switch_port.replace(":", "_"),
268 "switch_info": switch_id,
269 "fabric": self.fabric,
270 }
271 ]
272 }
273 vmi_dict = {
274 "virtual-machine-interface": {
275 "parent_type": "project",
276 "fq_name": [self.domain, self.project, vmi_name],
277 "virtual_network_refs": [{"to": [self.domain, self.project, network]}],
278 "virtual_machine_interface_properties": {
279 "sub_interface_vlan_tag": vlan
280 },
281 "virtual_machine_interface_bindings": {
282 "key_value_pair": [
283 {"key": "vnic_type", "value": "baremetal"},
284 {"key": "vif_type", "value": "vrouter"},
285 {"key": "vpg", "value": vpg_name},
286 {"key": "profile", "value": json.dumps(profile_dict)},
287 ]
288 },
289 }
290 }
291 endpoint = self.controller_url + "virtual-machine-interfaces"
292 self.logger.debug("vmi_dict: {}".format(vmi_dict))
293 resp = self.http.post_cmd(
294 url=endpoint,
295 headers=self.http_header,
296 post_fields_dict=vmi_dict,
297 )
299 if not resp:
300 raise SdnConnectorError("Error creating vmi: empty response")
302 vmi_info = json.loads(resp)
303 self.logger.debug("created vmi, info: {}".format(vmi_info))
305 return vmi_info.get("virtual-machine-interface").get("uuid"), vmi_info.get(
306 "virtual-machine-interface"
307 )
309 def get_vmi(self, vmi_uuid):
310 return self.get_by_uuid(
311 self.controller_url, "virtual-machine-interface", vmi_uuid
312 )
314 def delete_vmi(self, uuid):
315 self.logger.debug("delete vmi uuid: {}".format(uuid))
316 self.delete_by_uuid(self.controller_url, "virtual-machine-interface", uuid)
317 self.logger.debug("deleted vmi: {}".format(uuid))
319 def unref_vmi_vpg(self, vpg_id, vmi_id, vmi_fq_name):
320 self.delete_ref(
321 "virtual-port-group",
322 vpg_id,
323 "virtual-machine-interface",
324 vmi_id,
325 vmi_fq_name,
326 )