How to use create_resolver_rule method in localstack
Best Python code snippet using localstack_python
test_route53resolver_rule.py
Source:test_route53resolver_rule.py 
...14 """15 if not tags:16 tags = []17 random_num = get_random_hex(10)18 resolver_rule = client.create_resolver_rule(19 CreatorRequestId=random_num,20 Name=name if name else "X" + random_num,21 RuleType="FORWARD",22 DomainName=f"X{random_num}.com",23 TargetIps=[24 {"Ip": "10.0.1.200", "Port": 123},25 {"Ip": "10.0.0.20", "Port": 456},26 ],27 # ResolverEndpointId=random_num -- will test this separately28 Tags=tags,29 )30 return resolver_rule["ResolverRule"]31@mock_route53resolver32def test_route53resolver_invalid_create_rule_args():33 """Test invalid arguments to the create_resolver_rule API."""34 client = boto3.client("route53resolver", region_name=TEST_REGION)35 random_num = get_random_hex(10)36 # Verify ValidationException error messages are accumulated properly:37 # - creator requestor ID that exceeds the allowed length of 255.38 # - name that exceeds the allowed length of 64.39 # - rule_type that's not FORWARD, SYSTEM or RECURSIVE.40 # - domain_name that exceeds the allowed length of 256.41 long_id = random_num * 25 + "123456"42 long_name = random_num * 6 + "abcde"43 bad_rule_type = "foo"44 long_domain_name = "bar" * 8645 with pytest.raises(ClientError) as exc:46 client.create_resolver_rule(47 CreatorRequestId=long_id,48 Name=long_name,49 RuleType=bad_rule_type,50 DomainName=long_domain_name,51 )52 err = exc.value.response["Error"]53 assert err["Code"] == "ValidationException"54 assert "4 validation errors detected" in err["Message"]55 assert (56 f"Value '{long_id}' at 'creatorRequestId' failed to satisfy constraint: "57 f"Member must have length less than or equal to 255"58 ) in err["Message"]59 assert (60 f"Value '{long_name}' at 'name' failed to satisfy constraint: "61 f"Member must have length less than or equal to 64"62 ) in err["Message"]63 assert (64 f"Value '{bad_rule_type}' at 'ruleType' failed to satisfy constraint: "65 f"Member must satisfy enum value set: [FORWARD, SYSTEM, RECURSIVE]"66 ) in err["Message"]67 assert (68 f"Value '{long_domain_name}' at 'domainName' failed to satisfy "69 f"constraint: Member must have length less than or equal to 256"70 ) in err["Message"]71 # Some single ValidationException errors ...72 bad_target_ips = [73 {"Ip": "10.1.0.22", "Port": 5},74 {"Ip": "10.1.0.23", "Port": 700000},75 {"Ip": "10.1.0.24", "Port": 70},76 ]77 with pytest.raises(ClientError) as exc:78 client.create_resolver_rule(79 CreatorRequestId=random_num,80 Name="A" + random_num,81 RuleType="FORWARD",82 DomainName=f"{random_num}.com",83 TargetIps=bad_target_ips,84 )85 err = exc.value.response["Error"]86 assert err["Code"] == "ValidationException"87 assert "1 validation error detected" in err["Message"]88 assert (89 f"Value '{bad_target_ips[1]}' at 'targetIps.port' failed to "90 f"satisfy constraint: Member must have value less than or equal to "91 f"65535"92 ) in err["Message"]93 too_long_resolver_endpoint_id = "foo" * 2594 with pytest.raises(ClientError) as exc:95 client.create_resolver_rule(96 CreatorRequestId=random_num,97 Name="A" + random_num,98 RuleType="FORWARD",99 DomainName=f"{random_num}.com",100 ResolverEndpointId=too_long_resolver_endpoint_id,101 )102 err = exc.value.response["Error"]103 assert err["Code"] == "ValidationException"104 assert "1 validation error detected" in err["Message"]105 assert (106 f"Value '{too_long_resolver_endpoint_id}' at 'resolverEndpointId' "107 f"failed to satisfy constraint: Member must have length less than or "108 f"equal to 64"109 ) in err["Message"]110@mock_ec2111@mock_route53resolver112def test_route53resolver_create_resolver_rule(): # pylint: disable=too-many-locals113 """Test good create_resolver_rule API calls."""114 client = boto3.client("route53resolver", region_name=TEST_REGION)115 ec2_client = boto3.client("ec2", region_name=TEST_REGION)116 random_num = get_random_hex(10)117 # Create a good endpoint that we can use to test.118 created_endpoint = create_test_endpoint(client, ec2_client)119 endpoint_id = created_endpoint["Id"]120 creator_request_id = random_num121 name = "X" + random_num122 domain_name = f"{random_num}.test"123 target_ips = [{"Ip": "1.2.3.4", "Port": 5}]124 response = client.create_resolver_rule(125 CreatorRequestId=creator_request_id,126 Name=name,127 RuleType="FORWARD",128 DomainName=domain_name,129 TargetIps=target_ips,130 ResolverEndpointId=endpoint_id,131 )132 rule = response["ResolverRule"]133 id_value = rule["Id"]134 assert id_value.startswith("rslvr-rr-")135 assert rule["CreatorRequestId"] == creator_request_id136 assert (137 rule["Arn"]138 == f"arn:aws:route53resolver:{TEST_REGION}:{ACCOUNT_ID}:resolver-rule/{id_value}"139 )140 assert rule["DomainName"] == domain_name + "."141 assert rule["Status"] == "COMPLETE"142 assert "Successfully created Resolver Rule" in rule["StatusMessage"]143 assert rule["RuleType"] == "FORWARD"144 assert rule["Name"] == name145 assert len(rule["TargetIps"]) == 1146 assert rule["TargetIps"][0]["Ip"] == target_ips[0]["Ip"]147 assert rule["TargetIps"][0]["Port"] == target_ips[0]["Port"]148 assert rule["ResolverEndpointId"] == endpoint_id149 assert rule["OwnerId"] == ACCOUNT_ID150 assert rule["ShareStatus"] == "SHARED_WITH_ME"151 time_format = "%Y-%m-%dT%H:%M:%S.%f+00:00"152 now = datetime.now(timezone.utc).replace(tzinfo=None)153 creation_time = datetime.strptime(rule["CreationTime"], time_format)154 creation_time = creation_time.replace(tzinfo=None)155 assert creation_time <= now156 modification_time = datetime.strptime(rule["ModificationTime"], time_format)157 modification_time = modification_time.replace(tzinfo=None)158 assert modification_time <= now159@mock_ec2160@mock_route53resolver161def test_route53resolver_bad_create_resolver_rule():162 """Test error scenarios for create_resolver_rule API calls."""163 client = boto3.client("route53resolver", region_name=TEST_REGION)164 ec2_client = boto3.client("ec2", region_name=TEST_REGION)165 random_num = get_random_hex(10)166 # Create a good endpoint and rule that we can use to test.167 created_endpoint = create_test_endpoint(client, ec2_client)168 endpoint_id = created_endpoint["Id"]169 created_rule = create_test_rule(client)170 creator_request_id = created_rule["CreatorRequestId"]171 # Attempt to create another rule with the same creator request id.172 with pytest.raises(ClientError) as exc:173 client.create_resolver_rule(174 CreatorRequestId=creator_request_id,175 Name="B" + random_num,176 RuleType="FORWARD",177 DomainName=f"{random_num}.test",178 TargetIps=[{"Ip": "1.2.3.4", "Port": 5}],179 ResolverEndpointId=endpoint_id,180 )181 err = exc.value.response["Error"]182 assert err["Code"] == "ResourceExistsException"183 assert (184 f"Resolver rule with creator request ID '{creator_request_id}' already exists"185 ) in err["Message"]186 # Attempt to create a rule with a IPv6 address.187 with pytest.raises(ClientError) as exc:188 client.create_resolver_rule(189 CreatorRequestId=get_random_hex(10),190 Name="B" + random_num,191 RuleType="FORWARD",192 DomainName=f"{random_num}.test",193 TargetIps=[{"Ip": "201:db8:1234::", "Port": 5}],194 ResolverEndpointId=endpoint_id,195 )196 err = exc.value.response["Error"]197 assert err["Code"] == "InvalidParameterException"198 assert "Only IPv4 addresses may be used: '201:db8:1234::'" in err["Message"]199 # Attempt to create a rule with an invalid IPv4 address.200 with pytest.raises(ClientError) as exc:201 client.create_resolver_rule(202 CreatorRequestId=get_random_hex(10),203 Name="B" + random_num,204 RuleType="FORWARD",205 DomainName=f"{random_num}.test",206 TargetIps=[{"Ip": "20.1.2:", "Port": 5}],207 ResolverEndpointId=endpoint_id,208 )209 err = exc.value.response["Error"]210 assert err["Code"] == "InvalidParameterException"211 assert "Invalid IP address: '20.1.2:'" in err["Message"]212 # Attempt to create a rule with a non-existent resolver endpoint id.213 with pytest.raises(ClientError) as exc:214 client.create_resolver_rule(215 CreatorRequestId=get_random_hex(10),216 Name="B" + random_num,217 RuleType="FORWARD",218 DomainName=f"{random_num}.test",219 TargetIps=[{"Ip": "1.2.3.4", "Port": 5}],220 ResolverEndpointId="fooey",221 )222 err = exc.value.response["Error"]223 assert err["Code"] == "ResourceNotFoundException"224 assert "Resolver endpoint with ID 'fooey' does not exist" in err["Message"]225 # Create a rule with a resolver endpoint id and a rule type of SYSTEM.226 with pytest.raises(ClientError) as exc:227 client.create_resolver_rule(228 CreatorRequestId=get_random_hex(10),229 Name="B" + random_num,230 RuleType="SYSTEM",231 DomainName=f"{random_num}.test",232 TargetIps=[{"Ip": "1.2.3.4", "Port": 5}],233 ResolverEndpointId=endpoint_id,234 )235 err = exc.value.response["Error"]236 assert err["Code"] == "InvalidRequestException"237 assert (238 "Cannot specify resolver endpoint ID and target IP for SYSTEM type "239 "resolver rule"240 ) in err["Message"]241 # Too many rules.242 for _ in range(1000):243 create_test_rule(client)244 with pytest.raises(ClientError) as exc:245 create_test_rule(client)246 err = exc.value.response["Error"]247 assert err["Code"] == "LimitExceededException"248 assert f"Account '{ACCOUNT_ID}' has exceeded 'max-rules" in err["Message"]249@mock_route53resolver250def test_route53resolver_delete_resolver_rule():251 """Test good delete_resolver_rule API calls."""252 client = boto3.client("route53resolver", region_name=TEST_REGION)253 created_rule = create_test_rule(client)254 # Now delete the resolver rule and verify the response.255 response = client.delete_resolver_rule(ResolverRuleId=created_rule["Id"])256 rule = response["ResolverRule"]257 assert rule["Id"] == created_rule["Id"]258 assert rule["CreatorRequestId"] == created_rule["CreatorRequestId"]259 assert rule["Arn"] == created_rule["Arn"]260 assert rule["DomainName"] == created_rule["DomainName"]261 assert rule["Status"] == "DELETING"262 assert "Deleting" in rule["StatusMessage"]263 assert rule["RuleType"] == created_rule["RuleType"]264 assert rule["Name"] == created_rule["Name"]265 assert rule["TargetIps"] == created_rule["TargetIps"]266 assert rule["OwnerId"] == created_rule["OwnerId"]267 assert rule["ShareStatus"] == created_rule["ShareStatus"]268 assert rule["CreationTime"] == created_rule["CreationTime"]269@mock_ec2270@mock_route53resolver271def test_route53resolver_bad_delete_resolver_rule():272 """Test delete_resolver_rule API calls with a bad ID."""273 client = boto3.client("route53resolver", region_name=TEST_REGION)274 ec2_client = boto3.client("ec2", region_name=TEST_REGION)275 random_num = get_random_hex(10)276 # Use a resolver rule id that is too long.277 long_id = "0123456789" * 6 + "xxxxx"278 with pytest.raises(ClientError) as exc:279 client.delete_resolver_rule(ResolverRuleId=long_id)280 err = exc.value.response["Error"]281 assert err["Code"] == "ValidationException"282 assert "1 validation error detected" in err["Message"]283 assert (284 f"Value '{long_id}' at 'resolverRuleId' failed to satisfy "285 f"constraint: Member must have length less than or equal to 64"286 ) in err["Message"]287 # Delete a non-existent resolver rule.288 with pytest.raises(ClientError) as exc:289 client.delete_resolver_rule(ResolverRuleId=random_num)290 err = exc.value.response["Error"]291 assert err["Code"] == "ResourceNotFoundException"292 assert f"Resolver rule with ID '{random_num}' does not exist" in err["Message"]293 # Verify a rule can't be deleted if VPCs are associated with it.294 test_rule = create_test_rule(client)295 vpc_id = create_vpc(ec2_client)296 client.associate_resolver_rule(ResolverRuleId=test_rule["Id"], VPCId=vpc_id)297 with pytest.raises(ClientError) as exc:298 client.delete_resolver_rule(ResolverRuleId=test_rule["Id"])299 err = exc.value.response["Error"]300 assert err["Code"] == "ResourceInUseException"301 assert (302 "Please disassociate this resolver rule from VPC first before deleting"303 ) in err["Message"]304@mock_route53resolver305def test_route53resolver_get_resolver_rule():306 """Test good get_resolver_rule API calls."""307 client = boto3.client("route53resolver", region_name=TEST_REGION)308 created_rule = create_test_rule(client)309 # Now get the resolver rule and verify the response.310 response = client.get_resolver_rule(ResolverRuleId=created_rule["Id"])311 rule = response["ResolverRule"]312 assert rule["Id"] == created_rule["Id"]313 assert rule["CreatorRequestId"] == created_rule["CreatorRequestId"]314 assert rule["Arn"] == created_rule["Arn"]315 assert rule["DomainName"] == created_rule["DomainName"]316 assert rule["Status"] == created_rule["Status"]317 assert rule["StatusMessage"] == created_rule["StatusMessage"]318 assert rule["RuleType"] == created_rule["RuleType"]319 assert rule["Name"] == created_rule["Name"]320 assert rule["TargetIps"] == created_rule["TargetIps"]321 assert rule["OwnerId"] == created_rule["OwnerId"]322 assert rule["ShareStatus"] == created_rule["ShareStatus"]323 assert rule["CreationTime"] == created_rule["CreationTime"]324 assert rule["ModificationTime"] == created_rule["ModificationTime"]325@mock_route53resolver326def test_route53resolver_bad_get_resolver_rule():327 """Test get_resolver_rule API calls with a bad ID."""328 client = boto3.client("route53resolver", region_name=TEST_REGION)329 random_num = get_random_hex(10)330 # Use a resolver rule id that is too long.331 long_id = "0123456789" * 6 + "xxxxx"332 with pytest.raises(ClientError) as exc:333 client.get_resolver_rule(ResolverRuleId=long_id)334 err = exc.value.response["Error"]335 assert err["Code"] == "ValidationException"336 assert "1 validation error detected" in err["Message"]337 assert (338 f"Value '{long_id}' at 'resolverRuleId' failed to satisfy "339 f"constraint: Member must have length less than or equal to 64"340 ) in err["Message"]341 # Delete a non-existent resolver rule.342 with pytest.raises(ClientError) as exc:343 client.get_resolver_rule(ResolverRuleId=random_num)344 err = exc.value.response["Error"]345 assert err["Code"] == "ResourceNotFoundException"346 assert f"Resolver rule with ID '{random_num}' does not exist" in err["Message"]347@mock_route53resolver348def test_route53resolver_list_resolver_rules():349 """Test good list_resolver_rules API calls."""350 client = boto3.client("route53resolver", region_name=TEST_REGION)351 random_num = get_random_hex(10)352 # List rules when there are none.353 response = client.list_resolver_rules()354 assert len(response["ResolverRules"]) == 0355 assert response["MaxResults"] == 10356 assert "NextToken" not in response357 # Create 4 rules, verify all 4 are listed when no filters, max_results.358 for idx in range(4):359 create_test_rule(client, name=f"A{idx}-{random_num}")360 response = client.list_resolver_rules()361 rules = response["ResolverRules"]362 assert len(rules) == 4363 assert response["MaxResults"] == 10364 for idx in range(4):365 assert rules[idx]["Name"].startswith(f"A{idx}")366 # Set max_results to return 1 rule, use next_token to get remaining 3.367 response = client.list_resolver_rules(MaxResults=1)368 rules = response["ResolverRules"]369 assert len(rules) == 1370 assert response["MaxResults"] == 1371 assert "NextToken" in response372 assert rules[0]["Name"].startswith("A0")373 response = client.list_resolver_rules(NextToken=response["NextToken"])374 rules = response["ResolverRules"]375 assert len(rules) == 3376 assert response["MaxResults"] == 10377 assert "NextToken" not in response378 for idx, rule in enumerate(rules):379 assert rule["Name"].startswith(f"A{idx + 1}")380@mock_ec2381@mock_route53resolver382def test_route53resolver_list_resolver_rules_filters():383 """Test good list_resolver_rules API calls that use filters."""384 client = boto3.client("route53resolver", region_name=TEST_REGION)385 ec2_client = boto3.client("ec2", region_name=TEST_REGION)386 random_num = get_random_hex(10)387 # Create some endpoints and rules for testing purposes.388 endpoint1 = create_test_endpoint(client, ec2_client)["Id"]389 endpoint2 = create_test_endpoint(client, ec2_client)["Id"]390 rules = []391 for idx in range(1, 5):392 response = client.create_resolver_rule(393 CreatorRequestId=f"F{idx}-{random_num}",394 Name=f"F{idx}-{random_num}",395 RuleType="FORWARD" if idx % 2 else "RECURSIVE",396 DomainName=f"test{idx}.test",397 TargetIps=[{"Ip": f"10.0.1.{idx}", "Port": 50 + idx}],398 ResolverEndpointId=endpoint1 if idx % 2 else endpoint2,399 )400 rules.append(response["ResolverRule"])401 # Try all the valid filter names, including some of the old style names.402 response = client.list_resolver_rules(403 Filters=[{"Name": "CreatorRequestId", "Values": [f"F3-{random_num}"]}]404 )405 assert len(response["ResolverRules"]) == 1406 assert response["ResolverRules"][0]["CreatorRequestId"] == f"F3-{random_num}"...
zpadns.py
Source:zpadns.py
...64 },65 ]66)67DomainName="zscalerbeta.net"68resolver_rule = client.create_resolver_rule(69 CreatorRequestId=VpcId+'_ZSCCAppSegmentResolver_'+DomainName,70 Name='ZSCCAppSegmentResolver',71 RuleType='SYSTEM',72 DomainName=DomainName,73 Tags=[74 {75 'Key': 'Function',76 'Value': 'ZSCloudRule'77 },78 ]79)80resolver_associate = client.associate_resolver_rule(81 ResolverRuleId=resolver_rule['ResolverRule']['Id'],82 Name='ZIA_Association',83 VPCId=VpcId84)85DomainName='zscaler.com'86resolver_rule = client.create_resolver_rule(87 CreatorRequestId=VpcId+'_ZSCCAppSegmentResolver_'+DomainName,88 Name='ZSCCAppSegmentResolver',89 RuleType='SYSTEM',90 DomainName=DomainName,91 Tags=[92 {93 'Key': 'Function',94 'Value': 'ZSCloudRule'95 },96 ]97)98resolver_associate = client.associate_resolver_rule(99 ResolverRuleId=resolver_rule['ResolverRule']['Id'],100 Name='Zscaler_Association',101 VPCId=VpcId102)103DomainName='freebsd.org'104resolver_rule = client.create_resolver_rule(105 CreatorRequestId=VpcId+'_ZSCCAppSegmentResolver_'+DomainName,106 Name='ZSCCAppSegmentResolver',107 RuleType='SYSTEM',108 DomainName=DomainName,109 Tags=[110 {111 'Key': 'Function',112 'Value': 'ZSCloudRule'113 },114 ]115)116resolver_associate = client.associate_resolver_rule(117 ResolverRuleId=resolver_rule['ResolverRule']['Id'],118 Name='BSD_Association',119 VPCId=VpcId120)121DomainName='ntp.org'122resolver_rule = client.create_resolver_rule(123 CreatorRequestId=VpcId+'_ZSCCAppSegmentResolver_'+DomainName,124 Name='ZSCCAppSegmentResolver',125 RuleType='SYSTEM',126 DomainName=DomainName,127 Tags=[128 {129 'Key': 'Function',130 'Value': 'ZSCloudRule'131 },132 ]133)134resolver_associate = client.associate_resolver_rule(135 ResolverRuleId=resolver_rule['ResolverRule']['Id'],136 Name='NTP_Association',137 VPCId=VpcId138)139DomainName='amazonaws.com'140resolver_rule = client.create_resolver_rule(141 CreatorRequestId=VpcId+'_ZSCCAppSegmentResolver_'+DomainName,142 Name='ZSCCAppSegmentResolver',143 RuleType='SYSTEM',144 DomainName=DomainName,145 Tags=[146 {147 'Key': 'Function',148 'Value': 'ZSCloudRule'149 },150 ]151)152resolver_associate = client.associate_resolver_rule(153 ResolverRuleId=resolver_rule['ResolverRule']['Id'],154 Name='Amazon_Association',155 VPCId=VpcId156)157DomainName='zpabeta.net'158resolver_rule = client.create_resolver_rule(159 CreatorRequestId=VpcId+'_ZSCCAppSegmentResolver_'+DomainName,160 Name='ZSCCAppSegmentResolver',161 RuleType='SYSTEM',162 DomainName=DomainName,163 Tags=[164 {165 'Key': 'Function',166 'Value': 'ZSCloudRule'167 },168 ]169)170resolver_associate = client.associate_resolver_rule(171 ResolverRuleId=resolver_rule['ResolverRule']['Id'],172 Name='ZPA_Association',...
zpadns-add.py
Source:zpadns-add.py
...21 }22 ]23)24ResolverEndpointId=ResolverEndpoints['ResolverEndpoints'][0]['Id']25resolver_rule = client.create_resolver_rule(26 CreatorRequestId=VpcId+'_ZSCCAppSegmentResolver_'+DomainName,27 Name='ZSCCAppSegmentResolver',28 RuleType='FORWARD',29 DomainName=DomainName,30 TargetIps=[31 {32 'Ip': TargetIp,33 'Port': 5334 },35 {36 'Ip': TargetIp,37 'Port': 5338 },39 ],...
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?
