1# Copyright (c) 2015 Hewlett-Packard Development Company, L.P. 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""" 16test_endpoint 17---------------------------------- 18 19Functional tests for `shade` endpoint resource. 20""" 21 22import random 23import string 24 25from openstack.cloud.exc import OpenStackCloudException 26from openstack.cloud.exc import OpenStackCloudUnavailableFeature 27from openstack.tests.functional import base 28 29 30class TestEndpoints(base.KeystoneBaseFunctionalTest): 31 32 endpoint_attributes = ['id', 'region', 'publicurl', 'internalurl', 33 'service_id', 'adminurl'] 34 35 def setUp(self): 36 super(TestEndpoints, self).setUp() 37 38 # Generate a random name for services and regions in this test 39 self.new_item_name = 'test_' + ''.join( 40 random.choice(string.ascii_lowercase) for _ in range(5)) 41 42 self.addCleanup(self._cleanup_services) 43 self.addCleanup(self._cleanup_endpoints) 44 45 def _cleanup_endpoints(self): 46 exception_list = list() 47 for e in self.operator_cloud.list_endpoints(): 48 if e.get('region') is not None and \ 49 e['region'].startswith(self.new_item_name): 50 try: 51 self.operator_cloud.delete_endpoint(id=e['id']) 52 except Exception as e: 53 # We were unable to delete a service, let's try with next 54 exception_list.append(str(e)) 55 continue 56 if exception_list: 57 # Raise an error: we must make users aware that something went 58 # wrong 59 raise OpenStackCloudException('\n'.join(exception_list)) 60 61 def _cleanup_services(self): 62 exception_list = list() 63 for s in self.operator_cloud.list_services(): 64 if s['name'] is not None and \ 65 s['name'].startswith(self.new_item_name): 66 try: 67 self.operator_cloud.delete_service(name_or_id=s['id']) 68 except Exception as e: 69 # We were unable to delete a service, let's try with next 70 exception_list.append(str(e)) 71 continue 72 if exception_list: 73 # Raise an error: we must make users aware that something went 74 # wrong 75 raise OpenStackCloudException('\n'.join(exception_list)) 76 77 def test_create_endpoint(self): 78 service_name = self.new_item_name + '_create' 79 80 service = self.operator_cloud.create_service( 81 name=service_name, type='test_type', 82 description='this is a test description') 83 84 endpoints = self.operator_cloud.create_endpoint( 85 service_name_or_id=service['id'], 86 public_url='http://public.test/', 87 internal_url='http://internal.test/', 88 admin_url='http://admin.url/', 89 region=service_name) 90 91 self.assertNotEqual([], endpoints) 92 self.assertIsNotNone(endpoints[0].get('id')) 93 94 # Test None parameters 95 endpoints = self.operator_cloud.create_endpoint( 96 service_name_or_id=service['id'], 97 public_url='http://public.test/', 98 region=service_name) 99 100 self.assertNotEqual([], endpoints) 101 self.assertIsNotNone(endpoints[0].get('id')) 102 103 def test_update_endpoint(self): 104 ver = self.operator_cloud.config.get_api_version('identity') 105 if ver.startswith('2'): 106 # NOTE(SamYaple): Update endpoint only works with v3 api 107 self.assertRaises(OpenStackCloudUnavailableFeature, 108 self.operator_cloud.update_endpoint, 109 'endpoint_id1') 110 else: 111 service = self.operator_cloud.create_service( 112 name='service1', type='test_type') 113 endpoint = self.operator_cloud.create_endpoint( 114 service_name_or_id=service['id'], 115 url='http://admin.url/', 116 interface='admin', 117 region='orig_region', 118 enabled=False)[0] 119 120 new_service = self.operator_cloud.create_service( 121 name='service2', type='test_type') 122 new_endpoint = self.operator_cloud.update_endpoint( 123 endpoint.id, 124 service_name_or_id=new_service.id, 125 url='http://public.url/', 126 interface='public', 127 region='update_region', 128 enabled=True) 129 130 self.assertEqual(new_endpoint.url, 'http://public.url/') 131 self.assertEqual(new_endpoint.interface, 'public') 132 self.assertEqual(new_endpoint.region, 'update_region') 133 self.assertEqual(new_endpoint.service_id, new_service.id) 134 self.assertTrue(new_endpoint.enabled) 135 136 def test_list_endpoints(self): 137 service_name = self.new_item_name + '_list' 138 139 service = self.operator_cloud.create_service( 140 name=service_name, type='test_type', 141 description='this is a test description') 142 143 endpoints = self.operator_cloud.create_endpoint( 144 service_name_or_id=service['id'], 145 public_url='http://public.test/', 146 internal_url='http://internal.test/', 147 region=service_name) 148 149 observed_endpoints = self.operator_cloud.list_endpoints() 150 found = False 151 for e in observed_endpoints: 152 # Test all attributes are returned 153 for endpoint in endpoints: 154 if e['id'] == endpoint['id']: 155 found = True 156 self.assertEqual(service['id'], e['service_id']) 157 if 'interface' in e: 158 if e['interface'] == 'internal': 159 self.assertEqual('http://internal.test/', e['url']) 160 elif e['interface'] == 'public': 161 self.assertEqual('http://public.test/', e['url']) 162 else: 163 self.assertEqual('http://public.test/', 164 e['publicurl']) 165 self.assertEqual('http://internal.test/', 166 e['internalurl']) 167 self.assertEqual(service_name, e['region']) 168 169 self.assertTrue(found, msg='new endpoint not found in endpoints list!') 170 171 def test_delete_endpoint(self): 172 service_name = self.new_item_name + '_delete' 173 174 service = self.operator_cloud.create_service( 175 name=service_name, type='test_type', 176 description='this is a test description') 177 178 endpoints = self.operator_cloud.create_endpoint( 179 service_name_or_id=service['id'], 180 public_url='http://public.test/', 181 internal_url='http://internal.test/', 182 region=service_name) 183 184 self.assertNotEqual([], endpoints) 185 for endpoint in endpoints: 186 self.operator_cloud.delete_endpoint(endpoint['id']) 187 188 observed_endpoints = self.operator_cloud.list_endpoints() 189 found = False 190 for e in observed_endpoints: 191 for endpoint in endpoints: 192 if e['id'] == endpoint['id']: 193 found = True 194 break 195 self.failUnlessEqual( 196 False, found, message='new endpoint was not deleted!') 197