1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
# Copyright 2019 AT&T Intellectual Property. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import robot.utils
from robot.api.deco import keyword
from robot.libraries.BuiltIn import BuiltIn
import json
from ONAPLibrary.Utilities import Utilities
class BaseOpenstackKeywords(object):
"""SO is an ONAP testing library for Robot Framework that provides functionality for interacting with the serivce
orchestrator. """
def __init__(self):
super(BaseOpenstackKeywords, self).__init__()
self._cache = robot.utils.ConnectionCache('No connections created')
self.application_id = "robot-ete"
self.uuid = Utilities()
self.builtin = BuiltIn()
@keyword
def save_openstack_auth(self, alias, response, token, version='v2.0'):
"""Save Openstack Auth takes in an openstack auth response and saves it to allow easy retrival of token
and service catalog"""
self.builtin.log('Creating connection: %s' % alias, 'DEBUG')
json_response = json.loads(response)
json_response['auth_token'] = token
json_response['keystone_api_version'] = version
self._cache.register(json_response, alias=alias)
@keyword
def get_openstack_token(self, alias):
"""Get Openstack auth token from the current alias"""
response = self._cache.switch(alias)
if isinstance(response, str):
json_response = json.loads(response)
else:
json_response = response
if json_response['keystone_api_version'] == 'v2.0':
return json_response['access']['token']['id']
else:
return json_response['auth_token']
@keyword
def get_openstack_catalog(self, alias):
"""Get Openstack service catalog from the current alias"""
response = self._cache.switch(alias)
if isinstance(response, str):
json_response = json.loads(response)
else:
json_response = response
if json_response['keystone_api_version'] == 'v2.0':
return json_response['access']['serviceCatalog']
else:
return json_response['token']['catalog']
@keyword
def get_current_openstack_tenant(self, alias):
"""Get Openstack tenant from the current alias"""
response = self._cache.switch(alias)
if isinstance(response, str):
json_response = json.loads(response)
else:
json_response = response
if json_response['keystone_api_version'] == 'v2.0':
return json_response['access']['token']['tenant']
else:
return json_response['token']['project']
@keyword
def get_current_openstack_tenant_id(self, alias):
"""Get Openstack tenant id from the current alias"""
tenant = self.get_current_openstack_tenant(alias)
return tenant['id']
@keyword
def get_openstack_regions(self, alias):
"""Get all Openstack regions from the current alias"""
response = self._cache.switch(alias)
if isinstance(response, str):
json_response = json.loads(response)
else:
json_response = response
regions = []
if json_response['keystone_api_version'] == 'v2.0':
resp = json_response['access']['serviceCatalog']
else:
resp = json_response['token']['catalog']
for catalogEntry in resp:
list_of_endpoints = catalogEntry['endpoints']
for endpoint in list_of_endpoints:
if 'region' in endpoint:
if endpoint['region'] not in regions:
regions.append(endpoint['region'])
return regions
@keyword
def get_openstack_service_url(self, alias, servicetype, region=None, tenant_id=None):
"""Get Openstack service catalog from the current alias"""
response = self._cache.switch(alias)
if isinstance(response, str):
json_response = json.loads(response)
else:
json_response = response
endpoint = None
if json_response['keystone_api_version'] == 'v2.0':
resp = json_response['access']['serviceCatalog']
else:
resp = json_response['token']['catalog']
for catalogEntry in resp:
if self.__determine_match(catalogEntry['type'], servicetype):
list_of_endpoints = catalogEntry['endpoints']
# filter out non matching regions if provided
list_of_endpoints[:] = [x for x in list_of_endpoints if self.__determine_match(x['region'], region)]
# filter out non matching tenants if provided
# Only provide tenant id when authorizing without qualifying with tenant id
# WindRiver does not return the tenantId on the endpoint in this case.
if tenant_id is not None:
list_of_endpoints[:] = [y for y in list_of_endpoints if
self.__determine_match(y['tenantId'], tenant_id)]
if json_response['keystone_api_version'] == 'v3':
list_of_endpoints[:] = [z for z in list_of_endpoints if
self.__determine_match(z['interface'], 'public')]
if len(list_of_endpoints) > 0:
if json_response['keystone_api_version'] == 'v2.0':
endpoint = list_of_endpoints[0]['publicURL']
else:
endpoint = list_of_endpoints[0]['url']
if endpoint is None:
self.builtin.should_not_be_empty("", "Service Endpoint Url should not be empty")
return endpoint
@staticmethod
def __determine_match(list_item, item):
if item is None:
return True
elif list_item == item:
return True
else:
return False
|
