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