robotframework-onap/ONAPLibrary/BaseOpenstackKeywords.py

   1 # Copyright 2019 AT&T Intellectual Property. All rights reserved.
   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 implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import robot.utils
  16 from robot.api.deco import keyword
  17 from robot.libraries.BuiltIn import BuiltIn
  18 import json
  19
  20 from ONAPLibrary.Utilities import Utilities
  21
  22
  23 class BaseOpenstackKeywords(object):
  24     """SO is an ONAP testing library for Robot Framework that provides functionality for interacting with the serivce
  25     orchestrator. """
  26
  27     def __init__(self):
  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()
  33
  34     @keyword
  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)
  43
  44     @keyword
  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)
  50         else:
  51             json_response = response
  52         if json_response['keystone_api_version'] == 'v2.0':
  53             return json_response['access']['token']['id']
  54         else:
  55             return json_response['auth_token']
  56
  57     @keyword
  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)
  63         else:
  64             json_response = response
  65         if json_response['keystone_api_version'] == 'v2.0':
  66             return json_response['access']['serviceCatalog']
  67         else:
  68             return json_response['token']['catalog']
  69
  70     @keyword
  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)
  76         else:
  77             json_response = response
  78         if json_response['keystone_api_version'] == 'v2.0':
  79             return json_response['access']['token']['tenant']
  80         else:
  81             return json_response['token']['project']
  82
  83     @keyword
  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)
  87         return tenant['id']
  88
  89     @keyword
  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)
  95         else:
  96             json_response = response
  97         regions = []
  98         if json_response['keystone_api_version'] == 'v2.0':
  99             resp = json_response['access']['serviceCatalog']
 100         else:
 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'])
 108         return regions
 109
 110     @keyword
 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)
 116         else:
 117             json_response = response
 118         endpoint = None
 119         if json_response['keystone_api_version'] == 'v2.0':
 120             resp = json_response['access']['serviceCatalog']
 121         else:
 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']
 140                     else:
 141                         endpoint = list_of_endpoints[0]['url']
 142         if endpoint is None:
 143             self.builtin.should_not_be_empty("", "Service Endpoint Url should not be empty")
 144         return endpoint
 145
 146     @staticmethod
 147     def __determine_match(list_item, item):
 148         if item is None:
 149             return True
 150         elif list_item == item:
 151             return True
 152         else:
 153             return False