es & opensearch - add describe_elasticsearch_domains & describe_domai…

…ns support (#8386)

moto/es/models.py

@@ -142,5 +142,14 @@ def list_domain_names(self) -> List[Dict[str, str]]:
         """
         return [{"DomainName": domain.domain_name} for domain in self.domains.values()]
 
+    def describe_elasticsearch_domains(
+        self, domain_names: List[str]
+    ) -> List[Dict[str, Any]]:
+        queried_domains = []
+        for domain_name in domain_names:
+            if domain_name in self.domains:
+                queried_domains.append(self.domains[domain_name].to_json())
+        return queried_domains
+
 
 es_backends = BackendDict(ElasticsearchServiceBackend, "es")

moto/es/responses.py

@@ -13,7 +13,7 @@ class ElasticsearchServiceResponse(BaseResponse):
     """Handler for ElasticsearchService requests and responses."""
 
     def __init__(self) -> None:
-        super().__init__(service_name="elasticsearch")
+        super().__init__(service_name="es")
 
     @property
     def es_backend(self) -> ElasticsearchServiceBackend:
@@ -98,3 +98,10 @@ def describe_elasticsearch_domain(self) -> TYPE_RESPONSE:
     def list_domain_names(self) -> TYPE_RESPONSE:
         domain_names = self.es_backend.list_domain_names()
         return 200, {}, json.dumps({"DomainNames": domain_names})
+
+    def describe_elasticsearch_domains(self) -> TYPE_RESPONSE:
+        domain_names = self._get_param("DomainNames")
+        domain_list = self.es_backend.describe_elasticsearch_domains(
+            domain_names=domain_names,
+        )
+        return 200, {}, json.dumps({"DomainStatusList": domain_list})

moto/es/urls.py

@@ -11,11 +11,13 @@
     "{0}/2015-01-01/domain$": ElasticsearchServiceResponse.list_domains,
     "{0}/2015-01-01/es/domain$": ElasticsearchServiceResponse.domains,
     "{0}/2015-01-01/es/domain/(?P<domainname>[^/]+)": ElasticsearchServiceResponse.domain,
+    "{0}/2015-01-01/es/domain-info$": ElasticsearchServiceResponse.dispatch,
     "{0}/2021-01-01/domain$": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/opensearch/compatibleVersions": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/opensearch/domain": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/opensearch/domain/(?P<domainname>[^/]+)": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/opensearch/domain/(?P<domainname>[^/]+)/config": OpenSearchServiceResponse.dispatch,
+    "{0}/2021-01-01/opensearch/domain-info$": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/tags/?": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/tags-removal/": OpenSearchServiceResponse.dispatch,
 }

moto/opensearch/models.py

@@ -375,5 +375,12 @@ def list_domain_names(self, engine_type: str) -> List[Dict[str, str]]:
                 )
         return domains
 
+    def describe_domains(self, domain_names: List[str]) -> List[OpenSearchDomain]:
+        queried_domains = []
+        for domain_name in domain_names:
+            if domain_name in self.domains:
+                queried_domains.append(self.domains[domain_name])
+        return queried_domains
+
 
 opensearch_backends = BackendDict(OpenSearchServiceBackend, "opensearch")

moto/opensearch/responses.py

@@ -147,3 +147,11 @@ def list_domain_names(self) -> str:
             engine_type=engine_type,
         )
         return json.dumps(dict(DomainNames=domain_names))
+
+    def describe_domains(self) -> str:
+        domain_names = self._get_param("DomainNames")
+        domains = self.opensearch_backend.describe_domains(
+            domain_names=domain_names,
+        )
+        domain_list = [domain.to_dict() for domain in domains]
+        return json.dumps({"DomainStatusList": domain_list})

moto/opensearch/urls.py

@@ -8,4 +8,6 @@
 response = OpenSearchServiceResponse()
 
 
-url_paths = {"{0}/.*$": response.dispatch}
+url_paths = {
+    "{0}/.*$": response.dispatch,
+}

tests/test_es/test_es.py

@@ -215,3 +215,28 @@ def test_list_domain_names_with_multiple_domains():
     assert len(resp["DomainNames"]) == 4
     for name in domain_names:
         assert {"DomainName": name} in resp["DomainNames"]
+
+
+@mock_aws
+def test_describe_elasticsearch_domains():
+    client = boto3.client("es", region_name="us-east-1")
+    domain_names = [f"env{i}" for i in range(1, 5)]
+    for name in domain_names:
+        client.create_elasticsearch_domain(
+            DomainName=name,
+            ElasticsearchVersion="7.10",
+            AdvancedOptions={"option": "value"},
+            AdvancedSecurityOptions={"Enabled": False},
+        )
+    resp = client.describe_elasticsearch_domains(DomainNames=domain_names)
+
+    assert len(resp["DomainStatusList"]) == 4
+    for domain in resp["DomainStatusList"]:
+        assert domain["DomainName"] in domain_names
+        assert domain["ElasticsearchVersion"] == "7.10"
+        assert "AdvancedSecurityOptions" in domain.keys()
+        assert "AdvancedOptions" in domain.keys()
+
+    # Test for invalid domain name
+    resp = client.describe_elasticsearch_domains(DomainNames=["invalid"])
+    assert len(resp["DomainStatusList"]) == 0

tests/test_opensearch/test_opensearch.py

@@ -217,3 +217,23 @@ def test_list_unknown_domain_names_engine_type():
     err = exc.value.response["Error"]
     assert err["Code"] == "EngineTypeNotFoundException"
     assert err["Message"] == "Engine Type not found: testdn"
+
+
+@mock_aws
+def test_describe_domains():
+    client = boto3.client("opensearch", region_name="us-east-1")
+    domain_names = [f"env{i}" for i in range(1, 5)]
+    opensearch_engine_version = "OpenSearch_1.0"
+    for name in domain_names:
+        client.create_domain(DomainName=name, EngineVersion=opensearch_engine_version)
+    resp = client.describe_domains(DomainNames=domain_names)
+
+    assert len(resp["DomainStatusList"]) == 4
+    for domain in resp["DomainStatusList"]:
+        assert domain["DomainName"] in domain_names
+        assert "AdvancedSecurityOptions" in domain.keys()
+        assert "AdvancedOptions" in domain.keys()
+
+    # Test for invalid domain name
+    resp = client.describe_domains(DomainNames=["invalid"])
+    assert len(resp["DomainStatusList"]) == 0