1 # Copyright 2019 AT&T Intellectual Property. All rights reserved.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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 implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 from robot.api.deco import keyword
17 from robot.libraries.BuiltIn import BuiltIn
20 from ONAPLibrary.Utilities import Utilities
23 class BaseOpenstackKeywords(object):
24 """SO is an ONAP testing library for Robot Framework that provides functionality for interacting with the serivce
28 super(BaseOpenstackKeywords, self).__init__()
29 self._cache = robot.utils.ConnectionCache('No connections created')
30 self.application_id = "robot-ete"
31 self.uuid = Utilities()
32 self.builtin = BuiltIn()
35 def save_openstack_auth(self, alias, response, token, version='v2.0'):
36 """Save Openstack Auth takes in an openstack auth response and saves it to allow easy retrival of token
37 and service catalog"""
38 self.builtin.log('Creating connection: %s' % alias, 'DEBUG')
39 json_response = json.loads(response)
40 json_response['auth_token'] = token
41 json_response['keystone_api_version'] = version
42 self._cache.register(json_response, alias=alias)
45 def get_openstack_token(self, alias):
46 """Get Openstack auth token from the current alias"""
47 response = self._cache.switch(alias)
48 if isinstance(response, str):
49 json_response = json.loads(response)
51 json_response = response
52 if json_response['keystone_api_version'] == 'v2.0':
53 return json_response['access']['token']['id']
55 return json_response['auth_token']
58 def get_openstack_catalog(self, alias):
59 """Get Openstack service catalog from the current alias"""
60 response = self._cache.switch(alias)
61 if isinstance(response, str):
62 json_response = json.loads(response)
64 json_response = response
65 if json_response['keystone_api_version'] == 'v2.0':
66 return json_response['access']['serviceCatalog']
68 return json_response['token']['catalog']
71 def get_current_openstack_tenant(self, alias):
72 """Get Openstack tenant from the current alias"""
73 response = self._cache.switch(alias)
74 if isinstance(response, str):
75 json_response = json.loads(response)
77 json_response = response
78 if json_response['keystone_api_version'] == 'v2.0':
79 return json_response['access']['token']['tenant']
81 return json_response['token']['project']
84 def get_current_openstack_tenant_id(self, alias):
85 """Get Openstack tenant id from the current alias"""
86 tenant = self.get_current_openstack_tenant(alias)
90 def get_openstack_regions(self, alias):
91 """Get all Openstack regions from the current alias"""
92 response = self._cache.switch(alias)
93 if isinstance(response, str):
94 json_response = json.loads(response)
96 json_response = response
98 if json_response['keystone_api_version'] == 'v2.0':
99 resp = json_response['access']['serviceCatalog']
101 resp = json_response['token']['catalog']
102 for catalogEntry in resp:
103 list_of_endpoints = catalogEntry['endpoints']
104 for endpoint in list_of_endpoints:
105 if 'region' in endpoint:
106 if endpoint['region'] not in regions:
107 regions.append(endpoint['region'])
111 def get_openstack_service_url(self, alias, servicetype, region=None, tenant_id=None):
112 """Get Openstack service catalog from the current alias"""
113 response = self._cache.switch(alias)
114 if isinstance(response, str):
115 json_response = json.loads(response)
117 json_response = response
119 if json_response['keystone_api_version'] == 'v2.0':
120 resp = json_response['access']['serviceCatalog']
122 resp = json_response['token']['catalog']
123 for catalogEntry in resp:
124 if self.__determine_match(catalogEntry['type'], servicetype):
125 list_of_endpoints = catalogEntry['endpoints']
126 # filter out non matching regions if provided
127 list_of_endpoints[:] = [x for x in list_of_endpoints if self.__determine_match(x['region'], region)]
128 # filter out non matching tenants if provided
129 # Only provide tenant id when authorizing without qualifying with tenant id
130 # WindRiver does not return the tenantId on the endpoint in this case.
131 if tenant_id is not None:
132 list_of_endpoints[:] = [y for y in list_of_endpoints if
133 self.__determine_match(y['tenantId'], tenant_id)]
134 if json_response['keystone_api_version'] == 'v3':
135 list_of_endpoints[:] = [z for z in list_of_endpoints if
136 self.__determine_match(z['interface'], 'public')]
137 if len(list_of_endpoints) > 0:
138 if json_response['keystone_api_version'] == 'v2.0':
139 endpoint = list_of_endpoints[0]['publicURL']
141 endpoint = list_of_endpoints[0]['url']
143 self.builtin.should_not_be_empty("", "Service Endpoint Url should not be empty")
147 def __determine_match(list_item, item):
150 elif list_item == item: