Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Techdebt: Replace sure with regular assertions in Route53 (#6497)

Browse Source
This commit is contained in:
Bert Blommers 2023-07-08 21:53:09 +00:00 committed by GitHub
parent 39313ffc5b
commit cd906cea56
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 373 additions and 422 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
tests/test_route53/test_change_set_model.py
View File

@ -1,5 +1,4 @@
import copy
import sure # noqa # pylint: disable=unused-import
from moto.route53.models import ChangeList
@ -25,10 +24,10 @@ def test_last_dot_in_name_is_ignored():
change_list.append(change1)
change_list.append(change2)
change_list.should.contain(change1)
change_list.should.contain(change1_alt)
change_list.should.contain(change2)
change_list.should.contain(change2_alt)
assert change1 in change_list
assert change1_alt in change_list
assert change2 in change_list
assert change2_alt in change_list
def test_last_dot_is_not_stored():
@ -44,7 +43,7 @@ def test_last_dot_is_not_stored():
change_list = ChangeList()
change_list.append(change1)
change_list[0]["ResourceRecordSet"]["Name"].should.equal("test.google.com")
assert change_list[0]["ResourceRecordSet"]["Name"] == "test.google.com"
def test_optional_fields():
@ -55,4 +54,4 @@ def test_optional_fields():
change_list = ChangeList()
change_list.append(change)
change_list.should.equal([change])
assert change_list == [change]

373
tests/test_route53/test_route53.py
View File

@ -1,12 +1,10 @@
import boto3
from botocore.exceptions import ClientError
import sure # noqa # pylint: disable=unused-import
import botocore
import pytest
import requests
from botocore.exceptions import ClientError
from moto import mock_ec2, mock_route53, settings
@ -17,17 +15,17 @@ def test_create_hosted_zone():
Name="testdns.aws.com.", CallerReference=str(hash("foo"))
)
firstzone = response["HostedZone"]
firstzone.should.have.key("Id").match(r"/hostedzone/[A-Z0-9]+")
firstzone.should.have.key("Name").equal("testdns.aws.com.")
firstzone.should.have.key("Config").equal({"PrivateZone": False})
firstzone.should.have.key("ResourceRecordSetCount").equal(2)
assert "/hostedzone/" in firstzone["Id"]
assert firstzone["Name"] == "testdns.aws.com."
assert firstzone["Config"] == {"PrivateZone": False}
assert firstzone["ResourceRecordSetCount"] == 2
delegation = response["DelegationSet"]
delegation.should.have.key("NameServers").length_of(4)
delegation["NameServers"].should.contain("ns-2048.awsdns-64.com")
delegation["NameServers"].should.contain("ns-2049.awsdns-65.net")
delegation["NameServers"].should.contain("ns-2050.awsdns-66.org")
delegation["NameServers"].should.contain("ns-2051.awsdns-67.co.uk")
assert len(delegation["NameServers"]) == 4
assert "ns-2048.awsdns-64.com" in delegation["NameServers"]
assert "ns-2049.awsdns-65.net" in delegation["NameServers"]
assert "ns-2050.awsdns-66.org" in delegation["NameServers"]
assert "ns-2051.awsdns-67.co.uk" in delegation["NameServers"]
location = response["Location"]
if not settings.TEST_SERVER_MODE:
@ -39,7 +37,7 @@ def test_list_hosted_zones():
conn = boto3.client("route53", region_name="us-east-1")
res = conn.list_hosted_zones()["HostedZones"]
res.should.have.length_of(0)
assert len(res) == 0
zone1 = conn.create_hosted_zone(
Name="testdns1.aws.com.", CallerReference=str(hash("foo"))
@ -49,10 +47,10 @@ def test_list_hosted_zones():
)["HostedZone"]
res = conn.list_hosted_zones()["HostedZones"]
res.should.have.length_of(2)
assert len(res) == 2
res.should.contain(zone1)
res.should.contain(zone2)
assert zone1 in res
assert zone2 in res
@mock_route53
@ -67,7 +65,7 @@ def test_delete_hosted_zone():
conn.delete_hosted_zone(Id=zone1["Id"])
res = conn.list_hosted_zones()["HostedZones"]
res.should.have.length_of(1)
assert len(res) == 1
@mock_route53
@ -97,9 +95,10 @@ def test_delete_hosted_zone_with_change_sets():
with pytest.raises(ClientError) as exc:
conn.delete_hosted_zone(Id=zone_id)
err = exc.value.response["Error"]
err["Code"].should.equal("HostedZoneNotEmpty")
err["Message"].should.equal(
"The hosted zone contains resource records that are not SOA or NS records."
assert err["Code"] == "HostedZoneNotEmpty"
assert (
err["Message"]
== "The hosted zone contains resource records that are not SOA or NS records."
)
@ -107,9 +106,7 @@ def test_delete_hosted_zone_with_change_sets():
def test_get_hosted_zone_count_no_zones():
conn = boto3.client("route53", region_name="us-east-1")
zone_count = conn.get_hosted_zone_count()
zone_count.should.have.key("HostedZoneCount")
isinstance(zone_count["HostedZoneCount"], int).should.be.true
zone_count["HostedZoneCount"].should.be.equal(0)
assert zone_count["HostedZoneCount"] == 0
@mock_route53
@ -123,9 +120,7 @@ def test_get_hosted_zone_count_one_zone():
HostedZoneConfig=dict(PrivateZone=False, Comment=f"test {zone} com"),
)
zone_count = conn.get_hosted_zone_count()
zone_count.should.have.key("HostedZoneCount")
isinstance(zone_count["HostedZoneCount"], int).should.be.true
zone_count["HostedZoneCount"].should.be.equal(1)
assert zone_count["HostedZoneCount"] == 1
@mock_route53
@ -148,10 +143,7 @@ def test_get_hosted_zone_count_many_zones():
HostedZoneConfig=dict(PrivateZone=False, Comment=f"test {zone} com"),
)
zone_count = conn.get_hosted_zone_count()
zone_count.should.have.key("HostedZoneCount")
isinstance(zone_count["HostedZoneCount"], int).should.be.true
zone_count["HostedZoneCount"].shouldnt.be.equal(0)
zone_count["HostedZoneCount"].should.be.equal(len(zone_indexes))
assert zone_count["HostedZoneCount"] == len(zone_indexes)
@mock_route53
@ -162,8 +154,8 @@ def test_get_unknown_hosted_zone():
conn.get_hosted_zone(Id="unknown")
err = ex.value.response["Error"]
err["Code"].should.equal("NoSuchHostedZone")
err["Message"].should.equal("No hosted zone found with ID: unknown")
assert err["Code"] == "NoSuchHostedZone"
assert err["Message"] == "No hosted zone found with ID: unknown"
@mock_route53
@ -177,7 +169,7 @@ def test_update_hosted_zone_comment():
conn.update_hosted_zone_comment(Id=zone_id, Comment="yolo")
resp = conn.get_hosted_zone(Id=zone_id)["HostedZone"]
resp["Config"].should.have.key("Comment").equals("yolo")
assert resp["Config"]["Comment"] == "yolo"
@mock_route53
@ -188,8 +180,8 @@ def test_list_resource_record_set_unknown_zone():
conn.list_resource_record_sets(HostedZoneId="abcd")
err = ex.value.response["Error"]
err["Code"].should.equal("NoSuchHostedZone")
err["Message"].should.equal("No hosted zone found with ID: abcd")
assert err["Code"] == "NoSuchHostedZone"
assert err["Message"] == "No hosted zone found with ID: abcd"
@mock_route53
@ -203,8 +195,8 @@ def test_list_resource_record_set_unknown_type():
conn.list_resource_record_sets(HostedZoneId=zone["Id"], StartRecordType="A")
err = ex.value.response["Error"]
err["Code"].should.equal("400")
err["Message"].should.equal("Bad Request")
assert err["Code"] == "400"
assert err["Message"] == "Bad Request"
@mock_route53
@ -249,8 +241,8 @@ def test_use_health_check_in_resource_record_set():
record_sets = conn.list_resource_record_sets(HostedZoneId=zone_id)[
"ResourceRecordSets"
]
record_sets[2]["Name"].should.equal("foo.bar.testdns.aws.com.")
record_sets[2]["HealthCheckId"].should.equal(check_id)
assert record_sets[2]["Name"] == "foo.bar.testdns.aws.com."
assert record_sets[2]["HealthCheckId"] == check_id
@mock_route53
@ -265,10 +257,10 @@ def test_hosted_zone_comment_preserved():
zone_id = firstzone["HostedZone"]["Id"]
hosted_zone = conn.get_hosted_zone(Id=zone_id)
hosted_zone["HostedZone"]["Config"]["Comment"].should.equal("test comment")
assert hosted_zone["HostedZone"]["Config"]["Comment"] == "test comment"
hosted_zones = conn.list_hosted_zones()
hosted_zones["HostedZones"][0]["Config"]["Comment"].should.equal("test comment")
assert hosted_zones["HostedZones"][0]["Config"]["Comment"] == "test comment"
@mock_route53
@ -301,7 +293,7 @@ def test_deleting_weighted_route():
cnames = conn.list_resource_record_sets(
HostedZoneId=zone_id, StartRecordName="cname", StartRecordType="CNAME"
)["ResourceRecordSets"]
cnames.should.have.length_of(4)
assert len(cnames) == 4
conn.change_resource_record_sets(
HostedZoneId=zone_id,
@ -323,9 +315,9 @@ def test_deleting_weighted_route():
cnames = conn.list_resource_record_sets(
HostedZoneId=zone_id, StartRecordName="cname", StartRecordType="CNAME"
)["ResourceRecordSets"]
cnames.should.have.length_of(3)
cnames[-1]["Name"].should.equal("cname.testdns.aws.com.")
cnames[-1]["SetIdentifier"].should.equal("success-test-bar")
assert len(cnames) == 3
assert cnames[-1]["Name"] == "cname.testdns.aws.com."
assert cnames[-1]["SetIdentifier"] == "success-test-bar"
@mock_route53
@ -362,13 +354,13 @@ def test_deleting_latency_route():
cnames = conn.list_resource_record_sets(
HostedZoneId=zone_id, StartRecordName="cname", StartRecordType="CNAME"
)["ResourceRecordSets"]
cnames.should.have.length_of(4)
assert len(cnames) == 4
foo_cname = [
cname
for cname in cnames
if cname.get("SetIdentifier") and cname["SetIdentifier"] == "success-test-foo"
][0]
foo_cname["Region"].should.equal("us-west-2")
assert foo_cname["Region"] == "us-west-2"
conn.change_resource_record_sets(
HostedZoneId=zone_id,
@ -390,9 +382,9 @@ def test_deleting_latency_route():
cnames = conn.list_resource_record_sets(
HostedZoneId=zone_id, StartRecordName="cname", StartRecordType="CNAME"
)["ResourceRecordSets"]
cnames.should.have.length_of(3)
cnames[-1]["SetIdentifier"].should.equal("success-test-bar")
cnames[-1]["Region"].should.equal("us-west-1")
assert len(cnames) == 3
assert cnames[-1]["SetIdentifier"] == "success-test-bar"
assert cnames[-1]["Region"] == "us-west-1"
@mock_route53
@ -416,7 +408,7 @@ def test_list_or_change_tags_for_resource_request():
response = conn.list_tags_for_resource(
ResourceType="healthcheck", ResourceId=healthcheck_id
)
response["ResourceTagSet"]["Tags"].should.equal([])
assert response["ResourceTagSet"]["Tags"] == []
tag1 = {"Key": "Deploy", "Value": "True"}
tag2 = {"Key": "Name", "Value": "UnitTest"}
@ -430,13 +422,13 @@ def test_list_or_change_tags_for_resource_request():
response = conn.list_tags_for_resource(
ResourceType="healthcheck", ResourceId=healthcheck_id
)
response.should.contain("ResourceTagSet")
assert "ResourceTagSet" in response
# Validate that each key was added
response["ResourceTagSet"]["Tags"].should.contain(tag1)
response["ResourceTagSet"]["Tags"].should.contain(tag2)
assert tag1 in response["ResourceTagSet"]["Tags"]
assert tag2 in response["ResourceTagSet"]["Tags"]
len(response["ResourceTagSet"]["Tags"]).should.equal(2)
assert len(response["ResourceTagSet"]["Tags"]) == 2
# Try to remove the tags
conn.change_tags_for_resource(
@ -449,9 +441,9 @@ def test_list_or_change_tags_for_resource_request():
response = conn.list_tags_for_resource(
ResourceType="healthcheck", ResourceId=healthcheck_id
)
response.should.contain("ResourceTagSet")
response["ResourceTagSet"]["Tags"].should_not.contain(tag1)
response["ResourceTagSet"]["Tags"].should.contain(tag2)
assert "ResourceTagSet" in response
assert tag1 not in response["ResourceTagSet"]["Tags"]
assert tag2 in response["ResourceTagSet"]["Tags"]
# Remove the second tag
conn.change_tags_for_resource(
@ -463,7 +455,7 @@ def test_list_or_change_tags_for_resource_request():
response = conn.list_tags_for_resource(
ResourceType="healthcheck", ResourceId=healthcheck_id
)
response["ResourceTagSet"]["Tags"].should_not.contain(tag2)
assert tag2 not in response["ResourceTagSet"]["Tags"]
# Re-add the tags
conn.change_tags_for_resource(
@ -480,7 +472,7 @@ def test_list_or_change_tags_for_resource_request():
response = conn.list_tags_for_resource(
ResourceType="healthcheck", ResourceId=healthcheck_id
)
response["ResourceTagSet"]["Tags"].should.equal([])
assert response["ResourceTagSet"]["Tags"] == []
@mock_ec2
@ -500,11 +492,9 @@ def test_list_hosted_zones_by_name():
)
zone_b = conn.list_hosted_zones_by_name(DNSName="test.b.com.")
len(zone_b["HostedZones"]).should.equal(1)
zone_b["HostedZones"][0]["Name"].should.equal("test.b.com.")
zone_b["HostedZones"][0].should.have.key("Config")
zone_b["HostedZones"][0]["Config"].should.have.key("PrivateZone")
zone_b["HostedZones"][0]["Config"]["PrivateZone"].should.be.equal(True)
assert len(zone_b["HostedZones"]) == 1
assert zone_b["HostedZones"][0]["Name"] == "test.b.com."
assert zone_b["HostedZones"][0]["Config"]["PrivateZone"]
# We declared this a a private hosted zone above, so let's make
# sure it really is!
@ -512,22 +502,14 @@ def test_list_hosted_zones_by_name():
b_hosted_zone = conn.get_hosted_zone(Id=zone_b_id)
# Pull the HostedZone block out and test it.
b_hosted_zone.should.have.key("HostedZone")
b_hz = b_hosted_zone["HostedZone"]
b_hz.should.have.key("Config")
b_hz["Config"].should.have.key("PrivateZone")
b_hz["Config"]["PrivateZone"].should.be.equal(True)
assert b_hz["Config"]["PrivateZone"]
# Check for the VPCs block since this *should* be a VPC-Private Zone
b_hosted_zone.should.have.key("VPCs")
b_hosted_zone["VPCs"].should.have.length_of(1)
assert len(b_hosted_zone["VPCs"]) == 1
b_hz_vpcs = b_hosted_zone["VPCs"][0]
b_hz_vpcs.should.have.key("VPCId")
b_hz_vpcs.should.have.key("VPCRegion")
b_hz_vpcs["VPCId"].should_not.equal("")
b_hz_vpcs["VPCRegion"].should_not.equal("")
b_hz_vpcs["VPCId"].should.be.equal(vpc_id)
b_hz_vpcs["VPCRegion"].should.be.equal(region)
assert b_hz_vpcs["VPCId"] == vpc_id
assert b_hz_vpcs["VPCRegion"] == region
# Now create other zones and test them.
conn.create_hosted_zone(
@ -543,23 +525,19 @@ def test_list_hosted_zones_by_name():
# Now makes sure the other zones we created above are NOT private...
zones = conn.list_hosted_zones_by_name(DNSName="test.a.org.")
len(zones["HostedZones"]).should.equal(2)
zones["HostedZones"][0]["Name"].should.equal("test.a.org.")
zones["HostedZones"][0].should.have.key("Config")
zones["HostedZones"][0]["Config"].should.have.key("PrivateZone")
zones["HostedZones"][0]["Config"]["PrivateZone"].should.be.equal(False)
assert len(zones["HostedZones"]) == 2
assert zones["HostedZones"][0]["Name"] == "test.a.org."
assert zones["HostedZones"][0]["Config"]["PrivateZone"] is False
zones["HostedZones"][1]["Name"].should.equal("test.a.org.")
zones["HostedZones"][1].should.have.key("Config")
zones["HostedZones"][1]["Config"].should.have.key("PrivateZone")
zones["HostedZones"][1]["Config"]["PrivateZone"].should.be.equal(False)
assert zones["HostedZones"][1]["Name"] == "test.a.org."
assert zones["HostedZones"][1]["Config"]["PrivateZone"] is False
# test sort order
zones = conn.list_hosted_zones_by_name()
len(zones["HostedZones"]).should.equal(3)
zones["HostedZones"][0]["Name"].should.equal("test.b.com.")
zones["HostedZones"][1]["Name"].should.equal("test.a.org.")
zones["HostedZones"][2]["Name"].should.equal("test.a.org.")
assert len(zones["HostedZones"]) == 3
assert zones["HostedZones"][0]["Name"] == "test.b.com."
assert zones["HostedZones"][1]["Name"] == "test.a.org."
assert zones["HostedZones"][2]["Name"] == "test.a.org."
@mock_route53
@ -588,26 +566,25 @@ def test_list_hosted_zones_by_dns_name():
# test lookup
zones = conn.list_hosted_zones_by_name(DNSName="test.b.com.")
len(zones["HostedZones"]).should.equal(1)
zones["DNSName"].should.equal("test.b.com.")
assert len(zones["HostedZones"]) == 1
assert zones["DNSName"] == "test.b.com."
zones = conn.list_hosted_zones_by_name(DNSName="test.a.org.")
len(zones["HostedZones"]).should.equal(2)
zones["DNSName"].should.equal("test.a.org.")
zones["DNSName"].should.equal("test.a.org.")
assert len(zones["HostedZones"]) == 2
assert zones["DNSName"] == "test.a.org."
zones = conn.list_hosted_zones_by_name(DNSName="my.test.net.")
len(zones["HostedZones"]).should.equal(1)
zones["DNSName"].should.equal("my.test.net.")
assert len(zones["HostedZones"]) == 1
assert zones["DNSName"] == "my.test.net."
zones = conn.list_hosted_zones_by_name(DNSName="my.test.net")
len(zones["HostedZones"]).should.equal(1)
zones["DNSName"].should.equal("my.test.net.")
assert len(zones["HostedZones"]) == 1
assert zones["DNSName"] == "my.test.net."
# test sort order
zones = conn.list_hosted_zones_by_name()
len(zones["HostedZones"]).should.equal(4)
zones["HostedZones"][0]["Name"].should.equal("test.b.com.")
zones["HostedZones"][1]["Name"].should.equal("my.test.net.")
zones["HostedZones"][2]["Name"].should.equal("test.a.org.")
zones["HostedZones"][3]["Name"].should.equal("test.a.org.")
assert len(zones["HostedZones"]) == 4
assert zones["HostedZones"][0]["Name"] == "test.b.com."
assert zones["HostedZones"][1]["Name"] == "my.test.net."
assert zones["HostedZones"][2]["Name"] == "test.a.org."
assert zones["HostedZones"][3]["Name"] == "test.a.org."
@mock_route53
@ -620,8 +597,8 @@ def test_change_resource_record_sets_crud_valid():
)
zones = conn.list_hosted_zones_by_name(DNSName="db.")
len(zones["HostedZones"]).should.equal(1)
zones["HostedZones"][0]["Name"].should.equal("db.")
assert len(zones["HostedZones"]) == 1
assert zones["HostedZones"][0]["Name"] == "db."
hosted_zone_id = zones["HostedZones"][0]["Id"]
# Create A Record.
@ -644,12 +621,12 @@ def test_change_resource_record_sets_crud_valid():
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(3)
assert len(response["ResourceRecordSets"]) == 3
a_record_detail = response["ResourceRecordSets"][2]
a_record_detail["Name"].should.equal("prod.redis.db.")
a_record_detail["Type"].should.equal("A")
a_record_detail["TTL"].should.equal(10)
a_record_detail["ResourceRecords"].should.equal([{"Value": "127.0.0.1"}])
assert a_record_detail["Name"] == "prod.redis.db."
assert a_record_detail["Type"] == "A"
assert a_record_detail["TTL"] == 10
assert a_record_detail["ResourceRecords"] == [{"Value": "127.0.0.1"}]
# Update A Record.
cname_record_endpoint_payload = {
@ -671,12 +648,12 @@ def test_change_resource_record_sets_crud_valid():
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(3)
assert len(response["ResourceRecordSets"]) == 3
cname_record_detail = response["ResourceRecordSets"][2]
cname_record_detail["Name"].should.equal("prod.redis.db.")
cname_record_detail["Type"].should.equal("A")
cname_record_detail["TTL"].should.equal(60)
cname_record_detail["ResourceRecords"].should.equal([{"Value": "192.168.1.1"}])
assert cname_record_detail["Name"] == "prod.redis.db."
assert cname_record_detail["Type"] == "A"
assert cname_record_detail["TTL"] == 60
assert cname_record_detail["ResourceRecords"] == [{"Value": "192.168.1.1"}]
# Update to add Alias.
cname_alias_record_endpoint_payload = {
@ -703,17 +680,15 @@ def test_change_resource_record_sets_crud_valid():
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
cname_alias_record_detail = response["ResourceRecordSets"][2]
cname_alias_record_detail["Name"].should.equal("prod.redis.db.")
cname_alias_record_detail["Type"].should.equal("A")
cname_alias_record_detail["TTL"].should.equal(60)
cname_alias_record_detail["AliasTarget"].should.equal(
{
"HostedZoneId": hosted_zone_id,
"DNSName": "prod.redis.alias.",
"EvaluateTargetHealth": False,
}
)
cname_alias_record_detail.should_not.contain("ResourceRecords")
assert cname_alias_record_detail["Name"] == "prod.redis.db."
assert cname_alias_record_detail["Type"] == "A"
assert cname_alias_record_detail["TTL"] == 60
assert cname_alias_record_detail["AliasTarget"] == {
"HostedZoneId": hosted_zone_id,
"DNSName": "prod.redis.alias.",
"EvaluateTargetHealth": False,
}
assert "ResourceRecords" not in cname_alias_record_detail
# Delete record.
delete_payload = {
@ -734,7 +709,7 @@ def test_change_resource_record_sets_crud_valid():
HostedZoneId=hosted_zone_id, ChangeBatch=delete_payload
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(2)
assert len(response["ResourceRecordSets"]) == 2
@mock_route53
@ -747,8 +722,8 @@ def test_change_resource_record_sets_crud_valid_with_special_xml_chars():
)
zones = conn.list_hosted_zones_by_name(DNSName="db.")
len(zones["HostedZones"]).should.equal(1)
zones["HostedZones"][0]["Name"].should.equal("db.")
assert len(zones["HostedZones"]) == 1
assert zones["HostedZones"][0]["Name"] == "db."
hosted_zone_id = zones["HostedZones"][0]["Id"]
# Create TXT Record.
@ -771,12 +746,12 @@ def test_change_resource_record_sets_crud_valid_with_special_xml_chars():
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(3)
assert len(response["ResourceRecordSets"]) == 3
a_record_detail = response["ResourceRecordSets"][2]
a_record_detail["Name"].should.equal("prod.redis.db.")
a_record_detail["Type"].should.equal("TXT")
a_record_detail["TTL"].should.equal(10)
a_record_detail["ResourceRecords"].should.equal([{"Value": "SomeInitialValue"}])
assert a_record_detail["Name"] == "prod.redis.db."
assert a_record_detail["Type"] == "TXT"
assert a_record_detail["TTL"] == 10
assert a_record_detail["ResourceRecords"] == [{"Value": "SomeInitialValue"}]
# Update TXT Record with XML Special Character &.
txt_record_with_special_char_endpoint_payload = {
@ -799,14 +774,14 @@ def test_change_resource_record_sets_crud_valid_with_special_xml_chars():
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(3)
assert len(response["ResourceRecordSets"]) == 3
cname_record_detail = response["ResourceRecordSets"][2]
cname_record_detail["Name"].should.equal("prod.redis.db.")
cname_record_detail["Type"].should.equal("TXT")
cname_record_detail["TTL"].should.equal(60)
cname_record_detail["ResourceRecords"].should.equal(
[{"Value": "SomeInitialValue&NewValue"}]
)
assert cname_record_detail["Name"] == "prod.redis.db."
assert cname_record_detail["Type"] == "TXT"
assert cname_record_detail["TTL"] == 60
assert cname_record_detail["ResourceRecords"] == [
{"Value": "SomeInitialValue&NewValue"}
]
# Delete record.
delete_payload = {
@ -827,7 +802,7 @@ def test_change_resource_record_sets_crud_valid_with_special_xml_chars():
HostedZoneId=hosted_zone_id, ChangeBatch=delete_payload
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(2)
assert len(response["ResourceRecordSets"]) == 2
@mock_route53
@ -875,9 +850,10 @@ def test_change_resource_record_set__delete_should_match_create():
},
)
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidInput")
err["Message"].should.equal(
"Invalid request: Expected exactly one of [AliasTarget, all of [TTL, and ResourceRecords], or TrafficPolicyInstanceId], but found none in Change with [Action=DELETE, Name=example.com, Type=A, SetIdentifier=null]"
assert err["Code"] == "InvalidInput"
assert (
err["Message"]
== "Invalid request: Expected exactly one of [AliasTarget, all of [TTL, and ResourceRecords], or TrafficPolicyInstanceId], but found none in Change with [Action=DELETE, Name=example.com, Type=A, SetIdentifier=null]"
)
@ -988,9 +964,9 @@ def test_change_weighted_resource_record_sets():
for record in response["ResourceRecordSets"]:
if record.get("SetIdentifier"):
if record["SetIdentifier"] == "test1":
record["Weight"].should.equal(90)
assert record["Weight"] == 90
if record["SetIdentifier"] == "test2":
record["Weight"].should.equal(10)
assert record["Weight"] == 10
@mock_route53
@ -1021,7 +997,7 @@ def test_failover_record_sets():
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
record = response["ResourceRecordSets"][2]
record["Failover"].should.equal("PRIMARY")
assert record["Failover"] == "PRIMARY"
@mock_route53
@ -1062,8 +1038,8 @@ def test_geolocation_record_sets():
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
rrs = response["ResourceRecordSets"]
rrs[2]["GeoLocation"].should.equal({"ContinentCode": "EU"})
rrs[3]["GeoLocation"].should.equal({"CountryCode": "US", "SubdivisionCode": "NY"})
assert rrs[2]["GeoLocation"] == {"ContinentCode": "EU"}
assert rrs[3]["GeoLocation"] == {"CountryCode": "US", "SubdivisionCode": "NY"}
@mock_route53
@ -1076,8 +1052,8 @@ def test_change_resource_record_invalid():
)
zones = conn.list_hosted_zones_by_name(DNSName="db.")
len(zones["HostedZones"]).should.equal(1)
zones["HostedZones"][0]["Name"].should.equal("db.")
assert len(zones["HostedZones"]) == 1
assert zones["HostedZones"][0]["Name"] == "db."
hosted_zone_id = zones["HostedZones"][0]["Id"]
invalid_a_record_payload = {
@ -1101,7 +1077,7 @@ def test_change_resource_record_invalid():
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(2)
assert len(response["ResourceRecordSets"]) == 2
invalid_cname_record_payload = {
"Comment": "this should also fail",
@ -1124,7 +1100,7 @@ def test_change_resource_record_invalid():
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(2)
assert len(response["ResourceRecordSets"]) == 2
@mock_route53
@ -1137,8 +1113,8 @@ def test_change_resource_record_invalid_action_value():
)
zones = conn.list_hosted_zones_by_name(DNSName="db.")
len(zones["HostedZones"]).should.equal(1)
zones["HostedZones"][0]["Name"].should.equal("db.")
assert len(zones["HostedZones"]) == 1
assert zones["HostedZones"][0]["Name"] == "db."
hosted_zone_id = zones["HostedZones"][0]["Id"]
invalid_a_record_payload = {
@ -1162,14 +1138,14 @@ def test_change_resource_record_invalid_action_value():
)
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidInput")
err["Message"].should.equal(
"Invalid XML ; cvc-enumeration-valid: Value 'INVALID_ACTION' is not facet-valid"
" with respect to enumeration '[CREATE, DELETE, UPSERT]'. It must be a value from the enumeration."
assert err["Code"] == "InvalidInput"
assert (
err["Message"]
== "Invalid XML ; cvc-enumeration-valid: Value 'INVALID_ACTION' is not facet-valid with respect to enumeration '[CREATE, DELETE, UPSERT]'. It must be a value from the enumeration."
)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response["ResourceRecordSets"]).should.equal(2)
assert len(response["ResourceRecordSets"]) == 2
@mock_route53
@ -1201,9 +1177,10 @@ def test_change_resource_record_set_create__should_fail_when_record_already_exis
client.change_resource_record_sets(HostedZoneId=zone_id, ChangeBatch=changes)
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidChangeBatch")
err["Message"].should.equal(
"Tried to create resource record set [name='test.cname.local.', type='CNAME'] but it already exists"
assert err["Code"] == "InvalidChangeBatch"
assert (
err["Message"]
== "Tried to create resource record set [name='test.cname.local.', type='CNAME'] but it already exists"
)
@ -1232,8 +1209,8 @@ def test_change_resource_record_set__should_create_record_when_using_upsert():
response = route53_client.list_resource_record_sets(HostedZoneId=hosted_zone["Id"])
# The 1st and 2nd records are NS and SOA records, respectively.
len(response["ResourceRecordSets"]).should.equal(3)
response["ResourceRecordSets"][2].should.equal(resource_record)
assert len(response["ResourceRecordSets"]) == 3
assert response["ResourceRecordSets"][2] == resource_record
# a subsequest UPSERT with the same ChangeBatch should succeed as well
route53_client.change_resource_record_sets(
@ -1245,8 +1222,8 @@ def test_change_resource_record_set__should_create_record_when_using_upsert():
response = route53_client.list_resource_record_sets(HostedZoneId=hosted_zone["Id"])
# The 1st and 2nd records are NS and SOA records, respectively.
len(response["ResourceRecordSets"]).should.equal(3)
response["ResourceRecordSets"][2].should.equal(resource_record)
assert len(response["ResourceRecordSets"]) == 3
assert response["ResourceRecordSets"][2] == resource_record
@mock_route53
@ -1297,14 +1274,14 @@ def test_list_resource_record_sets_name_type_filters():
StartRecordName=all_records[start_with][1],
)
response["IsTruncated"].should.equal(False)
assert response["IsTruncated"] is False
returned_records = [
(record["Type"], record["Name"]) for record in response["ResourceRecordSets"]
]
len(returned_records).should.equal(len(all_records) - start_with)
assert len(returned_records) == len(all_records) - start_with
for desired_record in all_records[start_with:]:
returned_records.should.contain(desired_record)
assert desired_record in returned_records
@mock_route53
@ -1314,8 +1291,8 @@ def test_get_change():
change_id = "123456"
response = conn.get_change(Id=change_id)
response["ChangeInfo"]["Id"].should.equal(change_id)
response["ChangeInfo"]["Status"].should.equal("INSYNC")
assert response["ChangeInfo"]["Id"] == change_id
assert response["ChangeInfo"]["Status"] == "INSYNC"
@mock_route53
@ -1328,8 +1305,8 @@ def test_change_resource_record_sets_records_limit():
)
zones = conn.list_hosted_zones_by_name(DNSName="db.")
len(zones["HostedZones"]).should.equal(1)
zones["HostedZones"][0]["Name"].should.equal("db.")
assert len(zones["HostedZones"]) == 1
assert zones["HostedZones"][0]["Name"] == "db."
hosted_zone_id = zones["HostedZones"][0]["Id"]
# Changes creating exactly 1,000 resource records.
@ -1382,7 +1359,7 @@ def test_change_resource_record_sets_records_limit():
ChangeBatch=create_1001_resource_records_payload,
)
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidChangeBatch")
assert err["Code"] == "InvalidChangeBatch"
# Changes upserting exactly 500 resource records.
changes = []
@ -1434,8 +1411,8 @@ def test_change_resource_record_sets_records_limit():
HostedZoneId=hosted_zone_id, ChangeBatch=upsert_501_resource_records_payload
)
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidChangeBatch")
err["Message"].should.equal("Number of records limit of 1000 exceeded.")
assert err["Code"] == "InvalidChangeBatch"
assert err["Message"] == "Number of records limit of 1000 exceeded."
@mock_route53
@ -1448,8 +1425,8 @@ def test_list_resource_recordset_pagination():
)
zones = conn.list_hosted_zones_by_name(DNSName="db.")
len(zones["HostedZones"]).should.equal(1)
zones["HostedZones"][0]["Name"].should.equal("db.")
assert len(zones["HostedZones"]) == 1
assert zones["HostedZones"][0]["Name"] == "db."
hosted_zone_id = zones["HostedZones"][0]["Id"]
# Create A Record.
@ -1475,33 +1452,33 @@ def test_list_resource_recordset_pagination():
response = conn.list_resource_record_sets(
HostedZoneId=hosted_zone_id, MaxItems="100"
)
response.should.have.key("ResourceRecordSets").length_of(100)
response.should.have.key("IsTruncated").equals(True)
response.should.have.key("MaxItems").equals("100")
response.should.have.key("NextRecordName").equals("env187.redis.db.")
response.should.have.key("NextRecordType").equals("A")
assert len(response["ResourceRecordSets"]) == 100
assert response["IsTruncated"]
assert response["MaxItems"] == "100"
assert response["NextRecordName"] == "env187.redis.db."
assert response["NextRecordType"] == "A"
response = conn.list_resource_record_sets(
HostedZoneId=hosted_zone_id,
StartRecordName=response["NextRecordName"],
StartRecordType=response["NextRecordType"],
)
response.should.have.key("ResourceRecordSets").length_of(300)
response.should.have.key("IsTruncated").equals(True)
response.should.have.key("MaxItems").equals("300")
response.should.have.key("NextRecordName").equals("env457.redis.db.")
response.should.have.key("NextRecordType").equals("A")
assert len(response["ResourceRecordSets"]) == 300
assert response["IsTruncated"] is True
assert response["MaxItems"] == "300"
assert response["NextRecordName"] == "env457.redis.db."
assert response["NextRecordType"] == "A"
response = conn.list_resource_record_sets(
HostedZoneId=hosted_zone_id,
StartRecordName=response["NextRecordName"],
StartRecordType=response["NextRecordType"],
)
response.should.have.key("ResourceRecordSets").length_of(102)
response.should.have.key("IsTruncated").equals(False)
response.should.have.key("MaxItems").equals("300")
response.shouldnt.have.key("NextRecordName")
response.shouldnt.have.key("NextRecordType")
assert len(response["ResourceRecordSets"]) == 102
assert response["IsTruncated"] is False
assert response["MaxItems"] == "300"
assert "NextRecordName" not in response
assert "NextRecordType" not in response
@mock_route53
@ -1512,4 +1489,4 @@ def test_get_dns_sec():
Name="testdns.aws.com.", CallerReference=str(hash("foo"))
)["HostedZone"]["Id"]
dns_sec = client.get_dnssec(HostedZoneId=hosted_zone_id)
dns_sec.should.have.key("Status").equals({"ServeSignature": "NOT_SIGNING"})
assert dns_sec["Status"] == {"ServeSignature": "NOT_SIGNING"}

92
tests/test_route53/test_route53_cloudformation.py
View File

@ -1,6 +1,5 @@
import boto3
import json
import sure # noqa # pylint: disable=unused-import
from copy import deepcopy
from moto import mock_cloudformation, mock_ec2, mock_route53
@ -55,8 +54,8 @@ def test_create_stack_hosted_zone_by_id():
# then a hosted zone should exist
zone = conn.list_hosted_zones()["HostedZones"][0]
zone.should.have.key("Name").equal("foo.bar.baz")
zone.should.have.key("ResourceRecordSetCount").equal(2)
assert zone["Name"] == "foo.bar.baz"
assert zone["ResourceRecordSetCount"] == 2
# when adding a record set to this zone
cf_conn.create_stack(
@ -67,9 +66,9 @@ def test_create_stack_hosted_zone_by_id():
# then the hosted zone should have a record
updated_zone = conn.list_hosted_zones()["HostedZones"][0]
updated_zone.should.have.key("Id").equal(zone["Id"])
updated_zone.should.have.key("Name").equal("foo.bar.baz")
updated_zone.should.have.key("ResourceRecordSetCount").equal(3)
assert updated_zone["Id"] == zone["Id"]
assert updated_zone["Name"] == "foo.bar.baz"
assert updated_zone["ResourceRecordSetCount"] == 3
@mock_cloudformation
@ -82,33 +81,33 @@ def test_route53_roundrobin():
cf.create_stack(StackName="test_stack", TemplateBody=template_json)
zones = route53.list_hosted_zones()["HostedZones"]
zones.should.have.length_of(1)
assert len(zones) == 1
zone_id = zones[0]["Id"].split("/")[2]
rrsets = route53.list_resource_record_sets(HostedZoneId=zone_id)[
"ResourceRecordSets"
]
rrsets.should.have.length_of(4)
assert len(rrsets) == 4
record_set1 = rrsets[2]
record_set1["Name"].should.equal("test_stack.us-west-1.my_zone.")
record_set1["SetIdentifier"].should.equal("test_stack AWS")
record_set1["Type"].should.equal("CNAME")
record_set1["TTL"].should.equal(900)
record_set1["Weight"].should.equal(3)
record_set1["ResourceRecords"][0]["Value"].should.equal("aws.amazon.com")
assert record_set1["Name"] == "test_stack.us-west-1.my_zone."
assert record_set1["SetIdentifier"] == "test_stack AWS"
assert record_set1["Type"] == "CNAME"
assert record_set1["TTL"] == 900
assert record_set1["Weight"] == 3
assert record_set1["ResourceRecords"][0]["Value"] == "aws.amazon.com"
record_set2 = rrsets[3]
record_set2["Name"].should.equal("test_stack.us-west-1.my_zone.")
record_set2["SetIdentifier"].should.equal("test_stack Amazon")
record_set2["Type"].should.equal("CNAME")
record_set2["TTL"].should.equal(900)
record_set2["Weight"].should.equal(1)
record_set2["ResourceRecords"][0]["Value"].should.equal("www.amazon.com")
assert record_set2["Name"] == "test_stack.us-west-1.my_zone."
assert record_set2["SetIdentifier"] == "test_stack Amazon"
assert record_set2["Type"] == "CNAME"
assert record_set2["TTL"] == 900
assert record_set2["Weight"] == 1
assert record_set2["ResourceRecords"][0]["Value"] == "www.amazon.com"
stack = cf.describe_stacks(StackName="test_stack")["Stacks"][0]
output = stack["Outputs"][0]
output["OutputKey"].should.equal("DomainName")
output["OutputValue"].should.equal(f"arn:aws:route53:::hostedzone/{zone_id}")
assert output["OutputKey"] == "DomainName"
assert output["OutputValue"] == f"arn:aws:route53:::hostedzone/{zone_id}"
@mock_cloudformation
@ -127,21 +126,21 @@ def test_route53_ec2_instance_with_public_ip():
]
zones = route53.list_hosted_zones()["HostedZones"]
zones.should.have.length_of(1)
assert len(zones) == 1
zone_id = zones[0]["Id"].split("/")[2]
rrsets = route53.list_resource_record_sets(HostedZoneId=zone_id)[
"ResourceRecordSets"
]
rrsets.should.have.length_of(3)
assert len(rrsets) == 3
record_set = rrsets[2]
record_set["Name"].should.equal(f"{instance_id}.us-west-1.my_zone.")
record_set.shouldnt.have.key("SetIdentifier")
record_set["Type"].should.equal("A")
record_set["TTL"].should.equal(900)
record_set.shouldnt.have.key("Weight")
record_set["ResourceRecords"][0]["Value"].should.equal("10.0.0.25")
assert record_set["Name"] == f"{instance_id}.us-west-1.my_zone."
assert record_set["Type"] == "A"
assert record_set["TTL"] == 900
assert record_set["ResourceRecords"][0]["Value"] == "10.0.0.25"
assert "SetIdentifier" not in record_set
assert "Weight" not in record_set
@mock_cloudformation
@ -154,27 +153,27 @@ def test_route53_associate_health_check():
cf.create_stack(StackName="test_stack", TemplateBody=template_json)
checks = route53.list_health_checks()["HealthChecks"]
checks.should.have.length_of(1)
assert len(checks) == 1
check = checks[0]
health_check_id = check["Id"]
config = check["HealthCheckConfig"]
config["FailureThreshold"].should.equal(3)
config["IPAddress"].should.equal("10.0.0.4")
config["Port"].should.equal(80)
config["RequestInterval"].should.equal(10)
config["ResourcePath"].should.equal("/")
config["Type"].should.equal("HTTP")
assert config["FailureThreshold"] == 3
assert config["IPAddress"] == "10.0.0.4"
assert config["Port"] == 80
assert config["RequestInterval"] == 10
assert config["ResourcePath"] == "/"
assert config["Type"] == "HTTP"
zones = route53.list_hosted_zones()["HostedZones"]
zones.should.have.length_of(1)
assert len(zones) == 1
zone_id = zones[0]["Id"].split("/")[2]
rrsets = route53.list_resource_record_sets(HostedZoneId=zone_id)[
"ResourceRecordSets"
]
rrsets.should.have.length_of(3)
assert len(rrsets) == 3
record_set = rrsets[0]
record_set["HealthCheckId"].should.equal(health_check_id)
assert record_set["HealthCheckId"] == health_check_id
@mock_cloudformation
@ -187,7 +186,7 @@ def test_route53_with_update():
cf.create_stack(StackName="test_stack", TemplateBody=template_json)
zones = route53.list_hosted_zones()["HostedZones"]
zones.should.have.length_of(1)
assert len(zones) == 1
zone_id = zones[0]["Id"]
zone_id = zone_id.split("/")
zone_id = zone_id[2]
@ -195,10 +194,10 @@ def test_route53_with_update():
rrsets = route53.list_resource_record_sets(HostedZoneId=zone_id)[
"ResourceRecordSets"
]
rrsets.should.have.length_of(3)
assert len(rrsets) == 3
record_set = rrsets[0]
record_set["ResourceRecords"][0]["Value"].should.equal("my.example.com")
assert record_set["ResourceRecords"][0]["Value"] == "my.example.com"
# # given
template = deepcopy(route53_health_check.template)
@ -212,22 +211,21 @@ def test_route53_with_update():
# # then
zones = route53.list_hosted_zones()["HostedZones"]
zones.should.have.length_of(1)
assert len(zones) == 1
zone_id = zones[0]["Id"].split("/")[2]
rrsets = route53.list_resource_record_sets(HostedZoneId=zone_id)[
"ResourceRecordSets"
]
rrsets.should.have.length_of(3)
assert len(rrsets) == 3
record_set = rrsets[0]
record_set["ResourceRecords"][0]["Value"].should.equal("my_other.example.com")
assert record_set["ResourceRecords"][0]["Value"] == "my_other.example.com"
@mock_cloudformation
@mock_route53
def test_delete_route53_recordset():
cf = boto3.client("cloudformation", region_name="us-west-1")
# given a stack with a record set

42
tests/test_route53/test_route53_delegationsets.py
View File

@ -10,8 +10,8 @@ def test_list_reusable_delegation_set():
client = boto3.client("route53", region_name="us-east-1")
resp = client.list_reusable_delegation_sets()
resp.should.have.key("DelegationSets").equals([])
resp.should.have.key("IsTruncated").equals(False)
assert resp["DelegationSets"] == []
assert resp["IsTruncated"] is False
@mock_route53
@ -20,13 +20,11 @@ def test_create_reusable_delegation_set():
resp = client.create_reusable_delegation_set(CallerReference="r3f3r3nc3")
headers = resp["ResponseMetadata"]["HTTPHeaders"]
headers.should.have.key("location")
assert "location" in headers
resp.should.have.key("DelegationSet")
resp["DelegationSet"].should.have.key("Id")
resp["DelegationSet"].should.have.key("CallerReference").equals("r3f3r3nc3")
resp["DelegationSet"].should.have.key("NameServers").length_of(4)
assert "Id" in resp["DelegationSet"]
assert resp["DelegationSet"]["CallerReference"] == "r3f3r3nc3"
assert len(resp["DelegationSet"]["NameServers"]) == 4
@mock_route53
@ -42,7 +40,7 @@ def test_create_reusable_delegation_set_from_hosted_zone():
CallerReference="r3f3r3nc3", HostedZoneId=hosted_zone_id
)
set(resp["DelegationSet"]["NameServers"]).should.equal(hosted_zone_name_servers)
assert set(resp["DelegationSet"]["NameServers"]) == hosted_zone_name_servers
@mock_route53
@ -54,9 +52,7 @@ def test_create_reusable_delegation_set_from_hosted_zone_with_delegationsetid():
DelegationSetId="customdelegationsetid",
)
response.should.have.key("DelegationSet")
response["DelegationSet"].should.have.key("Id").equals("customdelegationsetid")
response["DelegationSet"].should.have.key("NameServers")
assert response["DelegationSet"]["Id"] == "customdelegationsetid"
hosted_zone_id = response["HostedZone"]["Id"]
hosted_zone_name_servers = set(response["DelegationSet"]["NameServers"])
@ -65,8 +61,8 @@ def test_create_reusable_delegation_set_from_hosted_zone_with_delegationsetid():
CallerReference="r3f3r3nc3", HostedZoneId=hosted_zone_id
)
resp["DelegationSet"].should.have.key("Id").shouldnt.equal("customdelegationsetid")
set(resp["DelegationSet"]["NameServers"]).should.equal(hosted_zone_name_servers)
assert resp["DelegationSet"]["Id"] != "customdelegationsetid"
assert set(resp["DelegationSet"]["NameServers"]) == hosted_zone_name_servers
@mock_route53
@ -78,11 +74,11 @@ def test_get_reusable_delegation_set():
resp = client.get_reusable_delegation_set(Id=ds_id)
resp.should.have.key("DelegationSet")
assert "DelegationSet" in resp
resp["DelegationSet"].should.have.key("Id").equals(ds_id)
resp["DelegationSet"].should.have.key("CallerReference").equals("r3f3r3nc3")
resp["DelegationSet"].should.have.key("NameServers").length_of(4)
assert resp["DelegationSet"]["Id"] == ds_id
assert resp["DelegationSet"]["CallerReference"] == "r3f3r3nc3"
assert len(resp["DelegationSet"]["NameServers"]) == 4
@mock_route53
@ -92,8 +88,8 @@ def test_get_reusable_delegation_set_unknown():
with pytest.raises(ClientError) as exc:
client.get_reusable_delegation_set(Id="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NoSuchDelegationSet")
err["Message"].should.equal("unknown")
assert err["Code"] == "NoSuchDelegationSet"
assert err["Message"] == "unknown"
@mock_route53
@ -103,8 +99,8 @@ def test_list_reusable_delegation_sets():
client.create_reusable_delegation_set(CallerReference="r3f3r3nc4")
resp = client.list_reusable_delegation_sets()
resp.should.have.key("DelegationSets").length_of(2)
resp.should.have.key("IsTruncated").equals(False)
assert len(resp["DelegationSets"]) == 2
assert resp["IsTruncated"] is False
@mock_route53
@ -116,4 +112,4 @@ def test_delete_reusable_delegation_set():
client.delete_reusable_delegation_set(Id=ds_id)
client.list_reusable_delegation_sets()["DelegationSets"].should.have.length_of(0)
assert len(client.list_reusable_delegation_sets()["DelegationSets"]) == 0

159
tests/test_route53/test_route53_healthchecks.py
View File

@ -1,6 +1,5 @@
import boto3
import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError
from moto import mock_route53
@ -21,30 +20,28 @@ def test_create_health_check():
"FailureThreshold": 2,
},
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(201)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 201
#
check = response["HealthCheck"]
check.should.have.key("Id")
check.should.have.key("CallerReference").being.equal(
"test-route53-health-HealthCheck-asdf"
)
check.should.have.key("HealthCheckConfig")
assert "Id" in check
assert check["CallerReference"] == "test-route53-health-HealthCheck-asdf"
assert "HealthCheckConfig" in check
#
config = check["HealthCheckConfig"]
config.should.have.key("IPAddress").being.equal("93.184.216.34")
config.should.have.key("Port").being.equal(80)
config.should.have.key("Type").being.equal("HTTP")
config.should.have.key("ResourcePath").being.equal("/")
config.should.have.key("FullyQualifiedDomainName").being.equal("example.com")
config.should.have.key("RequestInterval").being.equal(10)
config.should.have.key("FailureThreshold").being.equal(2)
config.should.have.key("MeasureLatency").being.equal(False)
config.should.have.key("Inverted").being.equal(False)
config.should.have.key("Disabled").being.equal(False)
config.should.have.key("EnableSNI").being.equal(False)
assert config["IPAddress"] == "93.184.216.34"
assert config["Port"] == 80
assert config["Type"] == "HTTP"
assert config["ResourcePath"] == "/"
assert config["FullyQualifiedDomainName"] == "example.com"
assert config["RequestInterval"] == 10
assert config["FailureThreshold"] == 2
assert config["MeasureLatency"] is False
assert config["Inverted"] is False
assert config["Disabled"] is False
assert config["EnableSNI"] is False
config.shouldnt.have.key("ChildHealthChecks")
config.shouldnt.have.key("HealthThreshold")
assert "ChildHealthChecks" not in config
assert "HealthThreshold" not in config
@mock_route53
@ -68,21 +65,19 @@ def test_create_health_check_with_additional_options():
"EnableSNI": True,
},
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(201)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 201
#
check = response["HealthCheck"]
check.should.have.key("CallerReference").being.equal(
"test-route53-health-HealthCheck-asdf"
)
check.should.have.key("HealthCheckVersion").equal(1)
check.should.have.key("HealthCheckConfig")
assert check["CallerReference"] == "test-route53-health-HealthCheck-asdf"
assert check["HealthCheckVersion"] == 1
assert "HealthCheckConfig" in check
#
config = check["HealthCheckConfig"]
check["HealthCheckConfig"].should.have.key("SearchString").equal("a good response")
config.should.have.key("MeasureLatency").being.equal(True)
config.should.have.key("Inverted").being.equal(True)
config.should.have.key("Disabled").being.equal(True)
config.should.have.key("EnableSNI").being.equal(True)
assert check["HealthCheckConfig"]["SearchString"] == "a good response"
assert config["MeasureLatency"] is True
assert config["Inverted"] is True
assert config["Disabled"] is True
assert config["EnableSNI"] is True
@mock_route53
@ -100,24 +95,22 @@ def test_create_calculated_health_check():
)
check = response["HealthCheck"]
check.should.have.key("Id")
check.should.have.key("CallerReference").being.equal(
"test-route53-health-HealthCheck-ZHV123"
)
assert "Id" in check
assert check["CallerReference"] == "test-route53-health-HealthCheck-ZHV123"
#
config = check["HealthCheckConfig"]
config.should.have.key("Type").being.equal("CALCULATED")
config.should.have.key("Inverted").being.equal(False)
config.should.have.key("Disabled").being.equal(False)
config.should.have.key("HealthThreshold").being.equal(1)
assert config["Type"] == "CALCULATED"
assert config["Inverted"] is False
assert config["Disabled"] is False
assert config["HealthThreshold"] == 1
#
config.shouldnt.have.key("IPAddress")
config.shouldnt.have.key("Port")
config.shouldnt.have.key("ResourcePath")
config.shouldnt.have.key("FullyQualifiedDomainName")
config.shouldnt.have.key("RequestInterval")
config.shouldnt.have.key("FailureThreshold")
config.shouldnt.have.key("MeasureLatency")
assert "IPAddress" not in config
assert "Port" not in config
assert "ResourcePath" not in config
assert "FullyQualifiedDomainName" not in config
assert "RequestInterval" not in config
assert "FailureThreshold" not in config
assert "MeasureLatency" not in config
@mock_route53
@ -159,19 +152,18 @@ def test_create_calculated_health_check_with_children():
)
check = parent["HealthCheck"]
check.should.have.key("Id")
check.should.have.key("CallerReference").being.equal(
"test-route53-health-HealthCheck-parent"
)
assert "Id" in check
assert check["CallerReference"] == "test-route53-health-HealthCheck-parent"
#
config = check["HealthCheckConfig"]
config.should.have.key("Type").being.equal("CALCULATED")
config.should.have.key("Inverted").being.equal(False)
config.should.have.key("Disabled").being.equal(False)
config.should.have.key("HealthThreshold").being.equal(1)
config.should.have.key("ChildHealthChecks").being.equal(
[child1["HealthCheck"]["Id"], child2["HealthCheck"]["Id"]]
)
assert config["Type"] == "CALCULATED"
assert config["Inverted"] is False
assert config["Disabled"] is False
assert config["HealthThreshold"] == 1
assert config["ChildHealthChecks"] == [
child1["HealthCheck"]["Id"],
child2["HealthCheck"]["Id"],
]
@mock_route53
@ -189,9 +181,9 @@ def test_get_health_check():
)["HealthCheck"]["Id"]
resp = client.get_health_check(HealthCheckId=hc_id)["HealthCheck"]
resp.should.have.key("Id").equals(hc_id)
resp.should.have.key("CallerReference").equals("callref")
resp.should.have.key("HealthCheckVersion").equals(1)
assert resp["Id"] == hc_id
assert resp["CallerReference"] == "callref"
assert resp["HealthCheckVersion"] == 1
@mock_route53
@ -201,15 +193,15 @@ def test_get_unknown_health_check():
with pytest.raises(ClientError) as exc:
client.get_health_check(HealthCheckId="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NoSuchHealthCheck")
err["Message"].should.equal("A health check with id unknown does not exist.")
assert err["Code"] == "NoSuchHealthCheck"
assert err["Message"] == "A health check with id unknown does not exist."
@mock_route53
def test_list_health_checks():
conn = boto3.client("route53", region_name="us-east-1")
conn.list_health_checks()["HealthChecks"].should.have.length_of(0)
assert len(conn.list_health_checks()["HealthChecks"]) == 0
check = conn.create_health_check(
CallerReference="?",
@ -225,16 +217,14 @@ def test_list_health_checks():
},
)["HealthCheck"]
checks = conn.list_health_checks()["HealthChecks"]
checks.should.have.length_of(1)
checks.should.contain(check)
assert conn.list_health_checks()["HealthChecks"] == [check]
@mock_route53
def test_delete_health_checks():
conn = boto3.client("route53", region_name="us-east-1")
conn.list_health_checks()["HealthChecks"].should.have.length_of(0)
assert len(conn.list_health_checks()["HealthChecks"]) == 0
check = conn.create_health_check(
CallerReference="?",
@ -253,7 +243,7 @@ def test_delete_health_checks():
conn.delete_health_check(HealthCheckId=check["Id"])
checks = conn.list_health_checks()["HealthChecks"]
checks.should.have.length_of(0)
assert len(checks) == 0
@mock_route53
@ -288,14 +278,14 @@ def test_update_health_check():
config = client.get_health_check(HealthCheckId=hc_id)["HealthCheck"][
"HealthCheckConfig"
]
config.should.have.key("Type").equals("CALCULATED")
config.should.have.key("ResourcePath").equals("rp")
config.should.have.key("FullyQualifiedDomainName").equals("example.com")
config.should.have.key("SearchString").equals("search")
config.should.have.key("Inverted").equals(False)
config.should.have.key("Disabled").equals(False)
config.should.have.key("ChildHealthChecks").equals(["child"])
config.should.have.key("Regions").equals(["us-east-1", "us-east-2", "us-west-1"])
assert config["Type"] == "CALCULATED"
assert config["ResourcePath"] == "rp"
assert config["FullyQualifiedDomainName"] == "example.com"
assert config["SearchString"] == "search"
assert config["Inverted"] is False
assert config["Disabled"] is False
assert config["ChildHealthChecks"] == ["child"]
assert config["Regions"] == ["us-east-1", "us-east-2", "us-west-1"]
@mock_route53
@ -313,19 +303,20 @@ def test_health_check_status():
)["HealthCheck"]["Id"]
resp = client.get_health_check_status(HealthCheckId=hc_id)
resp["HealthCheckObservations"].should.have.length_of(1)
assert len(resp["HealthCheckObservations"]) == 1
observation = resp["HealthCheckObservations"][0]
observation.should.have.key("Region").being.equals("us-east-1")
observation.should.have.key("IPAddress").being.equals("127.0.13.37")
observation.should.have.key("StatusReport")
observation["StatusReport"].should.have.key("Status").being.equals(
"Success: HTTP Status Code: 200. Resolved IP: 127.0.13.37. OK"
assert observation["Region"] == "us-east-1"
assert observation["IPAddress"] == "127.0.13.37"
assert "StatusReport" in observation
assert (
observation["StatusReport"]["Status"]
== "Success: HTTP Status Code: 200. Resolved IP: 127.0.13.37. OK"
)
with pytest.raises(ClientError) as exc:
client.get_health_check_status(HealthCheckId="bad-id")
err = exc.value.response["Error"]
err["Code"].should.equal("NoSuchHealthCheck")
err["Message"].should.equal("A health check with id bad-id does not exist.")
assert err["Code"] == "NoSuchHealthCheck"
assert err["Message"] == "A health check with id bad-id does not exist."

85
tests/test_route53/test_route53_vpcs.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
import pytest
from botocore.exceptions import ClientError
@ -25,20 +24,18 @@ def test_hosted_zone_private_zone_preserved():
zone_id = new_zone["HostedZone"]["Id"].split("/")[-1]
hosted_zone = conn.get_hosted_zone(Id=zone_id)
hosted_zone["HostedZone"]["Config"]["PrivateZone"].should.equal(True)
hosted_zone.should.have.key("VPCs")
hosted_zone["VPCs"].should.have.length_of(1)
hosted_zone["VPCs"][0].should.have.key("VPCId")
hosted_zone["VPCs"][0].should.have.key("VPCRegion")
hosted_zone["VPCs"][0]["VPCId"].should.be.equal(vpc_id)
hosted_zone["VPCs"][0]["VPCRegion"].should.be.equal(region)
assert hosted_zone["HostedZone"]["Config"]["PrivateZone"]
assert len(hosted_zone["VPCs"]) == 1
assert hosted_zone["VPCs"][0]["VPCId"] == vpc_id
assert hosted_zone["VPCs"][0]["VPCRegion"] == region
hosted_zones = conn.list_hosted_zones()
hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
assert hosted_zones["HostedZones"][0]["Config"]["PrivateZone"]
hosted_zones = conn.list_hosted_zones_by_name(DNSName="testdns.aws.com.")
hosted_zones["HostedZones"].should.have.length_of(1)
hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
assert len(hosted_zones["HostedZones"]) == 1
assert hosted_zones["HostedZones"][0]["Config"]["PrivateZone"]
# create_hosted_zone statements with PrivateZone=True,
# but without a _valid_ vpc-id should NOT fail.
@ -51,19 +48,19 @@ def test_hosted_zone_private_zone_preserved():
zone_id = no_vpc_zone["HostedZone"]["Id"].split("/")[-1]
hosted_zone = conn.get_hosted_zone(Id=zone_id)
hosted_zone["HostedZone"]["Config"]["PrivateZone"].should.equal(True)
hosted_zone.should.have.key("VPCs")
hosted_zone["VPCs"].should.have.length_of(0)
assert hosted_zone["HostedZone"]["Config"]["PrivateZone"]
assert len(hosted_zone["VPCs"]) == 0
hosted_zones = conn.list_hosted_zones()
hosted_zones["HostedZones"].should.have.length_of(2)
hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
hosted_zones["HostedZones"][1]["Config"]["PrivateZone"].should.equal(True)
assert len(hosted_zones["HostedZones"]) == 2
assert hosted_zones["HostedZones"][0]["Config"]["PrivateZone"]
assert hosted_zones["HostedZones"][1]["Config"]["PrivateZone"]
hosted_zones = conn.list_hosted_zones_by_name(DNSName=zone2_name)
hosted_zones["HostedZones"].should.have.length_of(1)
hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
hosted_zones["HostedZones"][0]["Name"].should.equal(zone2_name)
assert len(hosted_zones["HostedZones"]) == 1
assert hosted_zones["HostedZones"][0]["Config"]["PrivateZone"]
assert hosted_zones["HostedZones"][0]["Name"] == zone2_name
@mock_ec2
@ -88,19 +85,15 @@ def test_list_hosted_zones_by_vpc_with_multiple_vpcs():
# List the zones associated with this vpc
response = conn.list_hosted_zones_by_vpc(VPCId=vpc_id, VPCRegion=region)
response.should.have.key("ResponseMetadata")
response.should.have.key("HostedZoneSummaries")
response["HostedZoneSummaries"].should.have.length_of(3)
assert len(response["HostedZoneSummaries"]) == 3
# Loop through all zone summaries and verify they match what was created
for summary in response["HostedZoneSummaries"]:
# use the zone name as the index
index = summary["Name"].split(".")[1]
zone_id = zones[index]["HostedZone"]["Id"].split("/")[2]
summary.should.have.key("HostedZoneId")
summary["HostedZoneId"].should.equal(zone_id)
summary.should.have.key("Name")
summary["Name"].should.equal(zones[index]["HostedZone"]["Name"])
assert summary["HostedZoneId"] == zone_id
assert summary["Name"] == zones[index]["HostedZone"]["Name"]
@mock_ec2
@ -121,13 +114,10 @@ def test_list_hosted_zones_by_vpc():
zone_id = zone_b["HostedZone"]["Id"].split("/")[2]
response = conn.list_hosted_zones_by_vpc(VPCId=vpc_id, VPCRegion=region)
response.should.have.key("ResponseMetadata")
response.should.have.key("HostedZoneSummaries")
response["HostedZoneSummaries"].should.have.length_of(1)
response["HostedZoneSummaries"][0].should.have.key("HostedZoneId")
retured_zone = response["HostedZoneSummaries"][0]
retured_zone["HostedZoneId"].should.equal(zone_id)
retured_zone["Name"].should.equal(zone_b["HostedZone"]["Name"])
assert len(response["HostedZoneSummaries"]) == 1
returned_zone = response["HostedZoneSummaries"][0]
assert returned_zone["HostedZoneId"] == zone_id
assert returned_zone["Name"] == zone_b["HostedZone"]["Name"]
@mock_ec2
@ -148,8 +138,7 @@ def test_route53_associate_vpc():
VPC={"VPCId": vpc_id, "VPCRegion": "us-east-1"},
Comment="yolo",
)
resp.should.have.key("ChangeInfo")
resp["ChangeInfo"].should.have.key("Comment").equals("yolo")
assert resp["ChangeInfo"]["Comment"] == "yolo"
@mock_ec2
@ -171,9 +160,10 @@ def test_route53_associate_vpc_with_public_Zone():
Comment="yolo",
)
err = exc.value.response["Error"]
err["Code"].should.equal("PublicZoneVPCAssociation")
err["Message"].should.equal(
"You're trying to associate a VPC with a public hosted zone. Amazon Route 53 doesn't support associating a VPC with a public hosted zone."
assert err["Code"] == "PublicZoneVPCAssociation"
assert (
err["Message"]
== "You're trying to associate a VPC with a public hosted zone. Amazon Route 53 doesn't support associating a VPC with a public hosted zone."
)
@ -200,15 +190,15 @@ def test_route53_associate_and_disassociate_vpc():
)
zone_vpcs = conn.get_hosted_zone(Id=zone_id)["VPCs"]
zone_vpcs.should.have.length_of(2)
zone_vpcs.should.contain({"VPCRegion": region, "VPCId": vpc_id1})
zone_vpcs.should.contain({"VPCRegion": region, "VPCId": vpc_id2})
assert len(zone_vpcs) == 2
assert {"VPCRegion": region, "VPCId": vpc_id1} in zone_vpcs
assert {"VPCRegion": region, "VPCId": vpc_id2} in zone_vpcs
conn.disassociate_vpc_from_hosted_zone(HostedZoneId=zone_id, VPC={"VPCId": vpc_id1})
zone_vpcs = conn.get_hosted_zone(Id=zone_id)["VPCs"]
zone_vpcs.should.have.length_of(1)
zone_vpcs.should.contain({"VPCRegion": region, "VPCId": vpc_id2})
assert len(zone_vpcs) == 1
assert {"VPCRegion": region, "VPCId": vpc_id2} in zone_vpcs
@mock_ec2
@ -230,7 +220,8 @@ def test_route53_disassociate_last_vpc():
HostedZoneId=zone_id, VPC={"VPCId": vpc_id}
)
err = exc.value.response["Error"]
err["Code"].should.equal("LastVPCAssociation")
err["Message"].should.equal(
"The VPC that you're trying to disassociate from the private hosted zone is the last VPC that is associated with the hosted zone. Amazon Route 53 doesn't support disassociating the last VPC from a hosted zone."
assert err["Code"] == "LastVPCAssociation"
assert (
err["Message"]
== "The VPC that you're trying to disassociate from the private hosted zone is the last VPC that is associated with the hosted zone. Amazon Route 53 doesn't support disassociating the last VPC from a hosted zone."
)

3
tests/test_route53/test_server.py
View File

@ -1,4 +1,3 @@
import sure # noqa # pylint: disable=unused-import
import xmltodict
import moto.server as server
@ -22,7 +21,7 @@ def test_list_recordset():
# list record set
res = test_client.get(f"2013-04-01/hostedzone/{zone_id}/rrset")
# Ampersand should be properly encoded
res.data.decode("utf-8").should.contain("<Value><![CDATA[val&sth]]></Value>")
assert "<Value><![CDATA[val&sth]]></Value>" in res.data.decode("utf-8")
def parse_xml(body):

25
tests/test_route53resolver/test_route53resolver_endpoint.py
View File

@ -858,20 +858,20 @@ def test_associate_resolver_endpoint_ip_address():
# create resolver
random_num = mock_random.get_random_hex(10)
resolver = create_test_endpoint(client, ec2_client, name=f"A-{random_num}")
resolver.should.have.key("IpAddressCount").equals(2)
assert resolver["IpAddressCount"] == 2
# associate
resp = client.associate_resolver_endpoint_ip_address(
IpAddress={"Ip": "10.0.2.126", "SubnetId": subnet["SubnetId"]},
ResolverEndpointId=resolver["Id"],
)["ResolverEndpoint"]
resp.should.have.key("Id").equals(resolver["Id"])
resp.should.have.key("IpAddressCount").equals(3)
resp.should.have.key("SecurityGroupIds").should.have.length_of(1)
assert resp["Id"] == resolver["Id"]
assert resp["IpAddressCount"] == 3
assert len(resp["SecurityGroupIds"]) == 1
# verify ENI was created
enis = ec2_client.describe_network_interfaces()["NetworkInterfaces"]
ip_addresses = [eni["PrivateIpAddress"] for eni in enis]
ip_addresses.should.contain("10.0.2.126")
assert "10.0.2.126" in ip_addresses
@mock_route53resolver
@ -882,8 +882,8 @@ def test_associate_resolver_endpoint_ip_address__invalid_resolver():
IpAddress={"Ip": "notapplicable"}, ResolverEndpointId="unknown"
)
err = exc.value.response["Error"]
err["Code"].should.equal("ResourceNotFoundException")
err["Message"].should.equal("Resolver endpoint with ID 'unknown' does not exist")
assert err["Code"] == "ResourceNotFoundException"
assert err["Message"] == "Resolver endpoint with ID 'unknown' does not exist"
@mock_ec2
@ -912,7 +912,7 @@ def test_disassociate_resolver_endpoint_ip_address__using_ip():
)
# One ENI was deleted
enis_after = ec2_client.describe_network_interfaces()["NetworkInterfaces"]
(len(enis_after) + 1).should.equal(len(enis_before))
assert (len(enis_after) + 1) == len(enis_before)
@mock_ec2
@ -943,7 +943,7 @@ def test_disassociate_resolver_endpoint_ip_address__using_ipid_and_subnet():
ResolverEndpointId=resolver["Id"],
IpAddress={"SubnetId": subnet_id, "IpId": ip_id},
)["ResolverEndpoint"]
resp.should.have.key("IpAddressCount").equals(2)
assert resp["IpAddressCount"] == 2
@mock_ec2
@ -970,7 +970,8 @@ def test_disassociate_resolver_endpoint_ip_address__using_subnet_alone():
ResolverEndpointId=resolver["Id"], IpAddress={"SubnetId": subnet_id}
)
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidRequestException")
err["Message"].should.equal(
"[RSLVR-00503] Need to specify either the IP ID or both subnet and IP address in order to remove IP address."
assert err["Code"] == "InvalidRequestException"
assert (
err["Message"]
== "[RSLVR-00503] Need to specify either the IP ID or both subnet and IP address in order to remove IP address."
)

3
tests/test_route53resolver/test_route53resolver_rule.py
View File

@ -1,9 +1,8 @@
"""Unit tests for route53resolver rule-related APIs."""
from botocore.exceptions import ClientError
from datetime import datetime, timezone
import boto3
from botocore.exceptions import ClientError
import pytest
from moto import mock_route53resolver