Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use list_resource_record_sets method in localstack

Best Python code snippet using localstack_python

route53_tests.py

Source:route53_tests.py

...59            You cannot interact with resource record sets for a non-existent60            zone.61            """62            client = get_client(self)63            d = client.list_resource_record_sets(u"abcdefg12345678")64            self.assertFailure(d, Route53Error)65            def got_error(error):66                self.assertEqual(NOT_FOUND, int(error.status))67            d.addCallback(got_error)68            return d69        def test_change_resource_record_sets_nonexistent_zone(self):70            """71            You cannot interact with resource record sets for a non-existent72            zone.73            """74            rrset = RRSet(75                label=Name(u"foo.example.invalid."),76                type=u"CNAME",77                ttl=60,78                records={CNAME(canonical_name=Name(u"bar.example.invalid."))},79            )80            client = get_client(self)81            d = client.change_resource_record_sets(u"abcdefg12345678", [create_rrset(rrset)])82            self.assertFailure(d, Route53Error)83            def got_error(error):84                self.assertEqual(NOT_FOUND, int(error.status))85            d.addCallback(got_error)86            return d87        def test_create_existing_rrset(self):88            """89            It is an error to attempt to create a rrset which already exists.90            """91            zone_name = u"{}.test_create_existing_rrset.invalid.".format(uuid4())92            rrset = RRSet(93                label=Name(u"foo.{}".format(zone_name)),94                type=u"CNAME",95                ttl=60,96                records={CNAME(canonical_name=Name(u"bar.example.invalid."))},97            )98            client = get_client(self)99            d = client.create_hosted_zone(u"{}".format(time()), zone_name)100            def created_zone(zone):101                self.addCleanup(lambda: self._cleanup(client, zone.identifier))102                d = client.change_resource_record_sets(zone.identifier, [create_rrset(rrset)])103                d.addCallback(lambda ignored: zone)104                return d105            d.addCallback(created_zone)106            def created_rrset(zone):107                d = client.change_resource_record_sets(zone.identifier, [create_rrset(rrset)])108                self.assertFailure(d, Route53Error)109                return d110            d.addCallback(created_rrset)111            def got_error(error):112                self.assertEqual(BAD_REQUEST, int(error.status))113            d.addCallback(got_error)114            return d115        def test_delete_missing_rrset(self):116            """117            It is an error to attempt to delete an rrset which does not exist.118            """119            zone_name = u"{}.test_delete_missing_rrset.invalid.".format(uuid4())120            rrset = RRSet(121                label=Name(u"foo.{}".format(zone_name)),122                type=u"CNAME",123                ttl=60,124                records={CNAME(canonical_name=Name(u"bar.example.invalid."))},125            )126            client = get_client(self)127            d = client.create_hosted_zone(u"{}".format(time()), zone_name)128            def created_zone(zone):129                self.addCleanup(lambda: self._cleanup(client, zone.identifier))130                d = client.change_resource_record_sets(zone.identifier, [delete_rrset(rrset)])131                self.assertFailure(d, Route53Error)132                return d133            d.addCallback(created_zone)134            def got_error(error):135                self.assertEqual(BAD_REQUEST, int(error.status))136            d.addCallback(got_error)137            return d138        def _cleanup(self, client, zone_identifier):139            d = client.delete_hosted_zone(zone_identifier)140            d.addErrback(lambda err: None)141            return d142        @inlineCallbacks143        def test_resource_record_sets(self):144            zone_name = u"{}.example.invalid.".format(uuid4())145            cname = CNAME(canonical_name=Name(u"example.invalid."))146            client = get_client(self)147            zone = yield client.create_hosted_zone(u"{}".format(time()), zone_name)148            # At least try to clean up, to be as nice as possible.149            # This might fail and someone else might have to do the150            # cleanup - but it might not!151            self.addCleanup(lambda: self._cleanup(client, zone.identifier))152            cname_label = Name(u"foo.\N{SNOWMAN}.{}".format(zone_name))153            create = create_rrset(RRSet(154                label=cname_label,155                type=u"CNAME",156                ttl=60,157                records={cname},158            ))159            yield client.change_resource_record_sets(zone.identifier, [create])160            initial = yield client.list_resource_record_sets(zone.identifier)161            key = RRSetKey(cname_label, u"CNAME")162            self.assertIn(key, initial)163            cname_rrset = initial[key]164            self.assertEqual(165                RRSet(label=cname_label, type=u"CNAME", ttl=60, records={cname}),166                cname_rrset,167            )168            # Zones start with an SOA and some NS records.169            key = RRSetKey(Name(zone_name), u"SOA")170            self.assertIn(key, initial)171            soa = initial[key]172            self.assertEqual(173                len(soa.records), 1,174                "Expected one SOA record, got {}".format(soa.records)175            )176            key = RRSetKey(Name(zone_name), u"NS")177            self.assertIn(key, initial)178            ns = initial[key]179            self.assertNotEqual(180                set(), ns.records,181                "Expected some NS records, got none"182            )183            # Unrecognized change type184            # XXX This depends on _ChangeRRSet using attrs.185            bogus = attr.assoc(create, action=u"BOGUS")186            d = client.change_resource_record_sets(zone.identifier, [bogus])187            error = yield self.assertFailure(d, Route53Error)188            self.assertEqual(BAD_REQUEST, int(error.status))189            created_a = A(IPv4Address(u"10.0.0.1"))190            upsert_label = Name(u"upsert.{}".format(zone_name))191            upsert_create = upsert_rrset(RRSet(192                upsert_label,193                u"A",194                60,195                {created_a},196            ))197            updated_a = A(IPv4Address(u"10.0.0.2"))198            upsert_update = upsert_rrset(RRSet(199                upsert_create.rrset.label,200                upsert_create.rrset.type,201                upsert_create.rrset.ttl,202                {updated_a},203            ))204            yield client.change_resource_record_sets(zone.identifier, [upsert_create])205            rrsets = yield client.list_resource_record_sets(zone.identifier)206            self.assertEqual(rrsets[RRSetKey(upsert_label, u"A")].records, {created_a})207            yield client.change_resource_record_sets(zone.identifier, [upsert_update])208            rrsets = yield client.list_resource_record_sets(zone.identifier)209            self.assertEqual(rrsets[RRSetKey(upsert_label, u"A")].records, {updated_a})210            # Use the name and maxitems parameters to select exactly one resource record.211            rrsets = yield client.list_resource_record_sets(212                zone.identifier, maxitems=1, name=upsert_label, type=u"A",213            )214            self.assertEqual(1, len(rrsets), "Expected 1 rrset")215            self.assertEqual({updated_a}, rrsets[RRSetKey(upsert_label, u"A")].records)216            # It's invalid to specify type without name.217            d = client.list_resource_record_sets(zone.identifier, type=u"A")218            error = yield self.assertFailure(d, Route53Error)219            self.assertEqual(BAD_REQUEST, int(error.status))220            # It's invalid to delete the SOA record.221            d = client.change_resource_record_sets(222                zone.identifier, [delete_rrset(soa)],223            )224            error = yield self.assertFailure(d, Route53Error)225            self.assertEqual(BAD_REQUEST, int(error.status))226            # Likewise, the NS records.227            d = client.change_resource_record_sets(228                zone.identifier, [delete_rrset(ns)],229            )230            error = yield self.assertFailure(d, Route53Error)231            self.assertEqual(BAD_REQUEST, int(error.status))232            # Test deletion at the end so the zone is clean for the233            # naive cleanup logic.234            yield client.change_resource_record_sets(235                zone.identifier, [236                    delete_rrset(cname_rrset),237                    delete_rrset(upsert_update.rrset),238                ],239            )240            rrsets = yield client.list_resource_record_sets(zone.identifier)241            self.assertNotIn(cname_label, rrsets)242            self.assertNotIn(upsert_label, rrsets)243        def test_list_resource_record_sets_maxitems(self):244            """245            If C{maxitems} is used to limit the number of records returned by246            C{list_resource_record_sets}, the records returned are those that247            sort first according to the rules given by248            U{http://docs.aws.amazon.com/Route53/latest/APIReference/API_ListResourceRecordSets.html#API_ListResourceRecordSets_RequestSyntax}.249            """250            zone_name = u"{}.example.invalid.".format(uuid4())251            client = get_client(self)252            # extra sorts _after_ expected according to the AWS Route53253            # ordering rules but it sorts _before_ according to more naive254            # (incorrect) string ordering rules.255            extra = RRSet(256                Name(u"a.z.{}".format(zone_name)),257                u"A",258                60,259                {A(IPv4Address(u"10.0.0.1"))},260            )261            expected = RRSet(262                Name(u"b.y.{}".format(zone_name)),263                u"A",264                60,265                {A(IPv4Address(u"10.0.0.2"))},266            )267            d = client.create_hosted_zone(u"{}".format(time()), zone_name)268            def created_zone(zone):269                self.addCleanup(lambda: self._cleanup(client, zone.identifier))270                d = client.change_resource_record_sets(zone.identifier, [271                    create_rrset(extra),272                    create_rrset(expected),273                ])274                d.addCallback(lambda ignored: zone)275                return d276            d.addCallback(created_zone)277            def created_rrsets(zone):278                return client.list_resource_record_sets(279                    zone.identifier,280                    name=Name(u"a.{}".format(zone_name)),281                    type=u"A",282                    maxitems=1,283                )284            d.addCallback(created_rrsets)285            def listed_rrsets(rrsets):286                self.assertEqual(287                    {RRSetKey(expected.label, expected.type): expected},288                    rrsets,289                )290            d.addCallback(listed_rrsets)291            return d292    return Route53IntegrationTests

route53.py

Source:route53.py

...34    def list_hosted_zones(self, **kwargs):35        zones = list_hosted_zones(**kwargs)36        return [zone for zone in zones if not self.check_ignore_list(zone.get('Name', ''))]37    @record_exception(source="route53-watcher")38    def list_resource_record_sets(self, **kwargs):39        record_sets = list_resource_record_sets(**kwargs)40        return [record for record in record_sets if not self.check_ignore_list(record.get('Name', ''))]41    @record_exception(source="route53-watcher")42    def process_item(self, **kwargs):43        zone = kwargs['zone']44        record = kwargs['record']45        records = record.get('ResourceRecords', [])46        records = [r.get('Value') for r in records]47        record_name = record.get('Name', '')48        config = {49            'zonename': zone.get('Name'),50            'zoneid':   zone.get('Id'),51            'zoneprivate': zone.get('Config', {}).get('PrivateZone', 'Unknown'),52            'name':     record_name,53            'type':     record.get('Type'),54            'records':  records,55            'ttl':      record.get('TTL'),56        }57        print(config)58        return Route53Record(account=kwargs['account_name'], name=record_name, config=dict(config), source_watcher=self)59    def slurp(self):60        """61        :returns: item_list - list of Route53 zones.62        :returns: exception_map - A dict where the keys are a tuple containing the63            location of the exception and the value is the actual exception64        """65        self.prep_for_slurp()66        @iter_account_region(index=self.index, accounts=self.accounts, exception_record_region='universal')67        def slurp_items(**kwargs):68            app.logger.debug("Checking {}/{}".format(self.index, kwargs['account_name']))69            item_list = []70            zones = self.list_hosted_zones(**kwargs)71            if not zones:72                return item_list, kwargs['exception_map']73            74            app.logger.debug('Slurped {len_zones} {plural} from {account}.'.format(75                len_zones=len(zones),76                plural=self.i_am_plural,77                account=kwargs['account_name']))78            for zone in zones:79                record_sets = self.list_resource_record_sets(Id=zone['Id'], **kwargs)80                if not record_sets:81                    continue82                app.logger.debug("Slurped %s %s within %s %s (%s) from %s" %83                    (len(record_sets), self.i_have_plural, self.i_am_singular, zone['Name'], zone['Id'],84                     kwargs['account_name']))85                try:86                    for record in record_sets:87                        item = self.process_item(name=record['Name'], record=record, zone=zone, **kwargs)88                        item_list.append(item)89                except Exception as e:90                    raise91            return item_list, kwargs['exception_map']92        return slurp_items()93class Route53Record(ChangeItem):...

route53_list-resource-record-sets.py

Source:route53_list-resource-record-sets.py

...7def list_hosted_zones():8    route53 = boto3.client('route53')910# aws route53 list_resource_record_sets --hosted-zone-id <hosted_zone_ID>11def list_resource_record_sets():12    route53 = boto3.client('route53')13    response = route53.list_resource_record_sets(HostedZoneId='Z0338876FY905DAOM13I')14    15    16    for ResourceRecordsCounts in response['ResourceRecordSets']:17        res = [ ResourceRecordsCounts['Name'],  ResourceRecordsCounts['Type'], ResourceRecordsCounts['ResourceRecords'], ResourceRecordsCounts['TTL'] ]18        connect_mysql(res)1920def truncate_mysql(*args):21    # DB Connection Data22    host = "localhost"23    username = "root"24    password = "fdsa1234"25    port = 330626    database = "aws_db"27    28    # Insert Query29    conn = pymysql.connect(host=host, user=username, passwd=password, port=port, database=database)30    truncate_query = "truncate table list_resource_record_sets"31    32    cursors = conn.cursor()33    cursors.execute(truncate_query)34    35def connect_mysql(*args):36    # Argument Parsing37    data = ''38    for i in range(len(args[0][2])):39        data += args[0][2][i]['Value']40    41    # DB Connection Data42    host = "localhost"43    username = "root"44    password = "fdsa1234"45    port = 330646    database = "db_route53"47    48    # Insert Query49    conn = pymysql.connect(host=host, user=username, passwd=password, port=port, database=database)50    51    52    query = "INSERT INTO test_table \53        (record_name, record_type, record_routing_table, record_value, record_ttl, record_hosted_id) VALUE \54        (%s, %s, 'routing_table', %s, %s , 'woobeom')"55        56    cursors = conn.cursor()57    cursors.execute( query, (args[0][0], args[0][1], data, args[0][3] ))    58     59    60    # Query Commit61    conn.commit()62    conn.close()6364    65if __name__ == "__main__":66    list_resource_record_sets()67    # testCode()
...

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.