KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use list_resolver_rules method in localstack

Best Python code snippet using localstack_python

test_route53resolver_rule.py

Source:test_route53resolver_rule.py

...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}"407    response = client.list_resolver_rules(408        Filters=[409            {410                "Name": "CREATOR_REQUEST_ID",411                "Values": [f"F2-{random_num}", f"F4-{random_num}"],412            }413        ]414    )415    assert len(response["ResolverRules"]) == 2416    assert response["ResolverRules"][0]["CreatorRequestId"] == f"F2-{random_num}"417    assert response["ResolverRules"][1]["CreatorRequestId"] == f"F4-{random_num}"418    response = client.list_resolver_rules(419        Filters=[{"Name": "Type", "Values": ["FORWARD"]}]420    )421    assert len(response["ResolverRules"]) == 2422    assert response["ResolverRules"][0]["CreatorRequestId"] == f"F1-{random_num}"423    assert response["ResolverRules"][1]["CreatorRequestId"] == f"F3-{random_num}"424    response = client.list_resolver_rules(425        Filters=[{"Name": "Name", "Values": [f"F1-{random_num}"]}]426    )427    assert len(response["ResolverRules"]) == 1428    assert response["ResolverRules"][0]["Name"] == f"F1-{random_num}"429    response = client.list_resolver_rules(430        Filters=[431            {"Name": "RESOLVER_ENDPOINT_ID", "Values": [endpoint1, endpoint2]},432            {"Name": "TYPE", "Values": ["FORWARD"]},433            {"Name": "NAME", "Values": [f"F3-{random_num}"]},434        ]435    )436    assert len(response["ResolverRules"]) == 1437    assert response["ResolverRules"][0]["Name"] == f"F3-{random_num}"438    response = client.list_resolver_rules(439        Filters=[{"Name": "DomainName", "Values": ["test4.test."]}]440    )441    assert len(response["ResolverRules"]) == 1442    assert response["ResolverRules"][0]["Name"] == f"F4-{random_num}"443    response = client.list_resolver_rules(444        Filters=[{"Name": "Status", "Values": ["COMPLETE"]}]445    )446    assert len(response["ResolverRules"]) == 4447    response = client.list_resolver_rules(448        Filters=[{"Name": "Status", "Values": ["FAILED"]}]449    )450    assert len(response["ResolverRules"]) == 0451@mock_route53resolver452def test_route53resolver_bad_list_resolver_rules_filters():453    """Test bad list_resolver_rules API calls that use filters."""454    client = boto3.client("route53resolver", region_name=TEST_REGION)455    # botocore barfs on an empty "Values":456    # TypeError: list_resolver_rules() only accepts keyword arguments.457    # client.list_resolver_rules([{"Name": "Direction", "Values": []}])458    # client.list_resolver_rules([{"Values": []}])459    with pytest.raises(ClientError) as exc:460        client.list_resolver_rules(Filters=[{"Name": "foo", "Values": ["bar"]}])461    err = exc.value.response["Error"]462    assert err["Code"] == "InvalidParameterException"463    assert "The filter 'foo' is invalid" in err["Message"]464@mock_route53resolver465def test_route53resolver_bad_list_resolver_rules():466    """Test bad list_resolver_rules API calls."""467    client = boto3.client("route53resolver", region_name=TEST_REGION)468    # Bad max_results.469    random_num = get_random_hex(10)470    create_test_rule(client, name=f"A-{random_num}")471    with pytest.raises(ClientError) as exc:472        client.list_resolver_rules(MaxResults=250)473    err = exc.value.response["Error"]474    assert err["Code"] == "ValidationException"475    assert "1 validation error detected" in err["Message"]476    assert (477        "Value '250' at 'maxResults' failed to satisfy constraint: Member "478        "must have length less than or equal to 100"...

CFNAutoAssociate.py

Source:CFNAutoAssociate.py

1#!/usr/bin/python32import boto33import logging4import re5import utilities6logger = logging.getLogger()7logger.setLevel(logging.DEBUG)8def handler(event, context):9    logger.debug(f'Running AutoAssociate: {event}')10    try:11        if event['RequestType'] in ['Create', 'Update']:12            logger.debug(f'Processing: {event["RequestType"]}')13            sync_remote_associations(event, context)14        else:15            logger.debug(f'Processing: {event["RequestType"]}; NOOP')16        utilities.send_response('SUCCESS', 'OK', event, context)17    except Exception as e:18        utilities.send_response('FAILED', f'{e}', event, context)19def generate_access_token(event, context):20    role_arn = {21            'RoleArn': event['ResourceProperties']['RoleARN'],22            'RoleSessionName': 'cross_account_lambda'23            }24    peer = boto3.client('sts').assume_role(**role_arn)['Credentials']25    return {26            'aws_access_key_id': peer['AccessKeyId'],27            'aws_secret_access_key': peer['SecretAccessKey'],28            'aws_session_token': peer['SessionToken'],29            }30def sync_remote_associations(event, context):31    access_token = generate_access_token(event, context)32    local_exported_rules = set([33        re.sub('^.*\\/', '', resource['arn'])34        for resource35        in utilities.boto3_call('list_resources',36                                request={37                                    'resourceOwner': 'SELF'38                                    })39        if resource['resourceShareArn']40        == event['ResourceProperties']['ShareArn']41        and resource['type']42        == 'route53resolver:ResolverRule'43        ])44    remote_exported_vpcs = set([45        export['Value']46        for export47        in utilities.boto3_call('list_exports',48                                access_token=access_token)49        if utilities.is_exported_vpc(export)50        ])51    # `need`: associations we should have52    need = set()53    for vpc in remote_exported_vpcs - \54            set(event['ResourceProperties']['VpcDni']):55        for rule_id in local_exported_rules:56            need.add((vpc, rule_id))57    logger.debug(f'local needs: {need}')58    # `have`: associations we actually have59    have = set([60        (61            association['VPCId'],62            association['ResolverRuleId']63            )64        for association65        in utilities.boto3_call('list_resolver_rule_associations',66                                access_token=access_token)67        if association['ResolverRuleId'] in local_exported_rules68        ])69    logger.debug(f'remote has: {have}')70    remote_rules = [71            remote_rule['Id']72            for remote_rule73            in utilities.boto3_call('list_resolver_rules',74                                    access_token=access_token)75            ]76    logger.debug(f'=== {remote_rules} ===')77    # remove extra first78    for pair in have - need:79        logger.debug(f'Removing: {pair}')80        utilities.boto3_call('disassociate_resolver_rule',81                             request={82                                 'VPCId': pair[0],83                                 'ResolverRuleId': pair[1],84                                 },85                             access_token=access_token)86    # create missing; format error nicely as this happens often87    remote_rules = list()88    for pair in need - have:89        logger.debug(f'>>> Creating: {pair} <<<')90        try:91            for _ in range(3):92                if pair[1] in remote_rules:93                    logger.debug('=== Found! ===')94                    break95                logger.debug(f'=== Not found: {pair[1]} (retrying) ===')96                remote_rules = [97                        remote_rule['Id']98                        for remote_rule99                        in utilities.boto3_call('list_resolver_rules',100                                                access_token=access_token)101                        ]102            # ready or not - attempt to associate; raises on error103            utilities.boto3_call('associate_resolver_rule',104                                 request={105                                     'VPCId': pair[0],106                                     'ResolverRuleId': pair[1],107                                     'Name': 'Do not remove manually',108                                     },109                                 access_token=access_token)110        except Exception as e:111            account = re.sub('^arn:aws:iam::(\\d+):.*', '\\1',112                             event['ResourceProperties']['RoleARN'])113            domain = [114                    rr['DomainName']115                    for rr116                    in utilities.boto3_call('list_resolver_rules')117                    if rr['Id'] == pair[1]118                    ]119            raise RuntimeError(f'{account}: failed to associate RR {pair[1]}'120                               f' {domain}'...

r53resolver.py

Source:r53resolver.py

...9    ec2_resource = session.resource("ec2")10    for _vpc in ec2_resource.vpcs.all():11        vpc_list.append(_vpc.id)12    route53_resolver_resource = session.client("route53resolver")13    resolver_resources = route53_resolver_resource.list_resolver_rules()14    for _resolver in resolver_resources.get('ResolverRules'):15        resolver_list.append(_resolver.get('Id'))16    for vpc_id in filter(None, vpc_list):17        for resolver_id in filter(None, resolver_list):18            try:19                route53_resolver_resource.associate_resolver_rule(20                    Name="resolver-{}-{}".format(vpc_id, resolver_id),21                    VPCId=vpc_id,22                    ResolverRuleId=resolver_id23                )24                print("successfully attached {} to {}".format(resolver_id, vpc_id))25            except Exception as e:26                print("error occurred for resolver-id {}".format(resolver_id), e)27if __name__ == '__main__':...

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.