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

How to use change_resource_record_sets method in localstack

Best Python code snippet using localstack_python

route53_tests.py

Source:route53_tests.py

...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)...

dns_route53_test.py

Source:dns_route53_test.py

1"""Tests for certbot_dns_route53.dns_route53.Authenticator"""2import unittest3import mock4from botocore.exceptions import NoCredentialsError, ClientError5from certbot import errors6from certbot.compat import os7from certbot.plugins import dns_test_common8from certbot.plugins.dns_test_common import DOMAIN9class AuthenticatorTest(unittest.TestCase, dns_test_common.BaseAuthenticatorTest):10    # pylint: disable=protected-access11    def setUp(self):12        from certbot_dns_route53.dns_route53 import Authenticator13        super(AuthenticatorTest, self).setUp()14        self.config = mock.MagicMock()15        # Set up dummy credentials for testing16        os.environ["AWS_ACCESS_KEY_ID"] = "dummy_access_key"17        os.environ["AWS_SECRET_ACCESS_KEY"] = "dummy_secret_access_key"18        self.auth = Authenticator(self.config, "route53")19    def tearDown(self):20        # Remove the dummy credentials from env vars21        del os.environ["AWS_ACCESS_KEY_ID"]22        del os.environ["AWS_SECRET_ACCESS_KEY"]23        super(AuthenticatorTest, self).tearDown()24    def test_perform(self):25        self.auth._change_txt_record = mock.MagicMock()26        self.auth._wait_for_change = mock.MagicMock()27        self.auth.perform([self.achall])28        self.auth._change_txt_record.assert_called_once_with("UPSERT",29                                                             '_acme-challenge.' + DOMAIN,30                                                             mock.ANY)31        self.assertEqual(self.auth._wait_for_change.call_count, 1)32    def test_perform_no_credentials_error(self):33        self.auth._change_txt_record = mock.MagicMock(side_effect=NoCredentialsError)34        self.assertRaises(errors.PluginError,35                          self.auth.perform,36                          [self.achall])37    def test_perform_client_error(self):38        self.auth._change_txt_record = mock.MagicMock(39            side_effect=ClientError({"Error": {"Code": "foo"}}, "bar"))40        self.assertRaises(errors.PluginError,41                          self.auth.perform,42                          [self.achall])43    def test_cleanup(self):44        self.auth._attempt_cleanup = True45        self.auth._change_txt_record = mock.MagicMock()46        self.auth.cleanup([self.achall])47        self.auth._change_txt_record.assert_called_once_with("DELETE",48                                                             '_acme-challenge.'+DOMAIN,49                                                             mock.ANY)50    def test_cleanup_no_credentials_error(self):51        self.auth._attempt_cleanup = True52        self.auth._change_txt_record = mock.MagicMock(side_effect=NoCredentialsError)53        self.auth.cleanup([self.achall])54    def test_cleanup_client_error(self):55        self.auth._attempt_cleanup = True56        self.auth._change_txt_record = mock.MagicMock(57            side_effect=ClientError({"Error": {"Code": "foo"}}, "bar"))58        self.auth.cleanup([self.achall])59class ClientTest(unittest.TestCase):60    # pylint: disable=protected-access61    PRIVATE_ZONE = {62                        "Id": "BAD-PRIVATE",63                        "Name": "example.com",64                        "Config": {65                            "PrivateZone": True66                        }67                    }68    EXAMPLE_NET_ZONE = {69                            "Id": "BAD-WRONG-TLD",70                            "Name": "example.net",71                            "Config": {72                                "PrivateZone": False73                            }74                        }75    EXAMPLE_COM_ZONE = {76                            "Id": "EXAMPLE",77                            "Name": "example.com",78                            "Config": {79                                "PrivateZone": False80                            }81                        }82    FOO_EXAMPLE_COM_ZONE = {83                                "Id": "FOO",84                                "Name": "foo.example.com",85                                "Config": {86                                    "PrivateZone": False87                                }88                            }89    def setUp(self):90        from certbot_dns_route53.dns_route53 import Authenticator91        super(ClientTest, self).setUp()92        self.config = mock.MagicMock()93        # Set up dummy credentials for testing94        os.environ["AWS_ACCESS_KEY_ID"] = "dummy_access_key"95        os.environ["AWS_SECRET_ACCESS_KEY"] = "dummy_secret_access_key"96        self.client = Authenticator(self.config, "route53")97    def tearDown(self):98        # Remove the dummy credentials from env vars99        del os.environ["AWS_ACCESS_KEY_ID"]100        del os.environ["AWS_SECRET_ACCESS_KEY"]101        super(ClientTest, self).tearDown()102    def test_find_zone_id_for_domain(self):103        self.client.r53.get_paginator = mock.MagicMock()104        self.client.r53.get_paginator().paginate.return_value = [105            {106                "HostedZones": [107                    self.EXAMPLE_NET_ZONE,108                    self.EXAMPLE_COM_ZONE,109                ]110            }111        ]112        result = self.client._find_zone_id_for_domain("foo.example.com")113        self.assertEqual(result, "EXAMPLE")114    def test_find_zone_id_for_domain_pagination(self):115        self.client.r53.get_paginator = mock.MagicMock()116        self.client.r53.get_paginator().paginate.return_value = [117            {118                "HostedZones": [119                    self.PRIVATE_ZONE,120                    self.EXAMPLE_COM_ZONE,121                ]122            },123            {124                "HostedZones": [125                    self.PRIVATE_ZONE,126                    self.FOO_EXAMPLE_COM_ZONE,127                ]128            }129        ]130        result = self.client._find_zone_id_for_domain("foo.example.com")131        self.assertEqual(result, "FOO")132    def test_find_zone_id_for_domain_no_results(self):133        self.client.r53.get_paginator = mock.MagicMock()134        self.client.r53.get_paginator().paginate.return_value = []135        self.assertRaises(errors.PluginError,136                          self.client._find_zone_id_for_domain,137                          "foo.example.com")138    def test_find_zone_id_for_domain_no_correct_results(self):139        self.client.r53.get_paginator = mock.MagicMock()140        self.client.r53.get_paginator().paginate.return_value = [141            {142                "HostedZones": [143                    self.PRIVATE_ZONE,144                    self.EXAMPLE_NET_ZONE,145                ]146            },147        ]148        self.assertRaises(errors.PluginError,149                          self.client._find_zone_id_for_domain,150                          "foo.example.com")151    def test_change_txt_record(self):152        self.client._find_zone_id_for_domain = mock.MagicMock()153        self.client.r53.change_resource_record_sets = mock.MagicMock(154            return_value={"ChangeInfo": {"Id": 1}})155        self.client._change_txt_record("FOO", DOMAIN, "foo")156        call_count = self.client.r53.change_resource_record_sets.call_count157        self.assertEqual(call_count, 1)158    def test_change_txt_record_delete(self):159        self.client._find_zone_id_for_domain = mock.MagicMock()160        self.client.r53.change_resource_record_sets = mock.MagicMock(161            return_value={"ChangeInfo": {"Id": 1}})162        validation = "some-value"163        validation_record = {"Value": '"{0}"'.format(validation)}164        self.client._resource_records[DOMAIN] = [validation_record]165        self.client._change_txt_record("DELETE", DOMAIN, validation)166        call_count = self.client.r53.change_resource_record_sets.call_count167        self.assertEqual(call_count, 1)168        call_args = self.client.r53.change_resource_record_sets.call_args_list[0][1]169        call_args_batch = call_args["ChangeBatch"]["Changes"][0]170        self.assertEqual(call_args_batch["Action"], "DELETE")171        self.assertEqual(172            call_args_batch["ResourceRecordSet"]["ResourceRecords"],173            [validation_record])174    def test_change_txt_record_multirecord(self):175        self.client._find_zone_id_for_domain = mock.MagicMock()176        self.client._get_validation_rrset = mock.MagicMock()177        self.client._resource_records[DOMAIN] = [178            {"Value": "\"pre-existing-value\""},179            {"Value": "\"pre-existing-value-two\""},180        ]181        self.client.r53.change_resource_record_sets = mock.MagicMock(182            return_value={"ChangeInfo": {"Id": 1}})183        self.client._change_txt_record("DELETE", DOMAIN, "pre-existing-value")184        call_count = self.client.r53.change_resource_record_sets.call_count185        call_args = self.client.r53.change_resource_record_sets.call_args_list[0][1]186        call_args_batch = call_args["ChangeBatch"]["Changes"][0]187        self.assertEqual(call_args_batch["Action"], "UPSERT")188        self.assertEqual(189            call_args_batch["ResourceRecordSet"]["ResourceRecords"],190            [{"Value": "\"pre-existing-value-two\""}])191        self.assertEqual(call_count, 1)192    def test_wait_for_change(self):193        self.client.r53.get_change = mock.MagicMock(194            side_effect=[{"ChangeInfo": {"Status": "PENDING"}},195                         {"ChangeInfo": {"Status": "INSYNC"}}])196        self.client._wait_for_change(1)197        self.assertTrue(self.client.r53.get_change.called)198if __name__ == "__main__":...

disaster_recovery_failover_script_lambda.py

Source:disaster_recovery_failover_script_lambda.py

...15    except Exception as error:16        print(error)17def update_route53_aliastarget(Name,SetID,weight,hostedzoneid,Record,TTL):18    try: 19        response = route53.change_resource_record_sets(20            HostedZoneId='Z063006017GGX7X3V0BUZ', #your own hosted zone id21            ChangeBatch={22                'Comment': 'DR event for SprintQA DevOps team apps from east to west.',23                'Changes': [24                    {25                        'Action': 'UPSERT',26                        'ResourceRecordSet': {27                            'Name': Name,28                            'Type': 'A',29                            'SetIdentifier': SetID,30                            'Weight': weight,31                            'AliasTarget': {32                                'HostedZoneId': hostedzoneid, #aws elasticbeanstalk hosted zone id, maintained by aws 33                                'DNSName': Record,34                                'EvaluateTargetHealth': True35                            }36                        }37                    }38                ]39            }40        )41    except Exception as error:42        print(error)43#update_route53_aliastarget(Name,SetID,weight,Record,hostedzoneid)44def lambda_handler(event, context):45    failover_from =os.environ["Failover_from"]46    failover_to =os.environ["Failover_to"]47    Dynamo =os.environ["DynamoTableName"]48    table = dynamo.Table(Dynamo) #line 26 var name49    scan_response = table.scan(AttributesToGet=["Records","Name","Weight","Type","SetIdentifier","TTL"])50    for i in scan_response["Items"]:51        Weight = i["Weight"] # Weight from the table52        Type = i["Type"]53        SetID = i["SetIdentifier"]54        TTL = int(i["TTL"])55        Record = i["Records"]56        Name = i["Name"]57        58        if SetID=='2':59            app_color='blue'60        elif SetID=='1':61            app_color='green'62        if failover_from=='East' and failover_to=='West':63            print('failing over from East to West')64            if app_color=='green':65                Weight=066                if Type == "ALIAS":67                    weight = 0 # weight from the Route53 record set 68                    hostedzoneid = "Z117KPS5GTRQ2G" #ElasticBeanstalk us-east-1 Route53 hosted zone ID69                    print("Calling Update ALIAS function")70                    print(Name)71                    update_route53_aliastarget(Name,SetID,weight,hostedzoneid,Record,TTL)72                else:73                    response = route53.change_resource_record_sets(HostedZoneId="Z063006017GGX7X3V0BUZ", # your public hosted zone id74                            ChangeBatch = {75                                "Comment" : "",76                                "Changes": [77                                    {78                                        "Action" : "UPSERT",79                                        "ResourceRecordSet" : {80                                            "Name": Name,81                                            "Type" : Type,82                                            "Weight" : 0,83                                            "TTL" : TTL,84                                            "SetIdentifier": SetID,85                                            "ResourceRecords":[{"Value":Record}]86                                            }87                                        }88                                    ]89                                }90                            )91                update_dynamo(Name, Record, Weight, Dynamo)92            elif app_color=='blue':93                Weight=194                if Type == "ALIAS":95                    weight = 1 # weight from the Route53 record set 96                    hostedzoneid = "Z38NKT9BP95V3O" #ElasticBeanstalk us-west-2 Route53 hosted zone ID97                    print("Calling Update ALIAS function")98                    print(Name)99                    update_route53_aliastarget(Name,SetID,weight,hostedzoneid,Record,TTL)100                else:101                    response = route53.change_resource_record_sets(HostedZoneId="Z063006017GGX7X3V0BUZ", # your public hosted zone id102                            ChangeBatch = {103                                "Comment" : "",104                                "Changes": [105                                    {106                                        "Action" : "UPSERT",107                                        "ResourceRecordSet" : {108                                            "Name": Name,109                                            "Type" : Type,110                                            "Weight" : 1,111                                            "TTL" : TTL,112                                            "SetIdentifier": SetID,113                                            "ResourceRecords":[{"Value":Record}]114                                            }115                                        }116                                    ]117                                }118                            )119                update_dynamo(Name, Record, Weight, Dynamo)120            else:121                pass122        elif failover_from=='West' and failover_to=='East':123            print('failing over from West to East')124            if app_color=='green':125                Weight=1126                if Type == "ALIAS":127                    weight = 1 # weight from the Route53 record set 128                    hostedzoneid = "Z117KPS5GTRQ2G" #ElasticBeanstalk us-east-1 Route53 hosted zone ID129                    print("Calling Update ALIAS function")130                    print(Name)131                    update_route53_aliastarget(Name,SetID,weight,hostedzoneid,Record,TTL)132                else:133                    response = route53.change_resource_record_sets(HostedZoneId="Z063006017GGX7X3V0BUZ", # your public hosted zone id134                            ChangeBatch = {135                                "Comment" : "",136                                "Changes": [137                                    {138                                        "Action" : "UPSERT",139                                        "ResourceRecordSet" : {140                                            "Name": Name,141                                            "Type" : Type,142                                            "Weight" : 1,143                                            "TTL" : TTL,144                                            "SetIdentifier": SetID,145                                            "ResourceRecords":[{"Value":Record}]146                                            }147                                        }148                                    ]149                                }150                            )151                update_dynamo(Name, Record, Weight, Dynamo)152            elif app_color=='blue':153                Weight=0154                if Type == "ALIAS":155                    weight = 0 # weight from the Route53 record set 156                    hostedzoneid = "Z38NKT9BP95V3O" #ElasticBeanstalk us-west-2 Route53 hosted zone ID157                    print("Calling Update ALIAS function")158                    print(Name)159                    update_route53_aliastarget(Name,SetID,weight,hostedzoneid,Record,TTL)160                else:161                    response = route53.change_resource_record_sets(HostedZoneId="Z063006017GGX7X3V0BUZ", # your public hosted zone id162                            ChangeBatch = {163                                "Comment" : "",164                                "Changes": [165                                    {166                                        "Action" : "UPSERT",167                                        "ResourceRecordSet" : {168                                            "Name": Name,169                                            "Type" : Type,170                                            "Weight" : 0,171                                            "TTL" : TTL,172                                            "SetIdentifier": SetID,173                                            "ResourceRecords":[{"Value":Record}]174                                            }175                                        }...

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.