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

How to use route53_client method in localstack

Best Python code snippet using localstack_python

ingestor_test.py

Source:ingestor_test.py

1import pytest2from asyncio import Future, gather3from unittest.mock import MagicMock, call, patch4from test_utils.test_utils import future_of5from asyncio import sleep as real_sleep6with patch("asyncio.sleep") as sleeper:7    from dns_ingestor.ingestor import DnsZoneIngestor8    # Make the sleep take no time at all9    sleeper.return_value = future_of(None)10    @pytest.mark.unit11    @pytest.mark.asyncio12    async def test_no_zones_present():13        route53_client = MagicMock()14        record_consumer = MagicMock()15        ingestor = DnsZoneIngestor(route53_client)16        route53_client.list_hosted_zones.return_value = future_of({17            "HostedZones": [],18            "IsTruncated": False19        })20        await ingestor.ingest_zones()21        assert ingestor.num_records == 022        assert ingestor.num_zones == 023        route53_client.list_hosted_zones.assert_called_once()24        assert record_consumer.call_count == 025    # This test has a list of zones with 1 page containing 1 zone26    # This zone has a list of records with 1 page containing 1 record27    @pytest.mark.unit28    @pytest.mark.asyncio29    async def test_simple():30        route53_client = MagicMock()31        record_consumer_mock = MagicMock()32        async def record_consumer(record):33            return record_consumer_mock(record)34        ingestor = DnsZoneIngestor(route53_client)35        route53_client.list_hosted_zones.return_value = future_of({36            "HostedZones": [{37                "Id": "/hostedzone/Z31RX3GZS94JZS",38                "Name": "dsd.io.",39                "ResourceRecordSetCount": 140            }],41            "IsTruncated": False42        })43        route53_client.list_resource_record_sets.return_value = future_of({44            "ResourceRecordSets": [{45                "Name": "dsd.io.",46                "Type": "A",47                "AliasTarget": {48                    "HostedZoneId": "Z1BKCTXD74EZPE",49                    "DNSName": "s3-website-eu-west-1.amazonaws.com.",50                    "EvaluateTargetHealth": False51                }52            }],53            "IsTruncated": False54        })55        await ingestor.ingest_zones()56        await ingestor.ingest_records(record_consumer)57        assert ingestor.num_records == 158        assert ingestor.num_zones == 159        route53_client.list_hosted_zones.assert_called_once()60        record_consumer_mock.assert_called_once_with({61            "Name": "dsd.io.",62            "Type": "A",63            "AliasTarget": {64                "HostedZoneId": "Z1BKCTXD74EZPE",65                "DNSName": "s3-website-eu-west-1.amazonaws.com.",66                "EvaluateTargetHealth": False67            }68        })69    @pytest.mark.unit70    @pytest.mark.asyncio71    async def test_paging_zones():72        route53_client = MagicMock()73        record_consumer_mock = MagicMock()74        async def record_consumer(record):75            return record_consumer_mock(record)76        ingestor = DnsZoneIngestor(route53_client)77        route53_client.list_hosted_zones.side_effect = [78            future_of({79                "HostedZones": [{80                    "Id": "/hostedzone/Z31RX3GZS94JZS",81                    "Name": "dsd.io.",82                    "ResourceRecordSetCount": 183                }],84                "IsTruncated": True,85                "NextMarker": "Sandy"86            }),87            future_of({88                "HostedZones": [{89                    "Id": "/hostedzone/584EWFWE",90                    "Name": "waaaaa.io.",91                    "ResourceRecordSetCount": 192                }],93                "IsTruncated": True,94                "NextMarker": "Randy"95            }),96            future_of({97                "HostedZones": [{98                    "Id": "/hostedzone/kjonwofweno",99                    "Name": "Nananana.io.",100                    "ResourceRecordSetCount": 1101                }],102                "IsTruncated": False103            })104        ]105        route53_client.list_resource_record_sets.return_value = future_of({106            "ResourceRecordSets": [],107            "IsTruncated": False108        })109        await ingestor.ingest_zones()110        await ingestor.ingest_records(record_consumer)111        assert ingestor.num_records == 0112        assert ingestor.num_zones == 3113        assert route53_client.list_hosted_zones.call_args_list == [114            call(MaxItems="100"),115            call(MaxItems="100", Marker="Sandy"),116            call(MaxItems="100", Marker="Randy")117        ]118        assert record_consumer_mock.call_count == 0119    # N.B. that the next record identifier may or may not be present, so the test tests that is120    # is removed if not present121    @pytest.mark.unit122    @pytest.mark.asyncio123    async def test_paging_records():124        route53_client = MagicMock()125        record_consumer_mock = MagicMock()126        async def record_consumer(record):127            return record_consumer_mock(record)128        ingestor = DnsZoneIngestor(route53_client)129        route53_client.list_hosted_zones.return_value = future_of({130                "HostedZones": [{131                    "Id": "/hostedzone/Z31RX3GZS94JZS",132                    "Name": "dsd.io.",133                    "ResourceRecordSetCount": 1134                }],135                "IsTruncated": False136            })137        # Three pages, the first has NextRecordIdentifier, second doesn't and third is terminal138        route53_client.list_resource_record_sets.side_effect = [139            future_of({140                "ResourceRecordSets": [141                    {142                        "Name": "a.dsd.io.",143                        "Type": "A",144                        "AliasTarget": {145                            "HostedZoneId": "Z1BKCTXD74EZPE",146                            "DNSName": "a.s3-website-eu-west-1.amazonaws.com.",147                            "EvaluateTargetHealth": False148                        }149                    },150                    {151                        "Name": "b.dsd.io.",152                        "Type": "A",153                        "AliasTarget": {154                            "HostedZoneId": "Z1BKCTXD74EZPE",155                            "DNSName": "b.s3-website-eu-west-1.amazonaws.com.",156                            "EvaluateTargetHealth": False157                        }158                    }159                ],160                "IsTruncated": True,161                "NextRecordName": "Barny",162                "NextRecordType": "Collie",163                "NextRecordIdentifier": "Confucius"164            }),165            future_of({166                "ResourceRecordSets": [167                    {168                        "Name": "c.dsd.io.",169                        "Type": "A",170                        "AliasTarget": {171                            "HostedZoneId": "Z1BKCTXD74EZPE",172                            "DNSName": "c.s3-website-eu-west-1.amazonaws.com.",173                            "EvaluateTargetHealth": False174                        }175                    }176                ],177                "IsTruncated": True,178                "NextRecordName": "Froolio",179                "NextRecordType": "Cantooblek"180            }),181            future_of({182                "ResourceRecordSets": [183                    {184                        "Name": "d.dsd.io.",185                        "Type": "A",186                        "AliasTarget": {187                            "HostedZoneId": "Z1BKCTXD74EZPE",188                            "DNSName": "d.s3-website-eu-west-1.amazonaws.com.",189                            "EvaluateTargetHealth": False190                        }191                    }192                ],193                "IsTruncated": False194            })195        ]196        await ingestor.ingest_zones()197        await ingestor.ingest_records(record_consumer)198        assert ingestor.num_records == 4199        assert ingestor.num_zones == 1200        route53_client.list_hosted_zones.assert_called_once()201        assert route53_client.list_resource_record_sets.call_args_list == [202            call(HostedZoneId="/hostedzone/Z31RX3GZS94JZS", MaxItems="100"),203            call(204                HostedZoneId="/hostedzone/Z31RX3GZS94JZS",205                MaxItems="100",206                StartRecordName="Barny",207                StartRecordType="Collie",208                StartRecordIdentifier="Confucius"209            ),210            call(211                HostedZoneId="/hostedzone/Z31RX3GZS94JZS",212                MaxItems="100",213                StartRecordName="Froolio",214                StartRecordType="Cantooblek")215        ]216        assert record_consumer_mock.call_count == 4217    # N.B. that the next record identifier may or may not be present, so the test tests that is218    # is removed if not present219    @pytest.mark.unit220    @pytest.mark.asyncio221    async def test_sleeps_between_record_pages():222        route53_client = MagicMock()223        record_consumer_mock = MagicMock()224        async def record_consumer(record):225            return record_consumer_mock(record)226        ingestor = DnsZoneIngestor(route53_client)227        route53_client.list_hosted_zones.return_value = future_of({228            "HostedZones": [{229                "Id": "/hostedzone/Z31RX3GZS94JZS",230                "Name": "dsd.io.",231                "ResourceRecordSetCount": 1232            }],233            "IsTruncated": False234        })235        # The two pages give us the opportunity to test the sleep236        route53_client.list_resource_record_sets.side_effect = [237            future_of({238                "ResourceRecordSets": [239                    {240                        "Name": "a.dsd.io.",241                        "Type": "A",242                        "AliasTarget": {243                            "HostedZoneId": "Z1BKCTXD74EZPE",244                            "DNSName": "a.s3-website-eu-west-1.amazonaws.com.",245                            "EvaluateTargetHealth": False246                        }247                    }248                ],249                "IsTruncated": True,250                "NextRecordName": "A",251                "NextRecordType": "B",252                "NextRecordIdentifier": "C"253            }),254            future_of({255                "ResourceRecordSets": [256                    {257                        "Name": "d.dsd.io.",258                        "Type": "A",259                        "AliasTarget": {260                            "HostedZoneId": "Z1BKCTXD74EZPE",261                            "DNSName": "d.s3-website-eu-west-1.amazonaws.com.",262                            "EvaluateTargetHealth": False263                        }264                    }265                ],266                "IsTruncated": False267            })268        ]269        await ingestor.ingest_zones()270        sleeper.return_value = Future()271        # dispatch but do not await272        ingest_future = gather(ingestor.ingest_records(record_consumer))273        # Bit naff, I would rather have an event loop append task, but this is easier274        await real_sleep(0.1)275        # test that only one call to list_resource_record_sets is made before the sleep is completed276        assert ingestor.num_records == 1277        assert ingestor.num_zones == 1278        route53_client.list_hosted_zones.assert_called_once()279        route53_client.list_resource_record_sets.assert_called_once()280        assert record_consumer_mock.call_count == 1281        # Now resolve the sleep's future and await the completed ingestion process282        sleeper.return_value.set_result(None)283        await ingest_future284        # test that only one call to list_resource_record_sets is made before the sleep is completed285        assert ingestor.num_records == 2286        assert ingestor.num_zones == 1287        assert route53_client.list_resource_record_sets.call_count == 2...

inventory-route53.py

Source:inventory-route53.py

1import boto32from botocore.exceptions import ClientError3import json4import os5import time6from datetime import datetime, timezone7from dateutil import tz8from antiope.aws_account import *9from common import *10import logging11logger = logging.getLogger()12logger.setLevel(getattr(logging, os.getenv('LOG_LEVEL', default='INFO')))13logging.getLogger('botocore').setLevel(logging.WARNING)14logging.getLogger('boto3').setLevel(logging.WARNING)15logging.getLogger('urllib3').setLevel(logging.WARNING)16DOMAIN_RESOURCE_PATH = "route53/domain"17ZONE_RESOURCE_PATH = "route53/hostedzone"18def lambda_handler(event, context):19    logger.debug("Received event: " + json.dumps(event, sort_keys=True))20    message = json.loads(event['Records'][0]['Sns']['Message'])21    logger.info("Received message: " + json.dumps(message, sort_keys=True))22    try:23        target_account = AWSAccount(message['account_id'])24        discover_domains(target_account)25        discover_zones(target_account)26    except AntiopeAssumeRoleError as e:27        logger.error("Unable to assume role into account {}({})".format(target_account.account_name, target_account.account_id))28        return()29    except ClientError as e:30        logger.critical("AWS Error getting info for {}: {}".format(message['account_id'], e))31        capture_error(message, context, e, "ClientError for {}: {}".format(message['account_id'], e))32        raise33    except Exception as e:34        logger.critical("{}\nMessage: {}\nContext: {}".format(e, message, vars(context)))35        capture_error(message, context, e, "General Exception for {}: {}".format(message['account_id'], e))36        raise37def discover_domains(account):38    '''39        Gathers all the Route53Domains registered domains40    '''41    domains = []42    # Not all Public IPs are attached to instances. So we use ec2 describe_network_interfaces()43    # All results are saved to S3. Public IPs and metadata go to DDB (based on the the presense of PublicIp in the Association)44    route53_client = account.get_client('route53domains', region="us-east-1")  # Route53 Domains is only available in us-east-145    try:46        response = route53_client.list_domains()47    except ClientError as e:48        if e.response['Error']['Code'] == 'ThrottlingException':49            time.sleep(1)50            response = route53_client.list_domains()51        else:52            raise53    while 'NextPageMarker' in response:  # Gotta Catch 'em all!54        domains += response['Domains']55        try:56            response = route53_client.list_domains(Marker=response['NextPageMarker'])57        except ClientError as e:58            if e.response['Error']['Code'] == 'ThrottlingException':59                time.sleep(1)60                response = route53_client.list_domains(Marker=response['NextPageMarker'])61            else:62                raise63    domains += response['Domains']64    for d in domains:65        resource_item = {}66        resource_item['awsAccountId']                   = account.account_id67        resource_item['awsAccountName']                 = account.account_name68        resource_item['resourceType']                   = "AWS::Route53::Domain"69        resource_item['source']                         = "Antiope"70        # Get the real juicy details71        # Throttling/API Limit Info:72        # https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/DNSLimitations.html#limits-api-requests-route-5373        domain = route53_client.get_domain_detail(DomainName=d['DomainName'])74        del domain['ResponseMetadata']  # Remove response metadata. Not needed75        resource_item['configurationItemCaptureTime']   = str(datetime.datetime.now())76        resource_item['configuration']                  = domain77        resource_item['supplementaryConfiguration']     = {}78        resource_item['resourceId']                     = domain['DomainName']79        resource_item['resourceName']                   = domain['DomainName']80        resource_item['resourceCreationTime']           = domain['CreationDate']81        resource_item['errors']                         = {}82        # And the one bit of info that only list domains had.83        resource_item['supplementaryConfiguration']['TransferLock']     = d['TransferLock']84        # Not sure why Route53 product team makes me do a special call for tags.85        response = route53_client.list_tags_for_domain(DomainName=d['DomainName'])86        if 'TagList' in response:87            resource_item['Tags'] = response['TagList']88        # Need to make sure the resource name is unique and service identifiable.89        save_resource_to_s3(DOMAIN_RESOURCE_PATH, resource_item['resourceId'], resource_item)90        time.sleep(1) # To avoid throttling issues91def discover_zones(account):92    '''93        Queries AWS to determine what Route53 Zones are hosted in an AWS Account94    '''95    zones = []96    # Not all Public IPs are attached to instances. So we use ec2 describe_network_interfaces()97    # All results are saved to S3. Public IPs and metadata go to DDB (based on the the presense of PublicIp in the Association)98    route53_client = account.get_client('route53')99    # Route53 Describe calls have a very low API limit100    try:101        response = route53_client.list_hosted_zones()102    except ClientError as e:103        if e.response['Error']['Code'] == 'ThrottlingException':104            time.sleep(1)105            response = route53_client.list_hosted_zones()106        else:107            raise108    while 'IsTruncated' in response and response['IsTruncated'] is True:  # Gotta Catch 'em all!109        # Process this batch110        for zone in response['HostedZones']:111            process_zone(zone, account, route53_client)112        # Try and get some more113        try:114            response = route53_client.list_hosted_zones(Marker=response['NextMarker'])115        except ClientError as e:116            if e.response['Error']['Code'] == 'ThrottlingException':117                time.sleep(1)118                response = route53_client.list_hosted_zones(Marker=response['NextMarker'])119            else:120                raise121    # Finish Up122    for zone in response['HostedZones']:123        process_zone(zone, account, route53_client)124def process_zone(zone, account, route53_client):125    resource_item = {}126    resource_item['awsAccountId']                   = account.account_id127    resource_item['awsAccountName']                 = account.account_name128    resource_item['resourceType']                   = "AWS::Route53::HostedZone"129    resource_item['source']                         = "Antiope"130    resource_item['configurationItemCaptureTime']   = str(datetime.datetime.now())131    resource_item['configuration']                  = zone132    # resource_item['tags']                           = FIXME133    resource_item['supplementaryConfiguration']     = {}134    # Need to make sure the resource name is unique and service identifiable.135    # Zone Ids look like "/hostedzone/Z2UFNORDFDSFTZ"136    resource_item['resourceId']                     = zone['Id'].split("/")[2]137    resource_item['resourceName']                   = zone['Name']138    resource_item['errors']                         = {}139    # # Not sure why Route53 product team makes me do a special call for tags.140    # response = route53_client.list_tags_for_resource(141    #     ResourceType='hostedzone',142    #     ResourceId=zone['Id']143    # )144    # if 'ResourceTagSet' in response and 'Tags' in response['ResourceTagSet']:145    #     zone['Tags'] = response['ResourceTagSet']['Tags']146    # This also looks interesting from a data-leakage perspective147    response = route53_client.list_vpc_association_authorizations(HostedZoneId=zone['Id'])148    if 'VPCs' in response:149        resource_item['supplementaryConfiguration']['AuthorizedVPCs'] = response['VPCs']150    # This currently overloads the Route53 API Limits and exceeds Lambda Timeouts.151    # resource_item['supplementaryConfiguration']['ResourceRecordSets'] = get_resource_records(route53_client, zone['Id'])152    # resource_item['supplementaryConfiguration']['ResourceRecordSetCount'] = len(resource_item['supplementaryConfiguration']['ResourceRecordSets'])153    save_resource_to_s3(ZONE_RESOURCE_PATH, resource_item['resourceId'], resource_item)154def get_resource_records(route53_client, hostedzone_id):155    # Route 53 Resource Limits: https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/DNSLimitations.html#limits-api-requests-route-53156    # Maxitems can be 1000, frequency is hardlimited to 5 reqs per sec. Antiope will sleep 1 between calls157    rr_set = []158    response = route53_client.list_resource_record_sets(159        HostedZoneId=hostedzone_id,160        MaxItems="1000"161    )162    while response['IsTruncated']:163        rr_set += response['ResourceRecordSets']164        time.sleep(1)165        response = route53_client.list_resource_record_sets(166            HostedZoneId=hostedzone_id,167            MaxItems="1000",168            StartRecordName=response['NextRecordName']169        )170    rr_set += response['ResourceRecordSets']...

domain.py

Source:domain.py

1#!/usr/bin/python2# -*- coding: utf-8 -*-3"""Classes for Route53 domains."""4import uuid5class DomainManager:6    """Manage a Route 53 domain."""7    def __init__(self, session):8        """Create a DomainManager object."""9        self.session = session10        self.route53_client = self.session.client('route53')11    def find_hosted_zone(self, domain_name):12        """Find hosted zone by domain name."""13        paginator = self.route53_client.get_paginator('list_hosted_zones')14        for page in paginator.paginate():15            for zone in page['HostedZones']:16                if domain_name.endswith(zone['Name'][:-1]):17                    return zone18        return None19    def create_hosted_zone(self, domain_name):20        """Create Route 53 hosted zone."""21        zone_name = '.'.join(domain_name.split('.')[-2:]) + '.'22        return self.route53_client.create_hosted_zone(23            Name=zone_name,24            CallerReference=str(uuid.uuid4()),25            HostedZoneConfig={26                'Comment': 'Temporary hosted zone',27                'PrivateZone': False28            }29        )30    def create_s3_record(self, zone, domain_name, endpoint):31        """Create Alias record for the bucket."""32        return self.route53_client.change_resource_record_sets(33            HostedZoneId=zone['Id'],34            ChangeBatch={35                'Comment': 'Created by webotron',36                'Changes': [{37                    'Action': 'UPSERT',38                    'ResourceRecordSet': {39                        'Name': domain_name,40                        'Type': 'A',41                        'AliasTarget': {42                            'HostedZoneId': endpoint.zone,43                            'DNSName': endpoint.host,44                            'EvaluateTargetHealth': False45                        }46                    }47                }48                ]49            }50        )51    def create_cf_domain_record(self, zone, domain_name, cf_domain_name):52        """Create cloudfront record."""53        return self.route53_client.change_resource_record_sets(54            HostedZoneId=zone['Id'],55            ChangeBatch={56                'Comment': 'Created by webotron',57                'Changes': [{58                    'Action': 'UPSERT',59                    'ResourceRecordSet': {60                        'Name': domain_name,61                        'Type': 'A',62                        'AliasTarget': {63                            # according to aws docs64                            # https://docs.aws.amazon.com/Route53/latest/APIReference/API_AliasTarget.html65                            'HostedZoneId': 'Z2FDTNDATAQYW2',66                            'DNSName': cf_domain_name,67                            'EvaluateTargetHealth': False68                        }69                    }70                }71                ]72            }73        )74    def delete_cf_domain_record(self, zone, domain_name, cf_domain_name):75        """Delete CloudFront route53 record."""76        return self.route53_client.change_resource_record_sets(77            HostedZoneId=zone['Id'],78            ChangeBatch={79                'Changes': [{80                    'Action': 'DELETE',81                    'ResourceRecordSet': {82                        'Name': domain_name,83                        'Type': 'A',84                        'AliasTarget': {85                            # according to aws docs86                            # https://docs.aws.amazon.com/Route53/latest/APIReference/API_AliasTarget.html87                            'HostedZoneId': 'Z2FDTNDATAQYW2',88                            'DNSName': cf_domain_name,89                            'EvaluateTargetHealth': False90                        }91                    }92                }93                ]94            }...

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.