Add support for IoT Domain Configuration (#4648)

IMPLEMENTATION_COVERAGE.md

@@ -2711,7 +2711,7 @@
 
 ## iot
 <details>
-<summary>27% implemented</summary>
+<summary>29% implemented</summary>
 
 - [ ] accept_certificate_transfer
 - [ ] add_thing_to_billing_group
@@ -2735,7 +2735,7 @@
 - [ ] create_certificate_from_csr
 - [ ] create_custom_metric
 - [ ] create_dimension
-- [ ] create_domain_configuration
+- [X] create_domain_configuration
 - [ ] create_dynamic_thing_group
 - [ ] create_fleet_metric
 - [X] create_job
@@ -2765,7 +2765,7 @@
 - [X] delete_certificate
 - [ ] delete_custom_metric
 - [ ] delete_dimension
-- [ ] delete_domain_configuration
+- [X] delete_domain_configuration
 - [ ] delete_dynamic_thing_group
 - [ ] delete_fleet_metric
 - [X] delete_job
@@ -2802,7 +2802,7 @@
 - [ ] describe_default_authorizer
 - [ ] describe_detect_mitigation_actions_task
 - [ ] describe_dimension
-- [ ] describe_domain_configuration
+- [X] describe_domain_configuration
 - [X] describe_endpoint
 - [ ] describe_event_configurations
 - [ ] describe_fleet_metric
@@ -2860,7 +2860,7 @@
 - [ ] list_detect_mitigation_actions_executions
 - [ ] list_detect_mitigation_actions_tasks
 - [ ] list_dimensions
-- [ ] list_domain_configurations
+- [X] list_domain_configurations
 - [ ] list_fleet_metrics
 - [ ] list_indices
 - [X] list_job_executions_for_job
@@ -2932,7 +2932,7 @@
 - [X] update_certificate
 - [ ] update_custom_metric
 - [ ] update_dimension
-- [ ] update_domain_configuration
+- [X] update_domain_configuration
 - [ ] update_dynamic_thing_group
 - [ ] update_event_configurations
 - [ ] update_fleet_metric

moto/iot/models.py

@@ -484,6 +484,62 @@ def to_dict(self):
         }
 
 
+class FakeDomainConfiguration(BaseModel):
+    def __init__(
+        self,
+        region_name,
+        domain_configuration_name,
+        domain_name,
+        server_certificate_arns,
+        domain_configuration_status,
+        service_type,
+        authorizer_config,
+        domain_type,
+    ):
+        if service_type and service_type not in ["DATA", "CREDENTIAL_PROVIDER", "JOBS"]:
+            raise InvalidRequestException(
+                "An error occurred (InvalidRequestException) when calling the DescribeDomainConfiguration "
+                "operation: Service type %s not recognized." % service_type
+            )
+        self.domain_configuration_name = domain_configuration_name
+        self.domain_configuration_arn = "arn:aws:iot:%s:1:domainconfiguration/%s/%s" % (
+            region_name,
+            domain_configuration_name,
+            random_string(5),
+        )
+        self.domain_name = domain_name
+        self.server_certificates = []
+        if server_certificate_arns:
+            for sc in server_certificate_arns:
+                self.server_certificates.append(
+                    {"serverCertificateArn": sc, "serverCertificateStatus": "VALID"}
+                )
+        self.domain_configuration_status = domain_configuration_status
+        self.service_type = service_type
+        self.authorizer_config = authorizer_config
+        self.domain_type = domain_type
+        self.last_status_change_date = time.time()
+
+    def to_description_dict(self):
+        return {
+            "domainConfigurationName": self.domain_configuration_name,
+            "domainConfigurationArn": self.domain_configuration_arn,
+            "domainName": self.domain_name,
+            "serverCertificates": self.server_certificates,
+            "authorizerConfig": self.authorizer_config,
+            "domainConfigurationStatus": self.domain_configuration_status,
+            "serviceType": self.service_type,
+            "domainType": self.domain_type,
+            "lastStatusChangeDate": self.last_status_change_date,
+        }
+
+    def to_dict(self):
+        return {
+            "domainConfigurationName": self.domain_configuration_name,
+            "domainConfigurationArn": self.domain_configuration_arn,
+        }
+
+
 class IoTBackend(BaseBackend):
     def __init__(self, region_name=None):
         super(IoTBackend, self).__init__()
@@ -499,6 +555,7 @@ def __init__(self, region_name=None):
         self.principal_things = OrderedDict()
         self.rules = OrderedDict()
         self.endpoint = None
+        self.domain_configurations = OrderedDict()
 
     def reset(self):
         region_name = self.region_name
@@ -1408,6 +1465,64 @@ def disable_topic_rule(self, rule_name):
             raise ResourceNotFoundException()
         self.rules[rule_name].rule_disabled = True
 
+    def create_domain_configuration(
+        self,
+        domain_configuration_name,
+        domain_name,
+        server_certificate_arns,
+        validation_certificate_arn,
+        authorizer_config,
+        service_type,
+    ):
+        if domain_configuration_name in self.domain_configurations:
+            raise ResourceAlreadyExistsException(
+                "Domain configuration with given name already exists."
+            )
+        self.domain_configurations[domain_configuration_name] = FakeDomainConfiguration(
+            self.region_name,
+            domain_configuration_name,
+            domain_name,
+            server_certificate_arns,
+            "ENABLED",
+            service_type,
+            authorizer_config,
+            "CUSTOMER_MANAGED",
+        )
+        return self.domain_configurations[domain_configuration_name]
+
+    def delete_domain_configuration(self, domain_configuration_name):
+        if domain_configuration_name not in self.domain_configurations:
+            raise ResourceNotFoundException("The specified resource does not exist.")
+        del self.domain_configurations[domain_configuration_name]
+
+    def describe_domain_configuration(self, domain_configuration_name):
+        if domain_configuration_name not in self.domain_configurations:
+            raise ResourceNotFoundException("The specified resource does not exist.")
+        return self.domain_configurations[domain_configuration_name]
+
+    def list_domain_configurations(self):
+        return [_.to_dict() for _ in self.domain_configurations.values()]
+
+    def update_domain_configuration(
+        self,
+        domain_configuration_name,
+        authorizer_config,
+        domain_configuration_status,
+        remove_authorizer_config,
+    ):
+        if domain_configuration_name not in self.domain_configurations:
+            raise ResourceNotFoundException("The specified resource does not exist.")
+        domain_configuration = self.domain_configurations[domain_configuration_name]
+        if authorizer_config is not None:
+            domain_configuration.authorizer_config = authorizer_config
+        if domain_configuration_status is not None:
+            domain_configuration.domain_configuration_status = (
+                domain_configuration_status
+            )
+        if remove_authorizer_config is not None and remove_authorizer_config is True:
+            domain_configuration.authorizer_config = None
+        return domain_configuration
+
 
 iot_backends = {}
 for region in Session().get_available_regions("iot"):

moto/iot/responses.py

@@ -696,3 +696,40 @@ def enable_topic_rule(self):
     def disable_topic_rule(self):
         self.iot_backend.disable_topic_rule(rule_name=self._get_param("ruleName"))
         return json.dumps(dict())
+
+    def create_domain_configuration(self):
+        domain_configuration = self.iot_backend.create_domain_configuration(
+            domain_configuration_name=self._get_param("domainConfigurationName"),
+            domain_name=self._get_param("domainName"),
+            server_certificate_arns=self._get_param("serverCertificateArns"),
+            validation_certificate_arn=self._get_param("validationCertificateArn"),
+            authorizer_config=self._get_param("authorizerConfig"),
+            service_type=self._get_param("serviceType"),
+        )
+        return json.dumps(domain_configuration.to_dict())
+
+    def delete_domain_configuration(self):
+        self.iot_backend.delete_domain_configuration(
+            domain_configuration_name=self._get_param("domainConfigurationName")
+        )
+        return json.dumps(dict())
+
+    def describe_domain_configuration(self):
+        domain_configuration = self.iot_backend.describe_domain_configuration(
+            domain_configuration_name=self._get_param("domainConfigurationName")
+        )
+        return json.dumps(domain_configuration.to_description_dict())
+
+    def list_domain_configurations(self):
+        return json.dumps(
+            dict(domainConfigurations=self.iot_backend.list_domain_configurations())
+        )
+
+    def update_domain_configuration(self):
+        domain_configuration = self.iot_backend.update_domain_configuration(
+            domain_configuration_name=self._get_param("domainConfigurationName"),
+            authorizer_config=self._get_param("authorizerConfig"),
+            domain_configuration_status=self._get_param("domainConfigurationStatus"),
+            remove_authorizer_config=self._get_bool_param("removeAuthorizerConfig"),
+        )
+        return json.dumps(domain_configuration.to_dict())