Feature: Route53Domains (#7406)

This commit is contained in:
Neev Cohen 2024-03-05 01:16:02 +02:00 committed by GitHub
parent 1c11fbc0a2
commit d815421072
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 2255 additions and 3 deletions

View File

@ -142,6 +142,10 @@ backend_url_patterns = [
"route53resolver",
re.compile("https?://route53resolver\\.(.+)\\.amazonaws\\.com"),
),
(
"route53domains",
re.compile("https?://route53domains\\.(.+)\\.amazonaws\\.com"),
),
("s3", re.compile("https?://s3(?!-control)(.*)\\.amazonaws.com")),
(
"s3",

View File

@ -108,6 +108,7 @@ if TYPE_CHECKING:
from moto.resourcegroupstaggingapi.models import ResourceGroupsTaggingAPIBackend
from moto.robomaker.models import RoboMakerBackend
from moto.route53.models import Route53Backend
from moto.route53domains.models import Route53DomainsBackend
from moto.route53resolver.models import Route53ResolverBackend
from moto.s3.models import S3Backend
from moto.s3control.models import S3ControlBackend
@ -266,6 +267,7 @@ SERVICE_NAMES = Union[
"Literal['robomaker']",
"Literal['route53']",
"Literal['route53resolver']",
"Literal['route53domains']",
"Literal['s3']",
"Literal['s3bucket_path']",
"Literal['s3control']",
@ -506,6 +508,8 @@ def get_backend(name: "Literal['route53']") -> "BackendDict[Route53Backend]": ..
@overload
def get_backend(name: "Literal['route53resolver']") -> "BackendDict[Route53ResolverBackend]": ...
@overload
def get_backend(name: "Literal['route53domains']") -> "BackendDict[Route53DomainsBackend]": ...
@overload
def get_backend(name: "Literal['s3']") -> "BackendDict[S3Backend]": ...
@overload
def get_backend(name: "Literal['s3bucket_path']") -> "BackendDict[S3Backend]": ...

View File

@ -633,9 +633,7 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
if self.body is not None:
try:
return json.loads(self.body)[param_name]
except ValueError:
pass
except KeyError:
except (ValueError, KeyError):
pass
# try to get path parameter
if self.uri_match:

View File

@ -0,0 +1 @@
from .models import route53domains_backends # noqa: F401

View File

@ -0,0 +1,43 @@
from typing import List
from moto.core.exceptions import JsonRESTError
class DomainLimitExceededException(JsonRESTError):
code = 400
def __init__(self) -> None:
super().__init__(
"DomainLimitExceeded",
"The number of registered domains has exceeded the allowed threshold for this account. If you want to "
"register more domains please request a higher quota",
)
class DuplicateRequestException(JsonRESTError):
code = 400
def __init__(self) -> None:
super().__init__(
"DuplicateRequest", "The request is already in progress for the domain."
)
class InvalidInputException(JsonRESTError):
code = 400
def __init__(self, error_msgs: List[str]):
error_msgs_str = "\n\t".join(error_msgs)
super().__init__(
"InvalidInput", f"The requested item is not acceptable.\n\t{error_msgs_str}"
)
class UnsupportedTLDException(JsonRESTError):
code = 400
def __init__(self, tld: str):
super().__init__(
"UnsupportedTLD",
f"Amazon Route53 does not support the top-level domain (TLD) `.{tld}`.",
)

View File

@ -0,0 +1,302 @@
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
from moto.core.base_backend import BackendDict, BaseBackend
from moto.route53 import route53_backends
from moto.route53.models import Route53Backend
from moto.utilities.paginator import paginate
from .exceptions import (
DomainLimitExceededException,
DuplicateRequestException,
InvalidInputException,
)
from .validators import (
DOMAIN_OPERATION_STATUSES,
DOMAIN_OPERATION_TYPES,
DomainFilterField,
DomainsFilter,
DomainSortOrder,
DomainsSortCondition,
NameServer,
Route53Domain,
Route53DomainsContactDetail,
Route53DomainsOperation,
ValidationException,
)
class Route53DomainsBackend(BaseBackend):
"""Implementation of Route53Domains APIs."""
DEFAULT_MAX_DOMAINS_COUNT = 20
PAGINATION_MODEL = {
"list_domains": {
"input_token": "marker",
"limit_key": "max_items",
"limit_default": 20,
"unique_attribute": "domain_name",
},
"list_operations": {
"input_token": "marker",
"limit_key": "max_items",
"limit_default": 20,
"unique_attribute": "id",
},
}
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.__route53_backend: Route53Backend = route53_backends[account_id]["global"]
self.__domains: Dict[str, Route53Domain] = {}
self.__operations: Dict[str, Route53DomainsOperation] = {}
def register_domain(
self,
domain_name: str,
duration_in_years: int,
auto_renew: bool,
admin_contact: Dict[str, Any],
registrant_contact: Dict[str, Any],
tech_contact: Dict[str, Any],
private_protect_admin_contact: bool,
private_protect_registrant_contact: bool,
private_protect_tech_contact: bool,
extra_params: List[Dict[str, Any]],
) -> Route53DomainsOperation:
"""Register a domain"""
if len(self.__domains) == self.DEFAULT_MAX_DOMAINS_COUNT:
raise DomainLimitExceededException()
requested_operation = Route53DomainsOperation.validate(
domain_name=domain_name, status="SUCCESSFUL", type_="REGISTER_DOMAIN"
)
self.__validate_duplicate_operations(requested_operation)
expiration_date = datetime.now(timezone.utc) + timedelta(
days=365 * duration_in_years
)
try:
domain = Route53Domain.validate(
domain_name=domain_name,
auto_renew=auto_renew,
admin_contact=Route53DomainsContactDetail.validate_dict(admin_contact),
registrant_contact=Route53DomainsContactDetail.validate_dict(
registrant_contact
),
tech_contact=Route53DomainsContactDetail.validate_dict(tech_contact),
admin_privacy=private_protect_admin_contact,
registrant_privacy=private_protect_registrant_contact,
tech_privacy=private_protect_tech_contact,
expiration_date=expiration_date,
extra_params=extra_params,
)
except ValidationException as e:
raise InvalidInputException(e.errors)
self.__operations[requested_operation.id] = requested_operation
self.__route53_backend.create_hosted_zone(
name=domain.domain_name, private_zone=False
)
self.__domains[domain_name] = domain
return requested_operation
def delete_domain(self, domain_name: str) -> Route53DomainsOperation:
requested_operation = Route53DomainsOperation.validate(
domain_name=domain_name, status="SUCCESSFUL", type_="DELETE_DOMAIN"
)
self.__validate_duplicate_operations(requested_operation)
input_errors: List[str] = []
Route53Domain.validate_domain_name(domain_name, input_errors)
if input_errors:
raise InvalidInputException(input_errors)
if domain_name not in self.__domains:
raise InvalidInputException(
[f"Domain {domain_name} isn't registered in the current account"]
)
self.__operations[requested_operation.id] = requested_operation
del self.__domains[domain_name]
return requested_operation
def __validate_duplicate_operations(
self, requested_operation: Route53DomainsOperation
) -> None:
for operation in self.__operations.values():
if (
operation.domain_name == requested_operation.domain_name
and operation.type == requested_operation.type
):
raise DuplicateRequestException()
def get_domain(self, domain_name: str) -> Route53Domain:
input_errors: List[str] = []
Route53Domain.validate_domain_name(domain_name, input_errors)
if input_errors:
raise InvalidInputException(input_errors)
if domain_name not in self.__domains:
raise InvalidInputException(["Domain is not associated with this account"])
return self.__domains[domain_name]
@paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc]
def list_domains(
self,
filter_conditions: Optional[List[Dict[str, Any]]] = None,
sort_condition: Optional[Dict[str, Any]] = None,
) -> List[Route53Domain]:
try:
filters: List[DomainsFilter] = (
[DomainsFilter.validate_dict(f) for f in filter_conditions]
if filter_conditions
else []
)
sort: Optional[DomainsSortCondition] = (
DomainsSortCondition.validate_dict(sort_condition)
if sort_condition
else None
)
except ValidationException as e:
raise InvalidInputException(e.errors)
filter_fields = [f.name for f in filters]
if sort and filter_fields and sort.name not in filter_fields:
raise InvalidInputException(
["Sort condition must be the same as the filter condition"]
)
domains_to_return: List[Route53Domain] = []
for domain in self.__domains.values():
if all([f.filter(domain) for f in filters]):
domains_to_return.append(domain)
if sort:
if sort.name == DomainFilterField.DOMAIN_NAME:
domains_to_return.sort(
key=lambda d: d.domain_name,
reverse=(sort.sort_order == DomainSortOrder.DESCENDING),
)
else:
domains_to_return.sort(
key=lambda d: d.expiration_date,
reverse=(sort.sort_order == DomainSortOrder.DESCENDING),
)
return domains_to_return
@paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc]
def list_operations(
self,
submitted_since_timestamp: Optional[int] = None,
statuses: Optional[List[str]] = None,
types: Optional[List[str]] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = None,
) -> List[Route53DomainsOperation]:
input_errors: List[str] = []
statuses = statuses or []
types = types or []
if any(status not in DOMAIN_OPERATION_STATUSES for status in statuses):
input_errors.append("Status is invalid")
if any(type_ not in DOMAIN_OPERATION_TYPES for type_ in types):
input_errors.append("Type is invalid")
if input_errors:
raise InvalidInputException(input_errors)
submitted_since = (
datetime.fromtimestamp(submitted_since_timestamp, timezone.utc)
if submitted_since_timestamp
else None
)
operations_to_return: List[Route53DomainsOperation] = []
for operation in self.__operations.values():
if statuses and operation.status not in statuses:
continue
if types and operation.type not in types:
continue
if submitted_since and operation.submitted_date < submitted_since:
continue
operations_to_return.append(operation)
if sort_by == "SubmittedDate":
operations_to_return.sort(
key=lambda op: op.submitted_date,
reverse=sort_order == DomainSortOrder.ASCENDING,
)
return operations_to_return
def get_operation(self, operation_id: str) -> Route53DomainsOperation:
if operation_id not in self.__operations:
raise InvalidInputException(
[f"Operation with id {operation_id} doesn't exist"]
)
return self.__operations[operation_id]
def update_domain_nameservers(
self, domain_name: str, nameservers: List[Dict[str, Any]]
) -> Route53DomainsOperation:
input_errors: List[str] = []
Route53Domain.validate_domain_name(domain_name, input_errors)
if len(nameservers) < 1:
input_errors.append("Must supply nameservers")
servers: List[NameServer] = []
try:
servers = [NameServer.validate_dict(obj) for obj in nameservers]
except ValidationException as e:
input_errors += e.errors
for server in servers:
if domain_name in server.name and not server.glue_ips:
input_errors.append(
f"Must supply glue IPs for name server {server.name} because it is a subdomain of "
f"the domain"
)
if input_errors:
raise InvalidInputException(input_errors)
if domain_name not in self.__domains:
raise InvalidInputException(
[f"Domain {domain_name} is not registered to the current AWS account"]
)
requested_operation = Route53DomainsOperation.validate(
domain_name=domain_name, status="SUCCESSFUL", type_="UPDATE_NAMESERVER"
)
self.__validate_duplicate_operations(requested_operation)
domain = self.__domains[domain_name]
domain.nameservers = servers
self.__operations[requested_operation.id] = requested_operation
return requested_operation
route53domains_backends = BackendDict(
Route53DomainsBackend,
"route53domains",
use_boto3_regions=False,
additional_regions=["global"],
)

View File

@ -0,0 +1,138 @@
import json
from typing import Any, Dict
from moto.core.responses import BaseResponse
from moto.route53domains.models import Route53DomainsBackend, route53domains_backends
from moto.route53domains.validators import Route53Domain, Route53DomainsOperation
class Route53DomainsResponse(BaseResponse):
def __init__(self) -> None:
super().__init__(service_name="route53-domains")
@property
def route53domains_backend(self) -> Route53DomainsBackend:
return route53domains_backends[self.current_account]["global"]
def register_domain(self) -> str:
domain_name = self._get_param("DomainName")
duration_in_years = self._get_int_param("DurationInYears")
auto_renew = self._get_bool_param("AutoRenew", if_none=True)
admin_contact = self._get_param("AdminContact")
registrant_contact = self._get_param("RegistrantContact")
tech_contact = self._get_param("TechContact")
privacy_protection_admin_contact = self._get_bool_param(
"PrivacyProtectAdminContact", if_none=True
)
privacy_protection_registrant_contact = self._get_bool_param(
"PrivacyProtectRegistrantContact", if_none=True
)
privacy_protection_tech_contact = self._get_bool_param(
"PrivacyProtectTechContact", if_none=True
)
extra_params = self._get_param("ExtraParams")
operation = self.route53domains_backend.register_domain(
domain_name=domain_name,
duration_in_years=duration_in_years,
auto_renew=auto_renew,
admin_contact=admin_contact,
registrant_contact=registrant_contact,
tech_contact=tech_contact,
private_protect_admin_contact=privacy_protection_admin_contact,
private_protect_registrant_contact=privacy_protection_registrant_contact,
private_protect_tech_contact=privacy_protection_tech_contact,
extra_params=extra_params,
)
return json.dumps({"OperationId": operation.id})
def delete_domain(self) -> str:
domain_name = self._get_param("DomainName")
operation = self.route53domains_backend.delete_domain(domain_name)
return json.dumps({"OperationId": operation.id})
def get_domain_detail(self) -> str:
domain_name = self._get_param("DomainName")
return json.dumps(
self.route53domains_backend.get_domain(domain_name=domain_name).to_json()
)
def list_domains(self) -> str:
filter_conditions = self._get_param("FilterConditions")
sort_condition = self._get_param("SortCondition")
marker = self._get_param("Marker")
max_items = self._get_param("MaxItems")
domains, marker = self.route53domains_backend.list_domains(
filter_conditions=filter_conditions,
sort_condition=sort_condition,
marker=marker,
max_items=max_items,
)
res = {
"Domains": list(map(self.__map_domains_to_info, domains)),
}
if marker:
res["NextPageMarker"] = marker
return json.dumps(res)
@staticmethod
def __map_domains_to_info(domain: Route53Domain) -> Dict[str, Any]: # type: ignore[misc]
return {
"DomainName": domain.domain_name,
"AutoRenew": domain.auto_renew,
"Expiry": domain.expiration_date.timestamp(),
"TransferLock": True,
}
def list_operations(self) -> str:
submitted_since_timestamp = self._get_int_param("SubmittedSince")
max_items = self._get_int_param("MaxItems")
statuses = self._get_param("Status")
marker = self._get_param("Marker")
types = self._get_param("Type")
sort_by = self._get_param("SortBy")
sort_order = self._get_param("SortOrder")
operations, marker = self.route53domains_backend.list_operations(
submitted_since_timestamp=submitted_since_timestamp,
max_items=max_items,
marker=marker,
statuses=statuses,
types=types,
sort_by=sort_by,
sort_order=sort_order,
)
res = {
"Operations": [operation.to_json() for operation in operations],
}
if marker:
res["NextPageMarker"] = marker
return json.dumps(res)
def get_operation_detail(self) -> str:
operation_id = self._get_param("OperationId")
operation: Route53DomainsOperation = self.route53domains_backend.get_operation(
operation_id
)
return json.dumps(operation.to_json())
def update_domain_nameservers(self) -> str:
domain_name = self._get_param("DomainName")
nameservers = self._get_param("Nameservers")
operation: Route53DomainsOperation = (
self.route53domains_backend.update_domain_nameservers(
domain_name=domain_name, nameservers=nameservers
)
)
return json.dumps({"OperationId": operation.id})

View File

@ -0,0 +1,11 @@
"""route53domains base URL and path."""
from .responses import Route53DomainsResponse
url_bases = [
r"https?://route53domains\.(.+)\.amazonaws\.com",
]
url_paths = {
"{0}/$": Route53DomainsResponse.dispatch,
}

File diff suppressed because it is too large Load Diff

View File

View File

@ -0,0 +1,562 @@
from datetime import datetime, timedelta, timezone
from typing import Dict, List
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_aws
@pytest.fixture(name="domain_parameters")
def generate_domain_parameters() -> Dict:
return {
"DomainName": "domain.com",
"DurationInYears": 3,
"AutoRenew": True,
"AdminContact": {
"FirstName": "First",
"LastName": "Last",
"ContactType": "PERSON",
"AddressLine1": "address 1",
"AddressLine2": "address 2",
"City": "New York City",
"CountryCode": "US",
"ZipCode": "123123123",
"Email": "email@gmail.com",
"Fax": "+1.1234567890",
},
"RegistrantContact": {
"FirstName": "First",
"LastName": "Last",
"ContactType": "PERSON",
"AddressLine1": "address 1",
"AddressLine2": "address 2",
"City": "New York City",
"CountryCode": "US",
"ZipCode": "123123123",
"Email": "email@gmail.com",
"Fax": "+1.1234567890",
},
"TechContact": {
"FirstName": "First",
"LastName": "Last",
"ContactType": "PERSON",
"AddressLine1": "address 1",
"AddressLine2": "address 2",
"City": "New York City",
"CountryCode": "US",
"ZipCode": "123123123",
"Email": "email@gmail.com",
"Fax": "+1.1234567890",
},
"PrivacyProtectAdminContact": True,
"PrivacyProtectRegistrantContact": True,
"PrivacyProtectTechContact": True,
}
@pytest.fixture(name="invalid_domain_parameters")
def generate_invalid_domain_parameters(domain_parameters: Dict) -> Dict:
domain_parameters["DomainName"] = "a"
domain_parameters["DurationInYears"] = 500
return domain_parameters
@mock_aws
def test_register_domain(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
res = route53domains_client.register_domain(**domain_parameters)
operation_id = res["OperationId"]
operations = route53domains_client.list_operations(Type=["REGISTER_DOMAIN"])[
"Operations"
]
for operation in operations:
if operation["OperationId"] == operation_id:
return
assert operation_id in [
operation["OperationId"] for operation in operations
], "Could not find expected operation id returned from `register_domain` in operation list"
@mock_aws
def test_register_domain_creates_hosted_zone(
domain_parameters: Dict,
):
"""Test good register domain API calls."""
route53domains_client = boto3.client("route53domains", region_name="global")
route53_client = boto3.client("route53", region_name="global")
route53domains_client.register_domain(**domain_parameters)
res = route53_client.list_hosted_zones()
assert "domain.com" in [
zone["Name"] for zone in res["HostedZones"]
], "`register_domain` did not create a new hosted zone with the same name"
@mock_aws
def test_register_domain_fails_on_invalid_input(
invalid_domain_parameters: Dict,
):
route53domains_client = boto3.client("route53domains", region_name="global")
route53_client = boto3.client("route53", region_name="global")
with pytest.raises(ClientError) as exc:
route53domains_client.register_domain(**invalid_domain_parameters)
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
res = route53_client.list_hosted_zones()
assert len(res["HostedZones"]) == 0
@mock_aws
def test_register_domain_fails_on_invalid_tld(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53_client = boto3.client("route53", region_name="global")
params = domain_parameters.copy()
params["DomainName"] = "test.non-existing-tld"
with pytest.raises(ClientError) as exc:
route53domains_client.register_domain(**params)
err = exc.value.response["Error"]
assert err["Code"] == "UnsupportedTLD"
res = route53_client.list_hosted_zones()
assert len(res["HostedZones"]) == 0
@mock_aws
def test_list_operations(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
operations = route53domains_client.list_operations()["Operations"]
assert len(operations) == 1
future_time = datetime.now(timezone.utc) + timedelta(minutes=1)
operations = route53domains_client.list_operations(
SubmittedSince=future_time.timestamp()
)["Operations"]
assert len(operations) == 0
operations = route53domains_client.list_operations(Status=["SUCCESSFUL"])[
"Operations"
]
assert len(operations) == 1
operations = route53domains_client.list_operations(Status=["IN_PROGRESS"])[
"Operations"
]
assert len(operations) == 0
operations = route53domains_client.list_operations(Type=["REGISTER_DOMAIN"])[
"Operations"
]
assert len(operations) == 1
operations = route53domains_client.list_operations(Type=["DELETE_DOMAIN"])[
"Operations"
]
assert len(operations) == 0
@mock_aws
def test_list_operations_invalid_input():
route53domains_client = boto3.client("route53domains", region_name="global")
with pytest.raises(ClientError) as exc:
_ = route53domains_client.list_operations(Type=["INVALID_TYPE"])["Operations"]
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
with pytest.raises(ClientError) as exc:
_ = route53domains_client.list_operations(Status=["INVALID_STATUS"])[
"Operations"
]
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
@mock_aws
def test_get_operation_detail(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
res = route53domains_client.register_domain(**domain_parameters)
expected_operation_id = res["OperationId"]
operation = route53domains_client.get_operation_detail(
OperationId=expected_operation_id
)
assert operation["OperationId"] == expected_operation_id
assert operation["Status"] == "SUCCESSFUL"
assert operation["Type"] == "REGISTER_DOMAIN"
@mock_aws
def test_get_nonexistent_operation_detail():
route53domains_client = boto3.client("route53domains", region_name="global")
with pytest.raises(ClientError) as exc:
route53domains_client.get_operation_detail(
OperationId="non-exiting-operation-id"
)
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
@mock_aws
def test_duplicate_requests(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
with pytest.raises(ClientError) as exc:
route53domains_client.register_domain(**domain_parameters)
err = exc.value.response["Error"]
assert err["Code"] == "DuplicateRequest"
@mock_aws
def test_domain_limit(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
params = domain_parameters.copy()
for i in range(20):
params["DomainName"] = f"domain-{i}.com"
route53domains_client.register_domain(**params)
params["DomainName"] = "domain-20.com"
with pytest.raises(ClientError) as exc:
route53domains_client.register_domain(**params)
err = exc.value.response["Error"]
assert err["Code"] == "DomainLimitExceeded"
@mock_aws
def test_get_domain_detail(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
res = route53domains_client.get_domain_detail(
DomainName=domain_parameters["DomainName"]
)
assert res["DomainName"] == domain_parameters["DomainName"]
@mock_aws
def test_get_invalid_domain_detail(domain_parameters):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
with pytest.raises(ClientError) as exc:
route53domains_client.get_domain_detail(DomainName="not-a-domain")
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
with pytest.raises(ClientError) as exc:
route53domains_client.get_domain_detail(DomainName="test.non-existing-tld")
err = exc.value.response["Error"]
assert err["Code"] == "UnsupportedTLD"
@mock_aws
def test_list_domains(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
res = route53domains_client.list_domains()
assert len(res["Domains"]) == 1
params = domain_parameters.copy()
params["DomainName"] = "new-domain.com"
route53domains_client.register_domain(**params)
res = route53domains_client.list_domains()
assert len(res["Domains"]) == 2
@mock_aws
@pytest.mark.parametrize(
"filters,expected_domains_len",
[
(
[
{
"Name": "DomainName",
"Operator": "BEGINS_WITH",
"Values": ["some-non-registered-domain.com"],
}
],
0, # expected_domains_len
),
(
[
{
"Name": "DomainName",
"Operator": "BEGINS_WITH",
"Values": ["domain.com"],
}
],
1, # expected_domains_len
),
(
[
{
"Name": "Expiry",
"Operator": "GE",
"Values": [
str(
datetime.fromisocalendar(
year=2012, week=20, day=3
).timestamp()
)
],
}
],
1, # expected_domains_len
),
(
[
{
"Name": "Expiry",
"Operator": "GE",
"Values": [
str(
datetime.fromisocalendar(
year=2050, week=20, day=3
).timestamp()
)
],
}
],
0, # expected_domains_len
),
],
)
def test_list_domains_filters(
domain_parameters: Dict, filters: List[Dict], expected_domains_len: int
):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
res = route53domains_client.list_domains(FilterConditions=filters)
assert len(res["Domains"]) == expected_domains_len
@mock_aws
def test_list_domains_sort_condition(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
params = domain_parameters.copy()
params["DomainName"] = "adomain.com"
route53domains_client.register_domain(**params)
params["DomainName"] = "bdomain.com"
route53domains_client.register_domain(**params)
sort = {"Name": "DomainName", "SortOrder": "DES"}
res = route53domains_client.list_domains(SortCondition=sort)
domains = res["Domains"]
assert domains[0]["DomainName"] == "bdomain.com"
assert domains[1]["DomainName"] == "adomain.com"
sort = {"Name": "Expiry", "SortOrder": "ASC"}
res = route53domains_client.list_domains(SortCondition=sort)
domains = res["Domains"]
assert domains[0]["DomainName"] == "adomain.com"
assert domains[1]["DomainName"] == "bdomain.com"
@mock_aws
def test_list_domains_invalid_filter(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
filters = [
{
"Name": "InvalidField",
"Operator": "InvalidOperator",
"Values": ["value-1", "value-2"], # multiple values isn't supported
}
]
with pytest.raises(ClientError) as exc:
route53domains_client.list_domains(FilterConditions=filters)
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
@mock_aws
def test_list_domains_invalid_sort_condition(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
sort = {
"Name": "InvalidField",
"SortOrder": "InvalidOrder",
}
with pytest.raises(ClientError) as exc:
route53domains_client.list_domains(SortCondition=sort)
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
@mock_aws
def test_list_domains_sort_condition_not_the_same_as_filter_condition(
domain_parameters: Dict,
):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
sort = {
"Name": "Expiry",
"SortOrder": "ASC",
}
filters = [
{
"Name": "DomainName",
"Operator": "BEGINS_WITH",
"Values": ["domain.com"],
}
]
with pytest.raises(ClientError) as exc:
route53domains_client.list_domains(FilterConditions=filters, SortCondition=sort)
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
@mock_aws
def test_delete_domain(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
domains = route53domains_client.list_domains()["Domains"]
assert len(domains) == 1
route53domains_client.delete_domain(DomainName=domain_parameters["DomainName"])
domains = route53domains_client.list_domains()["Domains"]
assert len(domains) == 0
operations = route53domains_client.list_operations(Type=["DELETE_DOMAIN"])[
"Operations"
]
assert len(operations) == 1
@mock_aws
def test_delete_invalid_domain(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
domains = route53domains_client.list_domains()["Domains"]
assert len(domains) == 0
with pytest.raises(ClientError) as exc:
route53domains_client.delete_domain(DomainName=domain_parameters["DomainName"])
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
@mock_aws
@pytest.mark.parametrize(
"nameservers",
[
[{"Name": "1-nameserver.net"}, {"Name": "2-nameserver.net"}],
[
{"Name": "3-nameserver.net", "GlueIps": ["1.1.1.2"]},
{"Name": "4-nameserver.net", "GlueIps": ["1.1.1.1"]},
],
],
)
def test_update_domain_nameservers(domain_parameters: Dict, nameservers: List[Dict]):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
operation_id = route53domains_client.update_domain_nameservers(
DomainName=domain_parameters["DomainName"], Nameservers=nameservers
)["OperationId"]
domain = route53domains_client.get_domain_detail(
DomainName=domain_parameters["DomainName"]
)
assert domain["Nameservers"] == nameservers
operation = route53domains_client.get_operation_detail(OperationId=operation_id)
assert operation["Type"] == "UPDATE_NAMESERVER"
assert operation["Status"] == "SUCCESSFUL"
@mock_aws
@pytest.mark.parametrize(
"nameservers",
[
[{"Name": "1-nameserver.net", "GlueIps": ["1.1.1.1", "1.1.1.2"]}],
[
{
"Name": "1-nameserver.net",
"GlueIps": [
"2001:0db8:85a3:0000:0000:8a2e:0370:7334",
"2001:0db8:85a3:0000:0000:8a2e:0370:7334",
],
}
],
[
{
"Name": "1-nameserver.net",
"GlueIps": [
"2001:0db8:85a3:0000:0000:8a2e:0370:7334",
"2001:0db8:85a3:0000:0000:8a2e:0370:7334",
"1.1.1.1",
],
}
],
[
{
"Name": "1-nameserver.net",
"GlueIps": [
"2001:0db8:85a3:0000:0000:8a2e:0370:7334",
"2001:0db8:85a3:0000:0000:8a2e:0370:7334",
"1.1.1.1",
"1.1.1.2",
],
}
],
[
{
"Name": "1-nameserver.net",
"GlueIps": [
"2001:0db8:85a3:0000:0000:8a2e:0370:7334",
"1.1.1.1",
"not-an-ip-address",
],
}
],
],
)
def test_update_domain_nameservers_with_multiple_glue_ips(
domain_parameters: Dict, nameservers: List[Dict]
):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
with pytest.raises(ClientError) as exc:
route53domains_client.update_domain_nameservers(
DomainName=domain_parameters["DomainName"], Nameservers=nameservers
)
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
@mock_aws
def test_update_domain_nameservers_requires_glue_ips(domain_parameters: Dict):
route53domains_client = boto3.client("route53domains", region_name="global")
route53domains_client.register_domain(**domain_parameters)
domain_name = domain_parameters["DomainName"]
nameservers = [{"Name": f"subdomain.{domain_name}"}]
with pytest.raises(ClientError) as exc:
route53domains_client.update_domain_nameservers(
DomainName=domain_parameters["DomainName"], Nameservers=nameservers
)
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"
@mock_aws
def test_update_domain_nameservers_for_nonexistent_domain():
route53domains_client = boto3.client("route53domains", region_name="global")
nameservers = [{"Name": "1-nameserver.net"}, {"Name": "2-nameserver.net"}]
with pytest.raises(ClientError) as exc:
route53domains_client.update_domain_nameservers(
DomainName="non-existent-domain.com", Nameservers=nameservers
)
err = exc.value.response["Error"]
assert err["Code"] == "InvalidInput"