How to use list_endpoints method in tempest

Best Python code snippet using tempest_python

test_list_endpoints.py

Source:test_list_endpoints.py

1# Copyright (c) 2012 OpenStack Foundation2#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 at6#7#    http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or12# implied.13# See the License for the specific language governing permissions and14# limitations under the License.15import array16import json17import unittest18from tempfile import mkdtemp19from shutil import rmtree20import os21import mock22from swift.common import ring, utils23from swift.common.utils import split_path24from swift.common.swob import Request, Response25from swift.common.middleware import list_endpoints26from swift.common.storage_policy import StoragePolicy, POLICIES27from test.unit import patch_policies28class FakeApp(object):29    def __call__(self, env, start_response):30        return Response(body="FakeApp")(env, start_response)31def start_response(*args):32    pass33@patch_policies([StoragePolicy(0, 'zero', False),34                StoragePolicy(1, 'one', True)])35class TestListEndpoints(unittest.TestCase):36    def setUp(self):37        utils.HASH_PATH_SUFFIX = 'endcap'38        utils.HASH_PATH_PREFIX = ''39        self.testdir = mkdtemp()40        accountgz = os.path.join(self.testdir, 'account.ring.gz')41        containergz = os.path.join(self.testdir, 'container.ring.gz')42        objectgz = os.path.join(self.testdir, 'object.ring.gz')43        objectgz_1 = os.path.join(self.testdir, 'object-1.ring.gz')44        self.policy_to_test = 045        self.expected_path = ('v1', 'a', 'c', 'o1')46        # Let's make the rings slightly different so we can test47        # that the correct ring is consulted (e.g. we don't consult48        # the object ring to get nodes for a container)49        intended_replica2part2dev_id_a = [50            array.array('H', [3, 1, 3, 1]),51            array.array('H', [0, 3, 1, 4]),52            array.array('H', [1, 4, 0, 3])]53        intended_replica2part2dev_id_c = [54            array.array('H', [4, 3, 0, 1]),55            array.array('H', [0, 1, 3, 4]),56            array.array('H', [3, 4, 0, 1])]57        intended_replica2part2dev_id_o = [58            array.array('H', [0, 1, 0, 1]),59            array.array('H', [0, 1, 0, 1]),60            array.array('H', [3, 4, 3, 4])]61        intended_replica2part2dev_id_o_1 = [62            array.array('H', [1, 0, 1, 0]),63            array.array('H', [1, 0, 1, 0]),64            array.array('H', [4, 3, 4, 3])]65        intended_devs = [{'id': 0, 'zone': 0, 'weight': 1.0,66                          'ip': '10.1.1.1', 'port': 6000,67                          'device': 'sda1'},68                         {'id': 1, 'zone': 0, 'weight': 1.0,69                          'ip': '10.1.1.1', 'port': 6000,70                          'device': 'sdb1'},71                         None,72                         {'id': 3, 'zone': 2, 'weight': 1.0,73                          'ip': '10.1.2.1', 'port': 6000,74                          'device': 'sdc1'},75                         {'id': 4, 'zone': 2, 'weight': 1.0,76                          'ip': '10.1.2.2', 'port': 6000,77                          'device': 'sdd1'}]78        intended_part_shift = 3079        ring.RingData(intended_replica2part2dev_id_a,80                      intended_devs, intended_part_shift).save(accountgz)81        ring.RingData(intended_replica2part2dev_id_c,82                      intended_devs, intended_part_shift).save(containergz)83        ring.RingData(intended_replica2part2dev_id_o,84                      intended_devs, intended_part_shift).save(objectgz)85        ring.RingData(intended_replica2part2dev_id_o_1,86                      intended_devs, intended_part_shift).save(objectgz_1)87        self.app = FakeApp()88        self.list_endpoints = list_endpoints.filter_factory(89            {'swift_dir': self.testdir})(self.app)90    def tearDown(self):91        rmtree(self.testdir, ignore_errors=1)92    def FakeGetInfo(self, env, app, swift_source=None):93        info = {'status': 0, 'sync_key': None, 'meta': {},94                'cors': {'allow_origin': None, 'expose_headers': None,95                         'max_age': None},96                'sysmeta': {}, 'read_acl': None,97                'object_count': None, 'write_acl': None, 'versions': None,98                'bytes': None}99        info['storage_policy'] = self.policy_to_test100        (version, account, container, unused) = \101            split_path(env['PATH_INFO'], 3, 4, True)102        self.assertEqual((version, account, container),103                         self.expected_path[:3])104        return info105    def test_parse_response_version(self):106        expectations = {107            '': 1.0,  # legacy compat108            '/1': 1.0,109            '/v1': 1.0,110            '/1.0': 1.0,111            '/v1.0': 1.0,112            '/2': 2.0,113            '/v2': 2.0,114            '/2.0': 2.0,115            '/v2.0': 2.0,116        }117        accounts = (118            'AUTH_test',119            'test',120            'verybadreseller_prefix'121            'verybadaccount'122        )123        for expected_account in accounts:124            for version, expected in expectations.items():125                path = '/endpoints%s/%s/c/o' % (version, expected_account)126                req = Request.blank(path)127                version, account, container, obj = \128                    self.list_endpoints._parse_path(req)129                try:130                    self.assertEqual(version, expected)131                    self.assertEqual(account, expected_account)132                except AssertionError:133                    self.fail('Unexpected result from parse path %r: %r != %r'134                              % (path, (version, account),135                                 (expected, expected_account)))136    def test_parse_version_that_looks_like_account(self):137        """138        Demonstrate the failure mode for versions that look like accounts,139        if you can make _parse_path better and this is the *only* test that140        fails you can delete it ;)141        """142        bad_versions = (143            'v_3',144            'verybadreseller_prefix',145        )146        for bad_version in bad_versions:147            req = Request.blank('/endpoints/%s/a/c/o' % bad_version)148            version, account, container, obj = \149                self.list_endpoints._parse_path(req)150            self.assertEqual(version, 1.0)151            self.assertEqual(account, bad_version)152            self.assertEqual(container, 'a')153            self.assertEqual(obj, 'c/o')154    def test_parse_account_that_looks_like_version(self):155        """156        Demonstrate the failure mode for accounts that looks like versions,157        if you can make _parse_path better and this is the *only* test that158        fails you can delete it ;)159        """160        bad_accounts = (161            'v3.0', 'verybaddaccountwithnoprefix',162        )163        for bad_account in bad_accounts:164            req = Request.blank('/endpoints/%s/c/o' % bad_account)165            self.assertRaises(ValueError,166                              self.list_endpoints._parse_path, req)167        even_worse_accounts = {168            'v1': 1.0,169            'v2.0': 2.0,170        }171        for bad_account, guessed_version in even_worse_accounts.items():172            req = Request.blank('/endpoints/%s/c/o' % bad_account)173            version, account, container, obj = \174                self.list_endpoints._parse_path(req)175            self.assertEqual(version, guessed_version)176            self.assertEqual(account, 'c')177            self.assertEqual(container, 'o')178            self.assertEqual(obj, None)179    def test_get_object_ring(self):180        self.assertEqual(isinstance(self.list_endpoints.get_object_ring(0),181                                    ring.Ring), True)182        self.assertEqual(isinstance(self.list_endpoints.get_object_ring(1),183                                    ring.Ring), True)184        self.assertRaises(ValueError, self.list_endpoints.get_object_ring, 99)185    def test_parse_path_no_version_specified(self):186        req = Request.blank('/endpoints/a/c/o1')187        version, account, container, obj = \188            self.list_endpoints._parse_path(req)189        self.assertEqual(account, 'a')190        self.assertEqual(container, 'c')191        self.assertEqual(obj, 'o1')192    def test_parse_path_with_valid_version(self):193        req = Request.blank('/endpoints/v2/a/c/o1')194        version, account, container, obj = \195            self.list_endpoints._parse_path(req)196        self.assertEqual(version, 2.0)197        self.assertEqual(account, 'a')198        self.assertEqual(container, 'c')199        self.assertEqual(obj, 'o1')200    def test_parse_path_with_invalid_version(self):201        req = Request.blank('/endpoints/v3/a/c/o1')202        self.assertRaises(ValueError, self.list_endpoints._parse_path,203                          req)204    def test_parse_path_with_no_account(self):205        bad_paths = ('v1', 'v2', '')206        for path in bad_paths:207            req = Request.blank('/endpoints/%s' % path)208            try:209                self.list_endpoints._parse_path(req)210                self.fail('Expected ValueError to be raised')211            except ValueError as err:212                self.assertEqual(str(err), 'No account specified')213    def test_get_endpoint(self):214        # Expected results for objects taken from test_ring215        # Expected results for others computed by manually invoking216        # ring.get_nodes().217        resp = Request.blank('/endpoints/a/c/o1').get_response(218            self.list_endpoints)219        self.assertEqual(resp.status_int, 200)220        self.assertEqual(resp.content_type, 'application/json')221        self.assertEqual(json.loads(resp.body), [222            "http://10.1.1.1:6000/sdb1/1/a/c/o1",223            "http://10.1.2.2:6000/sdd1/1/a/c/o1"224        ])225        # test policies with no version endpoint name226        expected = [[227                    "http://10.1.1.1:6000/sdb1/1/a/c/o1",228                    "http://10.1.2.2:6000/sdd1/1/a/c/o1"], [229                    "http://10.1.1.1:6000/sda1/1/a/c/o1",230                    "http://10.1.2.1:6000/sdc1/1/a/c/o1"231                    ]]232        PATCHGI = 'swift.common.middleware.list_endpoints.get_container_info'233        for pol in POLICIES:234            self.policy_to_test = pol.idx235            with mock.patch(PATCHGI, self.FakeGetInfo):236                resp = Request.blank('/endpoints/a/c/o1').get_response(237                    self.list_endpoints)238            self.assertEqual(resp.status_int, 200)239            self.assertEqual(resp.content_type, 'application/json')240            self.assertEqual(json.loads(resp.body), expected[pol.idx])241        # Here, 'o1/' is the object name.242        resp = Request.blank('/endpoints/a/c/o1/').get_response(243            self.list_endpoints)244        self.assertEqual(resp.status_int, 200)245        self.assertEqual(json.loads(resp.body), [246            "http://10.1.1.1:6000/sdb1/3/a/c/o1/",247            "http://10.1.2.2:6000/sdd1/3/a/c/o1/"248        ])249        resp = Request.blank('/endpoints/a/c2').get_response(250            self.list_endpoints)251        self.assertEqual(resp.status_int, 200)252        self.assertEqual(json.loads(resp.body), [253            "http://10.1.1.1:6000/sda1/2/a/c2",254            "http://10.1.2.1:6000/sdc1/2/a/c2"255        ])256        resp = Request.blank('/endpoints/a1').get_response(257            self.list_endpoints)258        self.assertEqual(resp.status_int, 200)259        self.assertEqual(json.loads(resp.body), [260            "http://10.1.2.1:6000/sdc1/0/a1",261            "http://10.1.1.1:6000/sda1/0/a1",262            "http://10.1.1.1:6000/sdb1/0/a1"263        ])264        resp = Request.blank('/endpoints/').get_response(265            self.list_endpoints)266        self.assertEqual(resp.status_int, 400)267        resp = Request.blank('/endpoints/a/c 2').get_response(268            self.list_endpoints)269        self.assertEqual(resp.status_int, 200)270        self.assertEqual(json.loads(resp.body), [271            "http://10.1.1.1:6000/sdb1/3/a/c%202",272            "http://10.1.2.2:6000/sdd1/3/a/c%202"273        ])274        resp = Request.blank('/endpoints/a/c%202').get_response(275            self.list_endpoints)276        self.assertEqual(resp.status_int, 200)277        self.assertEqual(json.loads(resp.body), [278            "http://10.1.1.1:6000/sdb1/3/a/c%202",279            "http://10.1.2.2:6000/sdd1/3/a/c%202"280        ])281        resp = Request.blank('/endpoints/ac%20count/con%20tainer/ob%20ject') \282            .get_response(self.list_endpoints)283        self.assertEqual(resp.status_int, 200)284        self.assertEqual(json.loads(resp.body), [285            "http://10.1.1.1:6000/sdb1/3/ac%20count/con%20tainer/ob%20ject",286            "http://10.1.2.2:6000/sdd1/3/ac%20count/con%20tainer/ob%20ject"287        ])288        resp = Request.blank('/endpoints/a/c/o1', {'REQUEST_METHOD': 'POST'}) \289            .get_response(self.list_endpoints)290        self.assertEqual(resp.status_int, 405)291        self.assertEqual(resp.status, '405 Method Not Allowed')292        self.assertEqual(resp.headers['allow'], 'GET')293        resp = Request.blank('/not-endpoints').get_response(294            self.list_endpoints)295        self.assertEqual(resp.status_int, 200)296        self.assertEqual(resp.status, '200 OK')297        self.assertEqual(resp.body, 'FakeApp')298        # test policies with custom endpoint name299        for pol in POLICIES:300            # test custom path with trailing slash301            custom_path_le = list_endpoints.filter_factory({302                'swift_dir': self.testdir,303                'list_endpoints_path': '/some/another/path/'304            })(self.app)305            self.policy_to_test = pol.idx306            with mock.patch(PATCHGI, self.FakeGetInfo):307                resp = Request.blank('/some/another/path/a/c/o1') \308                    .get_response(custom_path_le)309            self.assertEqual(resp.status_int, 200)310            self.assertEqual(resp.content_type, 'application/json')311            self.assertEqual(json.loads(resp.body), expected[pol.idx])312            # test custom path without trailing slash313            custom_path_le = list_endpoints.filter_factory({314                'swift_dir': self.testdir,315                'list_endpoints_path': '/some/another/path'316            })(self.app)317            self.policy_to_test = pol.idx318            with mock.patch(PATCHGI, self.FakeGetInfo):319                resp = Request.blank('/some/another/path/a/c/o1') \320                    .get_response(custom_path_le)321            self.assertEqual(resp.status_int, 200)322            self.assertEqual(resp.content_type, 'application/json')323            self.assertEqual(json.loads(resp.body), expected[pol.idx])324    def test_v1_response(self):325        req = Request.blank('/endpoints/v1/a/c/o1')326        resp = req.get_response(self.list_endpoints)327        expected = ["http://10.1.1.1:6000/sdb1/1/a/c/o1",328                    "http://10.1.2.2:6000/sdd1/1/a/c/o1"]329        self.assertEqual(resp.body, json.dumps(expected))330    def test_v2_obj_response(self):331        req = Request.blank('/endpoints/v2/a/c/o1')332        resp = req.get_response(self.list_endpoints)333        expected = {334            'endpoints': ["http://10.1.1.1:6000/sdb1/1/a/c/o1",335                          "http://10.1.2.2:6000/sdd1/1/a/c/o1"],336            'headers': {'X-Backend-Storage-Policy-Index': "0"},337        }338        self.assertEqual(resp.body, json.dumps(expected))339        for policy in POLICIES:340            patch_path = 'swift.common.middleware.list_endpoints' \341                '.get_container_info'342            mock_get_container_info = lambda *args, **kwargs: \343                {'storage_policy': int(policy)}344            with mock.patch(patch_path, mock_get_container_info):345                resp = req.get_response(self.list_endpoints)346            part, nodes = policy.object_ring.get_nodes('a', 'c', 'o1')347            [node.update({'part': part}) for node in nodes]348            path = 'http://%(ip)s:%(port)s/%(device)s/%(part)s/a/c/o1'349            expected = {350                'headers': {351                    'X-Backend-Storage-Policy-Index': str(int(policy))},352                'endpoints': [path % node for node in nodes],353            }354            self.assertEqual(resp.body, json.dumps(expected))355    def test_v2_non_obj_response(self):356        # account357        req = Request.blank('/endpoints/v2/a')358        resp = req.get_response(self.list_endpoints)359        expected = {360            'endpoints': ["http://10.1.2.1:6000/sdc1/0/a",361                          "http://10.1.1.1:6000/sda1/0/a",362                          "http://10.1.1.1:6000/sdb1/0/a"],363            'headers': {},364        }365        # container366        self.assertEqual(resp.body, json.dumps(expected))367        req = Request.blank('/endpoints/v2/a/c')368        resp = req.get_response(self.list_endpoints)369        expected = {370            'endpoints': ["http://10.1.2.2:6000/sdd1/0/a/c",371                          "http://10.1.1.1:6000/sda1/0/a/c",372                          "http://10.1.2.1:6000/sdc1/0/a/c"],373            'headers': {},374        }375        self.assertEqual(resp.body, json.dumps(expected))376    def test_version_account_response(self):377        req = Request.blank('/endpoints/a')378        resp = req.get_response(self.list_endpoints)379        expected = ["http://10.1.2.1:6000/sdc1/0/a",380                    "http://10.1.1.1:6000/sda1/0/a",381                    "http://10.1.1.1:6000/sdb1/0/a"]382        self.assertEqual(resp.body, json.dumps(expected))383        req = Request.blank('/endpoints/v1.0/a')384        resp = req.get_response(self.list_endpoints)385        self.assertEqual(resp.body, json.dumps(expected))386        req = Request.blank('/endpoints/v2/a')387        resp = req.get_response(self.list_endpoints)388        expected = {389            'endpoints': ["http://10.1.2.1:6000/sdc1/0/a",390                          "http://10.1.1.1:6000/sda1/0/a",391                          "http://10.1.1.1:6000/sdb1/0/a"],392            'headers': {},393        }394        self.assertEqual(resp.body, json.dumps(expected))395if __name__ == '__main__':...

aws_s3outposts_info.py

Source:aws_s3outposts_info.py

...49            if client.can_paginate('list_endpoints'):50                paginator = client.get_paginator('list_endpoints')51                return paginator.paginate(), True52            else:53                return client.list_endpoints(), False54        else:55            return None, False56    except (BotoCoreError, ClientError) as e:57        module.fail_json_aws(e, msg='Failed to fetch Amazon S3 on Outposts details')58def main():59    argument_spec = dict(60        list_endpoints=dict(required=False, type=bool),61    )62    module = AnsibleAWSModule(63        argument_spec=argument_spec,64        required_if=(),65        mutually_exclusive=[],66    )67    client = module.client('s3outposts', retry_decorator=AWSRetry.exponential_backoff())...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.