How to use create_resolver_endpoint method in localstack

Best Python code snippet using localstack_python

test_route53resolver_endpoint.py

Source:test_route53resolver_endpoint.py Github

copy

Full Screen

...43 if not tags:44 tags = []45 random_num = get_random_hex(10)46 subnet_ids = create_subnets(ec2_client, create_vpc(ec2_client))47 resolver_endpoint = client.create_resolver_endpoint(48 CreatorRequestId=random_num,49 Name=name if name else "X" + random_num,50 SecurityGroupIds=[create_security_group(ec2_client)],51 Direction="INBOUND",52 IpAddresses=[53 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.200"},54 {"SubnetId": subnet_ids[1], "Ip": "10.0.0.20"},55 ],56 Tags=tags,57 )58 return resolver_endpoint["ResolverEndpoint"]59@mock_route53resolver60def test_route53resolver_invalid_create_endpoint_args():61 """Test invalid arguments to the create_resolver_endpoint API."""62 client = boto3.client("route53resolver", region_name=TEST_REGION)63 random_num = get_random_hex(10)64 # Verify ValidationException error messages are accumulated properly:65 # - creator requestor ID that exceeds the allowed length of 255.66 # - name that exceeds the allowed length of 64.67 # - direction that's neither INBOUND or OUTBOUND.68 # - more than 10 IP Address sets.69 # - too many security group IDs.70 long_id = random_num * 25 + "123456"71 long_name = random_num * 6 + "abcde"72 too_many_security_groups = ["sg-" + get_random_hex(63)]73 bad_direction = "foo"74 too_many_ip_addresses = [{"SubnetId": f"{x}", "Ip": f"{x}" * 7} for x in range(11)]75 with pytest.raises(ClientError) as exc:76 client.create_resolver_endpoint(77 CreatorRequestId=long_id,78 Name=long_name,79 SecurityGroupIds=too_many_security_groups,80 Direction=bad_direction,81 IpAddresses=too_many_ip_addresses,82 )83 err = exc.value.response["Error"]84 assert err["Code"] == "ValidationException"85 assert "5 validation errors detected" in err["Message"]86 assert (87 f"Value '{long_id}' at 'creatorRequestId' failed to satisfy constraint: "88 f"Member must have length less than or equal to 255"89 ) in err["Message"]90 assert (91 f"Value '{too_many_security_groups}' at 'securityGroupIds' failed to "92 f"satisfy constraint: Member must have length less than or equal to 64"93 ) in err["Message"]94 assert (95 f"Value '{long_name}' at 'name' failed to satisfy constraint: "96 f"Member must have length less than or equal to 64"97 ) in err["Message"]98 assert (99 f"Value '{bad_direction}' at 'direction' failed to satisfy constraint: "100 f"Member must satisfy enum value set: [INBOUND, OUTBOUND]"101 ) in err["Message"]102 assert (103 f"Value '{too_many_ip_addresses}' at 'ipAddresses' failed to satisfy "104 f"constraint: Member must have length less than or equal to 10"105 ) in err["Message"]106 # Some single ValidationException errors ...107 bad_chars_in_name = "0@*3"108 ok_group_ids = ["sg-" + random_num]109 ok_ip_addrs = [{"SubnetId": f"{x}", "Ip": f"{x}" * 7} for x in range(10)]110 with pytest.raises(ClientError) as exc:111 client.create_resolver_endpoint(112 CreatorRequestId=random_num,113 Name=bad_chars_in_name,114 SecurityGroupIds=ok_group_ids,115 Direction="INBOUND",116 IpAddresses=ok_ip_addrs,117 )118 err = exc.value.response["Error"]119 assert err["Code"] == "ValidationException"120 assert "1 validation error detected" in err["Message"]121 assert (122 rf"Value '{bad_chars_in_name}' at 'name' failed to satisfy constraint: "123 rf"Member must satisfy regular expression pattern: "124 rf"^(?!^[0-9]+$)([a-zA-Z0-9-_' ']+)$"125 ) in err["Message"]126 subnet_too_long = [{"SubnetId": "a" * 33, "Ip": "1.2.3.4"}]127 with pytest.raises(ClientError) as exc:128 client.create_resolver_endpoint(129 CreatorRequestId=random_num,130 Name="X" + random_num,131 SecurityGroupIds=ok_group_ids,132 Direction="OUTBOUND",133 IpAddresses=subnet_too_long,134 )135 err = exc.value.response["Error"]136 assert err["Code"] == "ValidationException"137 assert "1 validation error detected" in err["Message"]138 assert (139 f"Value '{subnet_too_long}' at 'ipAddresses.subnetId' failed to "140 f"satisfy constraint: Member must have length less than or equal to 32"141 ) in err["Message"]142@mock_ec2143@mock_route53resolver144def test_route53resolver_bad_create_endpoint_subnets():145 """Test bad subnet scenarios for create_resolver_endpoint API."""146 client = boto3.client("route53resolver", region_name=TEST_REGION)147 ec2_client = boto3.client("ec2", region_name=TEST_REGION)148 random_num = get_random_hex(10)149 # Need 2 IP addresses at the minimum.150 subnet_ids = create_subnets(ec2_client, create_vpc(ec2_client))151 with pytest.raises(ClientError) as exc:152 client.create_resolver_endpoint(153 CreatorRequestId=random_num,154 Name="X" + random_num,155 SecurityGroupIds=[f"sg-{random_num}"],156 Direction="INBOUND",157 IpAddresses=[{"SubnetId": subnet_ids[0], "Ip": "1.2.3.4"}],158 )159 err = exc.value.response["Error"]160 assert err["Code"] == "InvalidRequestException"161 assert "Resolver endpoint needs to have at least 2 IP addresses" in err["Message"]162 # Need an IP that's within in the subnet.163 bad_ip_addr = "1.2.3.4"164 with pytest.raises(ClientError) as exc:165 client.create_resolver_endpoint(166 CreatorRequestId=random_num,167 Name="X" + random_num,168 SecurityGroupIds=[f"sg-{random_num}"],169 Direction="INBOUND",170 IpAddresses=[171 {"SubnetId": subnet_ids[0], "Ip": bad_ip_addr},172 {"SubnetId": subnet_ids[1], "Ip": bad_ip_addr},173 ],174 )175 err = exc.value.response["Error"]176 assert err["Code"] == "InvalidRequestException"177 assert (178 f"IP address '{bad_ip_addr}' is either not in subnet "179 f"'{subnet_ids[0]}' CIDR range or is reserved"180 ) in err["Message"]181 # Bad subnet ID.182 with pytest.raises(ClientError) as exc:183 client.create_resolver_endpoint(184 CreatorRequestId=random_num,185 Name="X" + random_num,186 SecurityGroupIds=[f"sg-{random_num}"],187 Direction="INBOUND",188 IpAddresses=[189 {"SubnetId": "foo", "Ip": "1.2.3.4"},190 {"SubnetId": subnet_ids[1], "Ip": "1.2.3.4"},191 ],192 )193 err = exc.value.response["Error"]194 assert err["Code"] == "InvalidParameterException"195 assert "The subnet ID 'foo' does not exist" in err["Message"]196 # Can't reuse a ip address in a subnet.197 subnet_ids = create_subnets(ec2_client, create_vpc(ec2_client))198 with pytest.raises(ClientError) as exc:199 client.create_resolver_endpoint(200 CreatorRequestId="B" + random_num,201 Name="B" + random_num,202 SecurityGroupIds=[create_security_group(ec2_client)],203 Direction="INBOUND",204 IpAddresses=[205 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.200"},206 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.200"},207 ],208 )209 err = exc.value.response["Error"]210 assert err["Code"] == "ResourceExistsException"211 assert (212 f"The IP address '10.0.1.200' in subnet '{subnet_ids[0]}' is already in use"213 ) in err["Message"]214@mock_ec2215@mock_route53resolver216def test_route53resolver_bad_create_endpoint_security_groups():217 """Test bad security group scenarios for create_resolver_endpoint API."""218 client = boto3.client("route53resolver", region_name=TEST_REGION)219 ec2_client = boto3.client("ec2", region_name=TEST_REGION)220 random_num = get_random_hex(10)221 subnet_ids = create_subnets(ec2_client, create_vpc(ec2_client))222 ip_addrs = [223 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.200"},224 {"SubnetId": subnet_ids[1], "Ip": "10.0.0.20"},225 ]226 # Subnet must begin with "sg-".227 with pytest.raises(ClientError) as exc:228 client.create_resolver_endpoint(229 CreatorRequestId=random_num,230 Name="X" + random_num,231 SecurityGroupIds=["foo"],232 Direction="INBOUND",233 IpAddresses=ip_addrs,234 )235 err = exc.value.response["Error"]236 assert err["Code"] == "InvalidParameterException"237 assert (238 "Malformed security group ID: Invalid id: 'foo' (expecting 'sg-...')"239 ) in err["Message"]240 # Non-existent security group id.241 with pytest.raises(ClientError) as exc:242 client.create_resolver_endpoint(243 CreatorRequestId=random_num,244 Name="X" + random_num,245 SecurityGroupIds=["sg-abc"],246 Direction="INBOUND",247 IpAddresses=ip_addrs,248 )249 err = exc.value.response["Error"]250 assert err["Code"] == "ResourceNotFoundException"251 assert "The security group 'sg-abc' does not exist" in err["Message"]252 # Too many security group ids.253 with pytest.raises(ClientError) as exc:254 client.create_resolver_endpoint(255 CreatorRequestId=random_num,256 Name="X" + random_num,257 SecurityGroupIds=["sg-abc"] * 11,258 Direction="INBOUND",259 IpAddresses=ip_addrs,260 )261 err = exc.value.response["Error"]262 assert err["Code"] == "InvalidParameterException"263 assert "Maximum of 10 security groups are allowed" in err["Message"]264@mock_ec2265@mock_route53resolver266def test_route53resolver_create_resolver_endpoint(): # pylint: disable=too-many-locals267 """Test good create_resolver_endpoint API calls."""268 client = boto3.client("route53resolver", region_name=TEST_REGION)269 ec2_client = boto3.client("ec2", region_name=TEST_REGION)270 random_num = get_random_hex(10)271 vpc_id = create_vpc(ec2_client)272 subnet_ids = create_subnets(ec2_client, vpc_id)273 ip_addrs = [274 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.200"},275 {"SubnetId": subnet_ids[1], "Ip": "10.0.0.20"},276 ]277 security_group_id = create_security_group(ec2_client)278 creator_request_id = random_num279 name = "X" + random_num280 response = client.create_resolver_endpoint(281 CreatorRequestId=creator_request_id,282 Name=name,283 SecurityGroupIds=[security_group_id],284 Direction="INBOUND",285 IpAddresses=ip_addrs,286 )287 endpoint = response["ResolverEndpoint"]288 id_value = endpoint["Id"]289 assert id_value.startswith("rslvr-in-")290 assert endpoint["CreatorRequestId"] == creator_request_id291 assert (292 endpoint["Arn"]293 == f"arn:aws:route53resolver:{TEST_REGION}:{ACCOUNT_ID}:resolver-endpoint/​{id_value}"294 )295 assert endpoint["Name"] == name296 assert endpoint["SecurityGroupIds"] == [security_group_id]297 assert endpoint["Direction"] == "INBOUND"298 assert endpoint["IpAddressCount"] == 2299 assert endpoint["HostVPCId"] == vpc_id300 assert endpoint["Status"] == "OPERATIONAL"301 assert "Successfully created Resolver Endpoint" in endpoint["StatusMessage"]302 time_format = "%Y-%m-%dT%H:%M:%S.%f+00:00"303 now = datetime.now(timezone.utc).replace(tzinfo=None)304 creation_time = datetime.strptime(endpoint["CreationTime"], time_format)305 creation_time = creation_time.replace(tzinfo=None)306 assert creation_time <= now307 modification_time = datetime.strptime(endpoint["ModificationTime"], time_format)308 modification_time = modification_time.replace(tzinfo=None)309 assert modification_time <= now310@mock_ec2311@mock_route53resolver312def test_route53resolver_other_create_resolver_endpoint_errors():313 """Test other error scenarios for create_resolver_endpoint API calls."""314 client = boto3.client("route53resolver", region_name=TEST_REGION)315 ec2_client = boto3.client("ec2", region_name=TEST_REGION)316 # Create a good endpoint that we can use to test.317 created_endpoint = create_test_endpoint(client, ec2_client)318 request_id = created_endpoint["CreatorRequestId"]319 # Attempt to create another endpoint with the same creator request id.320 vpc_id = create_vpc(ec2_client)321 subnet_ids = create_subnets(ec2_client, vpc_id)322 with pytest.raises(ClientError) as exc:323 client.create_resolver_endpoint(324 CreatorRequestId=created_endpoint["CreatorRequestId"],325 Name="X" + get_random_hex(10),326 SecurityGroupIds=created_endpoint["SecurityGroupIds"],327 Direction="INBOUND",328 IpAddresses=[329 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.200"},330 {"SubnetId": subnet_ids[1], "Ip": "10.0.0.20"},331 ],332 )333 err = exc.value.response["Error"]334 assert err["Code"] == "ResourceExistsException"335 assert (336 f"Resolver endpoint with creator request ID '{request_id}' already exists"337 ) in err["Message"]338 # Too many endpoints.339 random_num = get_random_hex(10)340 for idx in range(4):341 create_test_endpoint(client, ec2_client, name=f"A{idx}-{random_num}")342 with pytest.raises(ClientError) as exc:343 create_test_endpoint(client, ec2_client, name=f"A5-{random_num}")344 err = exc.value.response["Error"]345 assert err["Code"] == "LimitExceededException"346 assert f"Account '{ACCOUNT_ID}' has exceeded 'max-endpoints" in err["Message"]347@mock_ec2348@mock_route53resolver349def test_route53resolver_delete_resolver_endpoint():350 """Test good delete_resolver_endpoint API calls."""351 client = boto3.client("route53resolver", region_name=TEST_REGION)352 ec2_client = boto3.client("ec2", region_name=TEST_REGION)353 created_endpoint = create_test_endpoint(client, ec2_client)354 # Now delete the resolver endpoint and verify the response.355 response = client.delete_resolver_endpoint(356 ResolverEndpointId=created_endpoint["Id"]357 )358 endpoint = response["ResolverEndpoint"]359 assert endpoint["CreatorRequestId"] == created_endpoint["CreatorRequestId"]360 assert endpoint["Id"] == created_endpoint["Id"]361 assert endpoint["Arn"] == created_endpoint["Arn"]362 assert endpoint["Name"] == created_endpoint["Name"]363 assert endpoint["SecurityGroupIds"] == created_endpoint["SecurityGroupIds"]364 assert endpoint["Direction"] == created_endpoint["Direction"]365 assert endpoint["IpAddressCount"] == created_endpoint["IpAddressCount"]366 assert endpoint["HostVPCId"] == created_endpoint["HostVPCId"]367 assert endpoint["Status"] == "DELETING"368 assert "Deleting" in endpoint["StatusMessage"]369 assert endpoint["CreationTime"] == created_endpoint["CreationTime"]370 # Verify there are no endpoints or no network interfaces.371 response = client.list_resolver_endpoints()372 assert len(response["ResolverEndpoints"]) == 0373 result = ec2_client.describe_network_interfaces()374 assert len(result["NetworkInterfaces"]) == 0375@mock_ec2376@mock_route53resolver377def test_route53resolver_bad_delete_resolver_endpoint():378 """Test delete_resolver_endpoint API calls with a bad ID."""379 client = boto3.client("route53resolver", region_name=TEST_REGION)380 ec2_client = boto3.client("ec2", region_name=TEST_REGION)381 random_num = get_random_hex(10)382 # Use a resolver endpoint id that is too long.383 long_id = "0123456789" * 6 + "xxxxx"384 with pytest.raises(ClientError) as exc:385 client.delete_resolver_endpoint(ResolverEndpointId=long_id)386 err = exc.value.response["Error"]387 assert err["Code"] == "ValidationException"388 assert "1 validation error detected" in err["Message"]389 assert (390 f"Value '{long_id}' at 'resolverEndpointId' failed to satisfy "391 f"constraint: Member must have length less than or equal to 64"392 ) in err["Message"]393 # Delete a non-existent resolver endpoint.394 with pytest.raises(ClientError) as exc:395 client.delete_resolver_endpoint(ResolverEndpointId=random_num)396 err = exc.value.response["Error"]397 assert err["Code"] == "ResourceNotFoundException"398 assert f"Resolver endpoint with ID '{random_num}' does not exist" in err["Message"]399 # Create an endpoint and a rule referencing that endpoint. Verify the400 # endpoint can't be deleted due to that rule.401 endpoint = create_test_endpoint(client, ec2_client)402 resolver_rule = client.create_resolver_rule(403 CreatorRequestId=random_num,404 RuleType="FORWARD",405 DomainName=f"X{random_num}.com",406 ResolverEndpointId=endpoint["Id"],407 )408 with pytest.raises(ClientError) as exc:409 client.delete_resolver_endpoint(ResolverEndpointId=endpoint["Id"])410 err = exc.value.response["Error"]411 assert err["Code"] == "InvalidRequestException"412 assert (413 f"Cannot delete resolver endpoint unless its related resolver rules "414 f"are deleted. The following rules still exist for this resolver "415 f"endpoint: {resolver_rule['ResolverRule']['Id']}"416 ) in err["Message"]417@mock_ec2418@mock_route53resolver419def test_route53resolver_get_resolver_endpoint():420 """Test good get_resolver_endpoint API calls."""421 client = boto3.client("route53resolver", region_name=TEST_REGION)422 ec2_client = boto3.client("ec2", region_name=TEST_REGION)423 created_endpoint = create_test_endpoint(client, ec2_client)424 # Now get the resolver endpoint and verify the response.425 response = client.get_resolver_endpoint(ResolverEndpointId=created_endpoint["Id"])426 endpoint = response["ResolverEndpoint"]427 assert endpoint["CreatorRequestId"] == created_endpoint["CreatorRequestId"]428 assert endpoint["Id"] == created_endpoint["Id"]429 assert endpoint["Arn"] == created_endpoint["Arn"]430 assert endpoint["Name"] == created_endpoint["Name"]431 assert endpoint["SecurityGroupIds"] == created_endpoint["SecurityGroupIds"]432 assert endpoint["Direction"] == created_endpoint["Direction"]433 assert endpoint["IpAddressCount"] == created_endpoint["IpAddressCount"]434 assert endpoint["HostVPCId"] == created_endpoint["HostVPCId"]435 assert endpoint["Status"] == created_endpoint["Status"]436 assert endpoint["StatusMessage"] == created_endpoint["StatusMessage"]437 assert endpoint["CreationTime"] == created_endpoint["CreationTime"]438 assert endpoint["ModificationTime"] == created_endpoint["ModificationTime"]439@mock_route53resolver440def test_route53resolver_bad_get_resolver_endpoint():441 """Test get_resolver_endpoint API calls with a bad ID."""442 client = boto3.client("route53resolver", region_name=TEST_REGION)443 random_num = get_random_hex(10)444 # Use a resolver endpoint id that is too long.445 long_id = "0123456789" * 6 + "xxxxx"446 with pytest.raises(ClientError) as exc:447 client.get_resolver_endpoint(ResolverEndpointId=long_id)448 err = exc.value.response["Error"]449 assert err["Code"] == "ValidationException"450 assert "1 validation error detected" in err["Message"]451 assert (452 f"Value '{long_id}' at 'resolverEndpointId' failed to satisfy "453 f"constraint: Member must have length less than or equal to 64"454 ) in err["Message"]455 # Delete a non-existent resolver endpoint.456 with pytest.raises(ClientError) as exc:457 client.get_resolver_endpoint(ResolverEndpointId=random_num)458 err = exc.value.response["Error"]459 assert err["Code"] == "ResourceNotFoundException"460 assert f"Resolver endpoint with ID '{random_num}' does not exist" in err["Message"]461@mock_ec2462@mock_route53resolver463def test_route53resolver_update_resolver_endpoint():464 """Test good update_resolver_endpoint API calls."""465 client = boto3.client("route53resolver", region_name=TEST_REGION)466 ec2_client = boto3.client("ec2", region_name=TEST_REGION)467 created_endpoint = create_test_endpoint(client, ec2_client)468 # Now update the resolver endpoint name and verify the response.469 new_name = "NewName" + get_random_hex(6)470 response = client.update_resolver_endpoint(471 ResolverEndpointId=created_endpoint["Id"], Name=new_name472 )473 endpoint = response["ResolverEndpoint"]474 assert endpoint["CreatorRequestId"] == created_endpoint["CreatorRequestId"]475 assert endpoint["Id"] == created_endpoint["Id"]476 assert endpoint["Arn"] == created_endpoint["Arn"]477 assert endpoint["Name"] == new_name478 assert endpoint["SecurityGroupIds"] == created_endpoint["SecurityGroupIds"]479 assert endpoint["Direction"] == created_endpoint["Direction"]480 assert endpoint["IpAddressCount"] == created_endpoint["IpAddressCount"]481 assert endpoint["HostVPCId"] == created_endpoint["HostVPCId"]482 assert endpoint["Status"] == created_endpoint["Status"]483 assert endpoint["StatusMessage"] == created_endpoint["StatusMessage"]484 assert endpoint["CreationTime"] == created_endpoint["CreationTime"]485 assert endpoint["ModificationTime"] != created_endpoint["ModificationTime"]486@mock_route53resolver487def test_route53resolver_bad_update_resolver_endpoint():488 """Test update_resolver_endpoint API calls with a bad ID."""489 client = boto3.client("route53resolver", region_name=TEST_REGION)490 random_num = get_random_hex(10)491 random_name = "Z" + get_random_hex(10)492 # Use a resolver endpoint id that is too long.493 long_id = "0123456789" * 6 + "xxxxx"494 with pytest.raises(ClientError) as exc:495 client.update_resolver_endpoint(ResolverEndpointId=long_id, Name=random_name)496 err = exc.value.response["Error"]497 assert err["Code"] == "ValidationException"498 assert "1 validation error detected" in err["Message"]499 assert (500 f"Value '{long_id}' at 'resolverEndpointId' failed to satisfy "501 f"constraint: Member must have length less than or equal to 64"502 ) in err["Message"]503 # Delete a non-existent resolver endpoint.504 with pytest.raises(ClientError) as exc:505 client.update_resolver_endpoint(ResolverEndpointId=random_num, Name=random_name)506 err = exc.value.response["Error"]507 assert err["Code"] == "ResourceNotFoundException"508 assert f"Resolver endpoint with ID '{random_num}' does not exist" in err["Message"]509@mock_ec2510@mock_route53resolver511def test_route53resolver_list_resolver_endpoint_ip_addresses():512 """Test good list_resolver_endpoint_ip_addresses API calls."""513 client = boto3.client("route53resolver", region_name=TEST_REGION)514 ec2_client = boto3.client("ec2", region_name=TEST_REGION)515 random_num = get_random_hex(10)516 subnet_ids = create_subnets(ec2_client, create_vpc(ec2_client))517 response = client.create_resolver_endpoint(518 CreatorRequestId="B" + random_num,519 Name="B" + random_num,520 SecurityGroupIds=[create_security_group(ec2_client)],521 Direction="INBOUND",522 IpAddresses=[523 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.200"},524 {"SubnetId": subnet_ids[1], "Ip": "10.0.0.20"},525 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.201"},526 ],527 )528 endpoint_id = response["ResolverEndpoint"]["Id"]529 response = client.list_resolver_endpoint_ip_addresses(530 ResolverEndpointId=endpoint_id531 )532 assert len(response["IpAddresses"]) == 3533 assert response["MaxResults"] == 10534 # Set max_results to return 1 address, use next_token to get remaining.535 response = client.list_resolver_endpoint_ip_addresses(536 ResolverEndpointId=endpoint_id, MaxResults=1537 )538 ip_addresses = response["IpAddresses"]539 assert len(ip_addresses) == 1540 assert response["MaxResults"] == 1541 assert "NextToken" in response542 assert ip_addresses[0]["IpId"].startswith("rni-")543 assert ip_addresses[0]["SubnetId"] == subnet_ids[0]544 assert ip_addresses[0]["Ip"] == "10.0.1.200"545 assert ip_addresses[0]["Status"] == "ATTACHED"546 assert ip_addresses[0]["StatusMessage"] == "This IP address is operational."547 assert "CreationTime" in ip_addresses[0]548 assert "ModificationTime" in ip_addresses[0]549 response = client.list_resolver_endpoint_ip_addresses(550 ResolverEndpointId=endpoint_id, NextToken=response["NextToken"]551 )552 assert len(response["IpAddresses"]) == 2553 assert response["MaxResults"] == 10554 assert "NextToken" not in response555@mock_ec2556@mock_route53resolver557def test_route53resolver_bad_list_resolver_endpoint_ip_addresses():558 """Test bad list_resolver_endpoint_ip_addresses API calls."""559 client = boto3.client("route53resolver", region_name=TEST_REGION)560 ec2_client = boto3.client("ec2", region_name=TEST_REGION)561 # Bad endpoint id.562 with pytest.raises(ClientError) as exc:563 client.list_resolver_endpoint_ip_addresses(ResolverEndpointId="foo")564 err = exc.value.response["Error"]565 assert err["Code"] == "ResourceNotFoundException"566 assert "Resolver endpoint with ID 'foo' does not exist" in err["Message"]567 # Good endpoint id, but bad max_results.568 random_num = get_random_hex(10)569 response = create_test_endpoint(client, ec2_client, name=f"A-{random_num}")570 with pytest.raises(ClientError) as exc:571 client.list_resolver_endpoint_ip_addresses(572 ResolverEndpointId=response["Id"], MaxResults=250573 )574 err = exc.value.response["Error"]575 assert err["Code"] == "ValidationException"576 assert "1 validation error detected" in err["Message"]577 assert (578 "Value '250' at 'maxResults' failed to satisfy constraint: Member "579 "must have length less than or equal to 100"580 ) in err["Message"]581@mock_ec2582@mock_route53resolver583def test_route53resolver_list_resolver_endpoints():584 """Test good list_resolver_endpoints API calls."""585 client = boto3.client("route53resolver", region_name=TEST_REGION)586 ec2_client = boto3.client("ec2", region_name=TEST_REGION)587 random_num = get_random_hex(10)588 # List endpoints when there are none.589 response = client.list_resolver_endpoints()590 assert len(response["ResolverEndpoints"]) == 0591 assert response["MaxResults"] == 10592 assert "NextToken" not in response593 # Create 5 endpoints, verify all 5 are listed when no filters, max_results.594 for idx in range(4):595 create_test_endpoint(client, ec2_client, name=f"A{idx}-{random_num}")596 response = client.list_resolver_endpoints()597 endpoints = response["ResolverEndpoints"]598 assert len(endpoints) == 4599 assert response["MaxResults"] == 10600 for idx in range(4):601 assert endpoints[idx]["Name"].startswith(f"A{idx}")602 # Set max_results to return 1 endpoint, use next_token to get remaining 3.603 response = client.list_resolver_endpoints(MaxResults=1)604 endpoints = response["ResolverEndpoints"]605 assert len(endpoints) == 1606 assert response["MaxResults"] == 1607 assert "NextToken" in response608 assert endpoints[0]["Name"].startswith("A0")609 response = client.list_resolver_endpoints(NextToken=response["NextToken"])610 endpoints = response["ResolverEndpoints"]611 assert len(endpoints) == 3612 assert response["MaxResults"] == 10613 assert "NextToken" not in response614 for idx, endpoint in enumerate(endpoints):615 assert endpoint["Name"].startswith(f"A{idx + 1}")616@mock_ec2617@mock_route53resolver618def test_route53resolver_list_resolver_endpoints_filters():619 """Test good list_resolver_endpoints API calls that use filters."""620 client = boto3.client("route53resolver", region_name=TEST_REGION)621 ec2_client = boto3.client("ec2", region_name=TEST_REGION)622 random_num = get_random_hex(10)623 # Create some endpoints for testing purposes624 security_group_id = create_security_group(ec2_client)625 vpc_id = create_vpc(ec2_client)626 subnet_ids = create_subnets(ec2_client, vpc_id)627 ip0_values = ["10.0.1.201", "10.0.1.202", "10.0.1.203", "10.0.1.204"]628 ip1_values = ["10.0.0.21", "10.0.0.22", "10.0.0.23", "10.0.0.24"]629 endpoints = []630 for idx in range(1, 5):631 ip_addrs = [632 {"SubnetId": subnet_ids[0], "Ip": "10.0.1.200"},633 {"SubnetId": subnet_ids[1], "Ip": "10.0.0.20"},634 {"SubnetId": subnet_ids[0], "Ip": ip0_values[idx - 1]},635 {"SubnetId": subnet_ids[1], "Ip": ip1_values[idx - 1]},636 ]637 response = client.create_resolver_endpoint(638 CreatorRequestId=f"F{idx}-{random_num}",639 Name=f"F{idx}-{random_num}",640 SecurityGroupIds=[security_group_id],641 Direction="INBOUND" if idx % 2 else "OUTBOUND",642 IpAddresses=ip_addrs,643 )644 endpoints.append(response["ResolverEndpoint"])645 # Try all the valid filter names, including some of the old style names.646 response = client.list_resolver_endpoints(647 Filters=[{"Name": "CreatorRequestId", "Values": [f"F3-{random_num}"]}]648 )649 assert len(response["ResolverEndpoints"]) == 1650 assert response["ResolverEndpoints"][0]["CreatorRequestId"] == f"F3-{random_num}"651 response = client.list_resolver_endpoints(...

Full Screen

Full Screen

client.py

Source:client.py Github

copy

Full Screen

...11 def associate_resolver_rule(self, ResolverRuleId: str, VPCId: str, Name: str = None) -> Dict:12 pass13 def can_paginate(self, operation_name: str = None):14 pass15 def create_resolver_endpoint(self, CreatorRequestId: str, SecurityGroupIds: List, Direction: str, IpAddresses: List, Name: str = None, Tags: List = None) -> Dict:16 pass17 def create_resolver_rule(self, CreatorRequestId: str, RuleType: str, DomainName: str, Name: str = None, TargetIps: List = None, ResolverEndpointId: str = None, Tags: List = None) -> Dict:18 pass19 def delete_resolver_endpoint(self, ResolverEndpointId: str) -> Dict:20 pass21 def delete_resolver_rule(self, ResolverRuleId: str) -> Dict:22 pass23 def disassociate_resolver_endpoint_ip_address(self, ResolverEndpointId: str, IpAddress: Dict) -> Dict:24 pass25 def disassociate_resolver_rule(self, VPCId: str, ResolverRuleId: str) -> Dict:26 pass27 def generate_presigned_url(self, ClientMethod: str = None, Params: Dict = None, ExpiresIn: int = None, HttpMethod: str = None):28 pass29 def get_paginator(self, operation_name: str = None) -> Paginator:...

Full Screen

Full Screen

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:

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.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful