KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use list_endpoints method in localstack
Best Python code snippet using localstack_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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
