KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
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?
